Merge pull request #85 from mgermain/worker

Enforce worker use and added a default to the worker pool size

Author: MarcCote

scripts/smart_dispatch.py

@@ -53,47 +53,43 @@ def main():
     else:
         raise ValueError("Unknown subcommand!")
 
-    # Pool of workers
-    if args.pool is not None:
-        command_manager = CommandManager(os.path.join(path_job_commands, "commands.txt"))
+    command_manager = CommandManager(os.path.join(path_job_commands, "commands.txt"))
 
-        # If resume mode, reset running jobs
-        if args.mode == "launch":
-            command_manager.set_commands_to_run(commands)
-        else:
-            # Verifying if there is are failed commands
-            failed_commands = command_manager.get_failed_commands()
-            if len(failed_commands) > 0:
-                FAILED_COMMAND_MESSAGE = dedent("""\
-                {nb_failed} command(s) are in a failed state. They won't be resumed.
-                Failed commands:
-                {failed_commands}
-                The actual errors can be found in the log folder under:
-                {failed_commands_err_file}""")
-                utils.print_boxed(FAILED_COMMAND_MESSAGE.format(
-                    nb_failed=len(failed_commands),
-                    failed_commands=''.join(failed_commands),
-                    failed_commands_err_file='\n'.join([utils.generate_uid_from_string(c[:-1])+'.err' for c in failed_commands])
-                ))
-
-                if not utils.yes_no_prompt("Do you want to continue?", 'n'):
-                    exit()
-
-            command_manager.reset_running_commands()
-            nb_commands = command_manager.get_nb_commands_to_run()
-
-        worker_command = 'smart_worker.py "{0}" "{1}"'.format(command_manager._commands_filename, path_job_logs)
-        # Replace commands with `args.pool` workers
-        commands = [worker_command] * args.pool
-
-    # Add redirect for output and error logs
-    for i, command in enumerate(commands):
-        # Change directory before executing command
-        commands[i] = 'cd "{cwd}"; '.format(cwd=os.getcwd()) + commands[i]
-        # Log command's output and command's error
-        log_filename = os.path.join(path_job_logs, smartdispatch.generate_name_from_command(command, max_length_arg=30))
-        commands[i] += ' 1>> "{output_log}"'.format(output_log=log_filename + ".o")
-        commands[i] += ' 2>> "{error_log}"'.format(error_log=log_filename + ".e")
+    # If resume mode, reset running jobs
+    if args.mode == "launch":
+        command_manager.set_commands_to_run(commands)
+    elif args.mode == "resume":
+        # Verifying if there are failed commands
+        failed_commands = command_manager.get_failed_commands()
+        if len(failed_commands) > 0:
+            FAILED_COMMAND_MESSAGE = dedent("""\
+            {nb_failed} command(s) are in a failed state. They won't be resumed.
+            Failed commands:
+            {failed_commands}
+            The actual errors can be found in the log folder under:
+            {failed_commands_err_file}""")
+            utils.print_boxed(FAILED_COMMAND_MESSAGE.format(
+                nb_failed=len(failed_commands),
+                failed_commands=''.join(failed_commands),
+                failed_commands_err_file='\n'.join([utils.generate_uid_from_string(c[:-1]) + '.err' for c in failed_commands])
+            ))
+
+            if not utils.yes_no_prompt("Do you want to continue?", 'n'):
+                exit()
+
+        command_manager.reset_running_commands()
+        nb_commands = command_manager.get_nb_commands_to_run()
+
+    # If no pool size is specified the number of commands is taken
+    if args.pool is None:
+        args.pool = command_manager.get_nb_commands_to_run()
+
+    # Generating all the worker commands
+    COMMAND_STRING = 'cd "{cwd}"; smart_worker.py "{commands_file}" "{log_folder}" '\
+                     '1>> "{log_folder}/worker/$PBS_JOBID\"\"_worker_{{ID}}.o" '\
+                     '2>> "{log_folder}/worker/$PBS_JOBID\"\"_worker_{{ID}}.e" '
+    COMMAND_STRING = COMMAND_STRING.format(cwd=os.getcwd(), commands_file=command_manager._commands_filename, log_folder=path_job_logs)
+    commands = [COMMAND_STRING.format(ID=i) for i in range(args.pool)]
 
     # TODO: use args.memPerNode instead of args.memPerNode
     queue = Queue(args.queueName, CLUSTER_NAME, args.walltime, args.coresPerNode, args.gpusPerNode, np.inf, args.modules)
@@ -129,17 +125,17 @@ def parse_arguments():
     parser.add_argument('-L', '--launcher', choices=['qsub', 'msub'], required=False, help='Which launcher to use. Default: qsub')
     parser.add_argument('-C', '--coresPerNode', type=int, required=False, help='How many cores there are per node.')
     parser.add_argument('-G', '--gpusPerNode', type=int, required=False, help='How many gpus there are per node.')
-    #parser.add_argument('-M', '--memPerNode', type=int, required=False, help='How much memory there are per node (in Gb).')
+    # parser.add_argument('-M', '--memPerNode', type=int, required=False, help='How much memory there are per node (in Gb).')
 
     parser.add_argument('-c', '--coresPerCommand', type=int, required=False, help='How many cores a command needs.', default=1)
     parser.add_argument('-g', '--gpusPerCommand', type=int, required=False, help='How many gpus a command needs.', default=1)
-    #parser.add_argument('-m', '--memPerCommand', type=float, required=False, help='How much memory a command needs (in Gb).')
+    # parser.add_argument('-m', '--memPerCommand', type=float, required=False, help='How much memory a command needs (in Gb).')
     parser.add_argument('-f', '--commandsFile', type=file, required=False, help='File containing commands to launch. Each command must be on a seperate line. (Replaces commandAndOptions)')
 
     parser.add_argument('-l', '--modules', type=str, required=False, help='List of additional modules to load.', nargs='+')
     parser.add_argument('-x', '--doNotLaunch', action='store_true', help='Creates the QSUB files without launching them.')
 
-    parser.add_argument('-p', '--pool', type=int, help="Number of workers that will be consuming commands.")
+    parser.add_argument('-p', '--pool', type=int, help="Number of workers that will be consuming commands. Default: Nb commands")
     subparsers = parser.add_subparsers(dest="mode")
 
     launch_parser = subparsers.add_parser('launch', help="Launch jobs.")
@@ -180,6 +176,7 @@ def get_job_folders(jobname):
 
     if not os.path.exists(path_job_logs):
         os.makedirs(path_job_logs)
+        os.makedirs(os.path.join(path_job_logs, "worker"))
 
     return path_job, path_job_logs, path_job_commands
 
@@ -193,6 +190,7 @@ def create_job_folders(jobname):
 
     if not os.path.exists(path_job_logs):
         os.makedirs(path_job_logs)
+        os.makedirs(os.path.join(path_job_logs, "worker"))
 
     return path_job, path_job_logs, path_job_commands
 

smartdispatch/smartdispatch.py

@@ -3,7 +3,7 @@
 import os
 import re
 import itertools
-from datetime import datetime
+import time as t
 
 import smartdispatch
 from smartdispatch import utils
@@ -35,53 +35,9 @@ def generate_name_from_command(command, max_length_arg=None, max_length=None):
     if max_length_arg is not None:
         max_length_arg = min(-max_length_arg, max_length_arg)
 
-    if max_length is not None:
-        max_length = min(-max_length, max_length)
-
-    name = '_'.join([utils.slugify(argvalue)[max_length_arg:] for argvalue in command.split()])
-    return name[max_length:]
-
-
-def generate_name_from_arguments(arguments, max_length_arg=None, max_length=None, prefix=datetime.now().strftime('%Y-%m-%d_%H-%M-%S_')):
-    ''' Generates name from given unfolded arguments.
-
-    Generate a name by concatenating the first and last values of every
-    unfolded arguments and by trimming lengthty (as defined by max_length_arg)
-    arguments.
-
-    Parameters
-    ----------
-    arguments : list of list of str
-        list of unfolded arguments
-    max_length_arg : int
-        arguments longer than this will be trimmed keeping last characters (Default: inf)
-    max_length : int
-        trim name if longer than this keeping last characters (Default: inf)
-    prefix : str
-        text to preprend to the name (Default: current datetime)
-
-    Returns
-    -------
-    name : str
-        slugified name
-    '''
-    if max_length_arg is not None:
-        max_length_arg = min(-max_length_arg, max_length_arg)
-
-    if max_length is not None:
-        max_length = min(-max_length, max_length)
-
-    name = []
-    for argvalues in arguments:
-        argvalues = map(utils.slugify, argvalues)
-        name.append(argvalues[0][max_length_arg:])
-        if len(argvalues) > 1:
-            name[-1] += '-' + argvalues[-1][max_length_arg:]
-
-    name = "_".join(name)
-
-    name = prefix + name[max_length:]
-    return name
+    name = t.strftime("%Y-%m-%d_%H-%M-%S_")
+    name += '_'.join([utils.slugify(argvalue)[max_length_arg:] for argvalue in command.split()])
+    return name[:max_length]
 
 
 def get_commands_from_file(fileobj):

smartdispatch/tests/test_command_manager.py

@@ -11,6 +11,7 @@ class CommandFilesTests(unittest.TestCase):
 
     def setUp(self):
         self._base_dir = tmp.mkdtemp()
+        self.nb_commands = 3
         self.command1 = "1\n"
         self.command2 = "2\n"
         self.command3 = "3\n"
@@ -75,7 +76,7 @@ def test_get_command_to_run(self):
         assert_true(not os.path.isfile(self.command_manager._finished_commands_filename))
 
     def test_get_nb_commands_to_run(self):
-        assert_equal(self.command_manager.get_nb_commands_to_run(), 3)
+        assert_equal(self.command_manager.get_nb_commands_to_run(), self.nb_commands)
 
     def test_set_running_command_as_finished(self):
         # SetUp

smartdispatch/tests/test_smartdispatch.py

@@ -3,66 +3,33 @@
 
 from nose.tools import assert_true, assert_equal
 from numpy.testing import assert_array_equal
-from datetime import datetime
 from smartdispatch import utils
 
 
 def test_generate_name_from_command():
+    date_length = 20
+
     command = "command arg1 arg2"
     expected = "_".join(command.split())
-    assert_equal(smartdispatch.generate_name_from_command(command), expected)
+    assert_equal(smartdispatch.generate_name_from_command(command)[date_length:], expected)
 
     max_length_arg = 7
     long_arg = "veryverylongarg1"
     command = "command " + long_arg + " arg2"
     expected = command.split()
     expected[1] = long_arg[-max_length_arg:]
     expected = "_".join(expected)
-    assert_equal(smartdispatch.generate_name_from_command(command, max_length_arg), expected)
+    assert_equal(smartdispatch.generate_name_from_command(command, max_length_arg)[date_length:], expected)
 
     max_length = 23
-    long_arg = "veryverylongarg1"
     command = "command veryverylongarg1 veryverylongarg1 veryverylongarg1 veryverylongarg1"
-    expected = command.split()
-    expected = "_".join(expected)[-max_length:]
-    assert_equal(smartdispatch.generate_name_from_command(command, max_length=max_length), expected)
+    expected = command[:max_length].replace(" ", "_")
+    assert_equal(smartdispatch.generate_name_from_command(command, max_length=max_length + date_length)[date_length:], expected)
 
     # Test path arguments in command
     command = "command path/number/one path/number/two"
     expected = "command_pathnumberone_pathnumbertwo"
-    assert_equal(smartdispatch.generate_name_from_command(command), expected)
-
-
-def test_generate_name_from_arguments():
-    prefix = "prefix_"
-
-    arguments = [["my_command"], ["args1a", "args1b", "args1c"], ["args2a", "args2b"]]
-    expected = prefix + "my_command_args1a-args1c_args2a-args2b"
-    assert_equal(smartdispatch.generate_name_from_arguments(arguments, prefix=prefix), expected)
-
-    max_length_arg = 7
-    arguments = [["command"], ["verylongargs1a", "verylongargs1b", "verylongargs1c"], ["args2a", "args2b"]]
-    expected = prefix + "command_" + arguments[1][0][-max_length_arg:] + "-" + arguments[1][-1][-max_length_arg:] + "_args2a-args2b"
-    assert_equal(smartdispatch.generate_name_from_arguments(arguments, max_length_arg, prefix=prefix), expected)
-
-    max_length = 23
-    arguments = [["command"], ["verylongargs1a", "verylongargs1b", "verylongargs1c"], ["args2a", "args2b"]]
-    expected = "command_" + arguments[1][0] + "-" + arguments[1][-1] + "_args2a-args2b"
-    expected = prefix + expected[-max_length:]
-    assert_equal(smartdispatch.generate_name_from_arguments(arguments, max_length=max_length, prefix=prefix), expected)
-
-    # Test path arguments in command
-    arguments = [["command"], ["path/argument/1", "path/argument/2", "path/argument/3"]]
-    expected = prefix + "command_pathargument1-pathargument3"
-    assert_equal(smartdispatch.generate_name_from_arguments(arguments, prefix=prefix), expected)
-
-    # Make sure default prefix does not raise exception
-    arguments = [["command"]]
-    results = smartdispatch.generate_name_from_arguments(arguments)
-    expect_datetime = datetime.now()
-    assert_equal(results.split("_")[-1], arguments[0][0])
-    result_datetime = datetime.strptime("_".join(results.split("_")[:-1]), '%Y-%m-%d_%H-%M-%S')
-    assert_true(result_datetime <= expect_datetime)
+    assert_equal(smartdispatch.generate_name_from_command(command)[date_length:], expected)
 
 
 def test_get_commands_from_file():
@@ -158,7 +125,7 @@ def test_replace_uid_tag():
 
     commands = ["a command with a {UID} tag"] * 10
     uid = utils.generate_uid_from_string(commands[0])
-    assert_array_equal(smartdispatch.replace_uid_tag(commands), [commands[0].replace("{UID}", uid)]*len(commands))
+    assert_array_equal(smartdispatch.replace_uid_tag(commands), [commands[0].replace("{UID}", uid)] * len(commands))
 
 
 def test_get_available_queues():

tests/test_smart_worker.py

@@ -1,7 +1,6 @@
 import os
 import unittest
 import tempfile
-import fcntl
 import time
 import shutil