Source code for pySPACE.environments.backends.ll_backend

""" Execute operations on a cluster with the LoadLeveler scheduler """

import glob
import logging
import logging.handlers
import multiprocessing
import os
import select
import socket
import subprocess as sub
import sys
import threading
import time
from collections import defaultdict
from functools import partial

from enum import Enum

import pySPACE
from pySPACE.environments.backends.base import Backend
from pySPACE.missions.operations.base import Operation
from import ProgressBar, Percentage, ETA, Bar
from import inform

    # noinspection PyPep8Naming
    import cPickle as pickle
except ImportError:
    import pickle

[docs]class LoadLevelerBackend(Backend): """ Commits every process to LoadLeveler cluster, which resumes parallel execution Each process corresponds to one combination of input data set and parameter choice. The process objects are first pickled. The path to the pickled object together with a helper script is then submitted to LoadLeveler. There the object is unpickled, called and the backend is informed when the results are stored. Communication between the independent processes and the backend is done via TCP socket connection (see :class:`~pySPACE.environments.backends.ll_backend.LoadLevelerComHandler` for detailed information). :Author: Anett Seeland ( :Created: 2011/06/08 :LastChange: 2012/09/06 Add communication to SubflowHandler """ LL_COMMAND_FILE_TEMPLATE = """ # @ job_type = serial # @ notification = never # @ class = {job_class} # @ resources = ConsumableMemory({memory}) ConsumableCPUs({CPUs}) # @ requirements = {requirements} # @ executable = {executable} # @ arguments = {arguments} # @ output = %(op_result_dir)s/log/pySPACE_$(jobid).out # @ error = %(op_result_dir)s/log/pySPACE_$(jobid).err # @ queue"""
[docs] def __init__(self): super(LoadLevelerBackend, self).__init__() self.state = "idling" # create command file template for Loadleveler if "job_class" not in pySPACE.configuration or not pySPACE.configuration["job_class"]: pySPACE.configuration["job_class"] = "general" if "consumable_memory" not in pySPACE.configuration or not pySPACE.configuration["consumable_memory"]: pySPACE.configuration["consumable_memory"] = "3250mb" if "consumable_cpus" not in pySPACE.configuration or not pySPACE.configuration["consumable_cpus"]: pySPACE.configuration["consumable_cpus"] = 1 if "anode" not in pySPACE.configuration: pySPACE.configuration["anode"] = "" assert (pySPACE.configuration["job_class"] in ['critical', 'critical_forking', 'general', 'general_forking', 'longterm', 'longterm_forking', 'test']),\ "LL_Backend:: Job class not existing! Check your pySPACE config file!" self.template_file = LoadLevelerBackend.LL_COMMAND_FILE_TEMPLATE.format( executable=sys.executable, arguments=" ".join([os.path.join(pySPACE.configuration.root_dir, "environments", "backends", ""), "%(process_file_path)s", self.SERVER_IP, "%(server_port)d"]), job_class=pySPACE.configuration["job_class"], memory=pySPACE.configuration["consumable_memory"], CPUs=pySPACE.configuration["consumable_cpus"], requirements=pySPACE.configuration["anode"]) self._log("Using '%s' as template", logging.DEBUG) # queue for execution self.result_handlers = None self.progress_bar = None self.process_dir = "" self._log("Created LoadLeveler Backend.")
[docs] def stage_in(self, operation): """ Stage the given operation. :param operation: The operation to stage. :type operation: Operation """ super(LoadLevelerBackend, self).stage_in(operation) # set up queue self.result_handlers = multiprocessing.Queue(200) # Set up progress bar widgets = ['Operation progress: ', Percentage(), ' ', Bar(), ' ', ETA()] self.progress_bar = ProgressBar(widgets=widgets, maxval=self.current_operation.number_processes) self.progress_bar.start() self._log("Operation - staged") self.state = "staged"
[docs] def execute(self, timeout=1e6): """ Execute all processes specified in the currently staged operation """ assert (self.state == "staged") self._log("Operation - executing") self.state = "executing" # The handler that is used remotely for logging handler_class = logging.handlers.SocketHandler handler_args = {"host":, "port": self.port} # the communication properties to talk to LoadLevelerComHandler backend_com = (self.SERVER_IP, self.SERVER_PORT) self._log('--> Loadleveler Communication : \n\t\t host:%s, port:%s' % (self.SERVER_IP, self.SERVER_PORT)) # Prepare the directory where processes are stored before submitted # to LoadLeveler self.process_dir = os.sep.join([self.current_operation.result_directory, ".processes"]) if not os.path.exists(self.process_dir): os.mkdir(self.process_dir) process_counter = 0 # create and start server socket thread self.listener = LoadLevelerComHandler(self.sock, self.result_handlers, self.progress_bar, self.template_file, log_func=self._log, operation_dir=self.current_operation.result_directory) self.listener.start() # create a client socket to talk to server socket thread send_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) send_socket.connect((self.SERVER_IP, self.SERVER_PORT)) try: # Until not all Processes have been created, prepare all processes # from the queue for remote execution and execute them get_process = partial(self.current_operation.processes.get, timeout=timeout) for process in iter(get_process, False): process.prepare(pySPACE.configuration, handler_class, handler_args, backend_com) # since preparing the process might be quite faster than executing # it we need another queue where processes get out when they have # finished execution self.result_handlers.put(process) # pickle the process object proc_file_name = os.sep.join([self.process_dir, "process_%d.pickle" % process_counter]) with open(proc_file_name, "wb") as proc_file: pickle.dump(process, proc_file, pickle.HIGHEST_PROTOCOL) # fill out LoadLeveler template llfile = self.template_file % { "process_file_path": proc_file_name, "server_port": self.SERVER_PORT, "op_result_dir": self.current_operation.result_directory} llfilepath = os.path.join(self.current_operation.result_directory, "ll_call.cmd") with open(llfilepath, 'w') as f: f.write(llfile) # submit to LoadLeveler error_counter = 0 while True: outlog, errlog = sub.Popen(["llsubmit", llfilepath], stdout=sub.PIPE, stderr=sub.PIPE).communicate() if errlog == "": break elif error_counter < 100: self._log("Warning: Job submission to LoadLeveler failed" " with %s. Job will be resubmitted." % errlog, logging.WARNING) time.sleep(1) error_counter += 1 else: self._log("Warning: Job submission to LoadLeveler failed %d times" " with %s. skipping job" % (error_counter, errlog), logging.WARNING) break # parse job_id for monitoring loadl_id = outlog.split("\"")[1].split(".")[-1] # inform listener that we successfully submitted the job # noinspection PyTypeChecker send_socket = LoadLevelerComHandler.send_message(send_socket, self.SERVER_IP, self.SERVER_PORT, LoadLevelerComHandler.MESSAGES.SUBMITTED, process_counter, loadl_id) # update process_counter process_counter += 1 # send message 'creation finished' to listener # noinspection PyTypeChecker send_socket = LoadLevelerComHandler.send_message(send_socket, self.SERVER_IP, self.SERVER_PORT, LoadLevelerComHandler.MESSAGES.CREATION_FINISHED) finally: self.listener.creation_finished = True send_socket.shutdown(socket.SHUT_RDWR) send_socket.close()
[docs] def check_status(self): """ Return a description of the current state of the operations execution .. todo:: do we really need this method??? """ # Returns the current state of the operation return self.state
[docs] def retrieve(self, timeout=1e6): """ Wait for all results of the operation This call blocks until all processes are finished or the given timeout is reached. If the timeout is zero, the timeout is disabled. :param timeout: The time to wait until a job is considered as "finished" and will be stopped. :type timeout: int """ assert (self.state == "executing") self._log("All processes submitted. Waiting for finishing.") # since self.current_operation.number_processes is not reliable (maybe # to high) we wait until the listener thread is terminated self.listener.finished.wait(timeout=timeout) self._log("Worker processes have exited gracefully") self.current_operation.processes.close() # if process creation has another thread if self.current_operation.create_process is not None: self.current_operation.create_process.join(timeout=timeout) self.result_handlers.close() # join also listener thread self.listener.join(timeout=timeout) # Change the state to finished self._log("Operation - retrieved") self.state = "retrieved" return True
[docs] def consolidate(self): """ Consolidate the single processes' results into a consistent result of the whole operation """ assert (self.state == "retrieved") self.current_operation.consolidate() self._log("Operation - consolidated") # collect all log file def _merge_files(file_list, delete=True): result_str = "" for filename in file_list: tmp_str = "" try: if os.path.getsize(filename) != 0: tmp_str += filename.split(os.sep)[-1] + "\n" + len(filename.split(os.sep)[-1]) * "-" + "\n" with open(filename, 'r') as f: tmp_str += tmp_str += 80 * "-" + "\n" if delete: os.remove(filename) except (IOError, OSError), e: self._log("Problems with file %s: %s." % (filename, e), logging.WARNING) result_str += tmp_str return result_str outlist = glob.glob(self.current_operation.result_directory + "/log/pySPACE*.out") out = _merge_files(outlist) errlist = glob.glob(self.current_operation.result_directory + "/log/pySPACE*.err") err = _merge_files(errlist) with open(self.current_operation.result_directory + "/pySPACE.out", 'w') as merged_out: merged_out.write(out) with open(self.current_operation.result_directory + "/pySPACE.err", 'w') as merged_err: merged_err.write(err) try: outlist = glob.glob(self.current_operation.result_directory + "/sub_log/pySPACE*.out") out = _merge_files(outlist) errlist = glob.glob(self.current_operation.result_directory + "/sub_log/pySPACE*.err") err = _merge_files(errlist) with open(self.current_operation.result_directory + "/pySPACE_sub.out", 'w') as merged_out: merged_out.write(out) with open(self.current_operation.result_directory + "/pySPACE_sub.err", 'w') as merged_err: merged_err.write(err) except IOError: pass self._log("Process Logging - consolidated") self.state = "consolidated"
[docs] def cleanup(self): """ Remove the current operation and all potential results that have been stored in this object """ self.state = "idling" try: # Remove .process dir try: if os.path.isdir(self.process_dir): os.rmdir(self.process_dir) except OSError, e: self._log("Deleting process folder failed with error: %s" % e, level=logging.CRITICAL) try: os.rmdir(self.current_operation.result_directory + "/log") self._log("Operation - logging folder cleaned up") except OSError, e: self._log("Deleting log folder failed with error %s. " "Maybe no processes started. " "Please check your operation file constraints." % e, level=logging.CRITICAL) try: os.rmdir(self.current_operation.result_directory + "/sub_log") self._log("Operation - subflow log folder cleaned up") except OSError: pass finally: # Remove the file logger for this operation logging.getLogger('').removeHandler(self.file_handler) # close listener socket self.sock.shutdown(socket.SHUT_RDWR) self.sock.close() self._log("Operation - logging closed") self.current_operation = None self._log("Operation - cleaned up")
[docs]class LoadLevelerComHandler(threading.Thread): """ Server socket thread for accepting connections and reacting on it A helper class for :class:`LoadLevelerBackend<pySPACE.environments.backends.ll_backend.LoadLevelerBackend>`, which releases finished processes from the execution queue making new submits to LoadLeveler possible. In addition the Backends progress bar is updated. It is also possible to communicate with this thread via the :class:`~pySPACE.environments.chains.node_chain.SubflowHandler` or via terminal (port and ip can be checked in 'll_call.cmd'). The benefit of communication is to find out, which processes are running and to find out the details on the current state of the processing. **Parameters** :sock: The server socket which listens to messages :executing_queue: A queue where one element has to be removed if 'finished' is sent. :progress_bar: A progress bar object with is updated, if 'finished' is sent. :loadl_temp: A template for a cmd-file that is used to submit subflow-jobs. :Author: Anett Seeland ( :Created: 2011/06/08 :LastChange: 2012/09/06 added communication with SubflowHandler """ # to label message end when communicating via socket connection JOB_END_TOKEN = "!END!"
[docs] def send_message(cls, connection, host, port, message_type, *args): """ Sends a message with the given type and optional args via the given connection. If the connection is established, it is used. Otherwise a new connection is created and will be returned. :param connection: The connection to use for sending. :type connection: socket.socket :param host: The host to send the message to if the connection is not established :type host: basestring :param port: The port to use for establishing a new connection :type port: int :param message_type: The type of the message :param args: Optional arguments :return: The connection that has been used to send the message :rtype: socket.socket """ message = str(message_type.value) for arg in args: message += ";%s" % arg message += cls.JOB_END_TOKEN return inform(message, conn=connection, ip_port=(host, port))
[docs] def __init__(self, sock, executing_queue, progress_bar, loadl_temp, log_func, operation_dir=None): threading.Thread.__init__(self) self.sock = sock self.executing_queue = executing_queue self.progress_bar = progress_bar self.loadl_temp = loadl_temp # variables for monitoring self.process_loadl_mapping = defaultdict(list) self.num_running_processes = 0 self.subflow_ids_running = set() self.subflows_waiting = [] self.MAX_RUNNING = 400 self.num_finished_processes = 0 self.subflow_ids_finished = set() self.creation_finished = False self.finished = threading.Event() # initialize select concept (multiplexing of socket connections) self.sock.listen(socket.SOMAXCONN) # define potentially readers and writers self.readers = [self.sock] self.writers = [] # data structure to store all established connections and messages for # reading and writing: data[connection] = [message_read, message_to_write] = {} # a task queue for subflows to be started; starting is swapped to # another thread to be faster in handling all incoming requests self.subflow_msg = multiprocessing.Queue() self.batch_size = 1 self.operation_dir = operation_dir self._log = log_func
[docs] def run(self): """ Accept, read and write on connections until all processes are finished """ # start thread for handling long messages, i.e. executing subflows subflow_starter = SubflowStarter(self.subflow_msg, self.sock, self.loadl_temp, log_func=self._log, operation_dir=self.operation_dir) subflow_starter.start() while not (self.creation_finished and self.num_running_processes == 0): # multiplexing on potentially requests (in self.readers/writers) readable, writable, others =, self.writers, [], 1.0) if self.sock in readable: conn, _ = self.sock.accept() self.readers.append(conn)[conn] = ["", ""] readable.remove(self.sock) for reader in readable: try: tmp = reader.recv(4096) except socket.error, e: self._log('recv %s' % e, logging.WARNING) self.close_sock(reader) else: if tmp:[reader][0] += tmp # Complete messages are processed if self.JOB_END_TOKEN in[reader][0]: self.parse_message(reader, subflow_starter) # New data to send. Make sure client is in the # server's writer queue. if[reader][1] != "" and reader not in self.writers: self.writers.append(reader) else: self.close_sock(reader) for writer in writable: try: # send data; tmp is #chars sent (may not be all in write buffer) tmp = writer.send([writer][1]) except socket.error, e: self._log('send: %s' % e, logging.WARNING) self.close_sock(writer) else: # Removed sent characters from write buffer[writer][1] =[writer][1][tmp:] # If write buffer is empty, remove socket from potentially writers if not[writer][1]: self.writers.remove(writer) # send all_tasks_finished signal to thread self.subflow_msg.put((False, False, False)) # give thread some time to realize end of tasks time.sleep(1) self.subflow_msg.close() subflow_starter.join(timeout=1e6) # raise event so that backend knows we have finished self.finished.set()
[docs] def close_sock(self, conn): """ Close the given connection and remove it from lists of potentially readers/writers :param conn: The connection to close :type conn: socket.socket """ try: conn.shutdown(socket.SHUT_RDWR) except socket.error: pass conn.close() if conn in self.readers: self.readers.remove(conn) if conn in self.writers: self.writers.remove(conn) del[conn]
[docs] def parse_message(self, conn, subflow_starter): """ Parse incoming message and react :param conn: The socket to parse the message from :type conn: socket.socket :param subflow_starter: A callable to start possible subflows with :type subflow_starter: SubflowStarter The following string messages can be send: :name: Send name of the Backend back, e.g. 'loadl' or 'local'. :subflow_batchsize;*batchsize*: *batch_size* determines how many subflows are executed in one serial LoadLeveler job. :submitted;*process_nr*;*loadl_id*: Informs the listener that the pickled process `process_*process_nr*.pickle` has been submitted to LoadLeveler. In addition the LoadLeveler job id (without step id) is stored to be available for debugging purposes. :creation_finished: Informs the listener that no further process will be created anymore. Based on that the listener has only to process further messages until the number of running processes equals zero. :finished: Informs the listener that a process has been successfully executed. Hence, the number of running processes is decremented (the Backend is able to submit a new job) and the progress bar is updated. :finished;*flow_id*: Informs the listener that a subflow with unique identifier *flow_id* has stored its results. :is_ready;*nr_subflows*;*subflow_ids*: Asks the listener which of the *nr_subflows* subflows (identified by their subflow_id) have already finished executing. *subflow_ids* must be a string representation of a set. The listener sends the set of finished ids back. :execute_subflows;*path*;*nr_subflows*;*subflow_ids*;*runs*: Asks the listener to execute *nr_subflows* subflows. *path* is the absolute path where the subflows (named by their unique subflow id) are stored on disk, e.g. the *temp_dir* of a node. Hence, *subflow_ids* is a list of the filenames in *path* (without file extension) that have been stored before in pickle-format. *runs* is a list containing the run numbers the flow should be executed with: the *run_number* determines the random seed, e.g., for a splitter node. In addition it is assumed, that the data the subflows need for execution is also present in *path* (as 'subflow_data.pickle'). Since the reaction of this request may take quite some time, it is swapped to another :class:`Thread<pySPACE.environments.backends.ll_backend.SubflowStarter>`. :get;*attribute_name*: Getter methods for debugging purposes. Sends the value of *attribute_name* back. :set;*attribute_name*;*value*: Setter methods for debugging purposes. Equals to 'self.attribute_name = value'. Note that *value* must be evaluable. .. note:: It is important that you don't forget the end_flog at each message! """ end_ind =[conn][0].find(self.JOB_END_TOKEN) message =[conn][0][:end_ind] message_type = message.split(";")[0] message_args = message.split(";")[1:] # noinspection PyBroadException try: message_type = self.MESSAGES(int(message_type)) if message_type == self.MESSAGES.NAME:[conn][1] = 'loadl%s' % self.JOB_END_TOKEN elif message_type == self.MESSAGES.SUBFLOW_BATCHSIZE: batch_size = eval(message_args[-1]) subflow_starter.batch_size = batch_size self.batch_size = batch_size elif message_type == self.MESSAGES.SUBMITTED: key = eval(message_args[0]) self.process_loadl_mapping[key].append(message_args[1]) self.num_running_processes += 1 elif message_type == self.MESSAGES.FINISHED: # it is a Process that has finished if not message_args: self.num_finished_processes += 1 self.num_running_processes -= 1 self.executing_queue.get(timeout=1e100) self.progress_bar.update(self.num_finished_processes) # it is a subflow that has finished else: self.subflow_ids_running.remove(message_args[0]) self.subflow_ids_finished.add(message_args[0])[conn][1] = "OK%s" % self.JOB_END_TOKEN elif message_type == self.MESSAGES.CREATION_FINISHED: self.creation_finished = True elif message_type == self.MESSAGES.EXECUTE_SUBFLOWS: path, runs = message_args[0], message_args[-1] nr_subflows, subflow_ids = [eval(s) for s in message_args[1:-1]] assert (nr_subflows == len(subflow_ids)), "incorrect number of subflows" # check if we can submit new jobs self.control_subflow_submission(path, runs, subflow_ids)[conn][1] = "OK%s" % self.JOB_END_TOKEN elif message_type == self.MESSAGES.IS_READY: nr_requested, requested_subflows = [eval(s) for s in message_args] assert (nr_requested == len(requested_subflows)), "incorrect number" \ " of subflows" # check which subflows have already finished and tell to client finished = requested_subflows & self.subflow_ids_finished # .. todo: maybe reduced self.subflow_ids_finished since they are # unique and will never be requested again[conn][1] = str(finished) + self.JOB_END_TOKEN # to be able to communicate with the backend in case something went # wrong give the user the possibility to get and set attributes elif message_type == self.MESSAGES.GET: self._log("variable wanted", logging.INFO) name = message_args[0][conn][1] = str(getattr(self, name, None)) elif message_type == self.MESSAGES.SET: name, value = message_args[0], eval(message_args[1]) setattr(self, name, value) else: self._log("Got unknown message: %s" % message, logging.WARNING)[conn][0] =[conn][0][end_ind + len(self.JOB_END_TOKEN):] # try to submit waiting jobs if self.subflows_waiting != [] and ( self.MAX_RUNNING - (len(self.subflow_ids_running) / self.batch_size) > 0): path, runs, subflow_ids = self.subflows_waiting.pop() self.control_subflow_submission(path, runs, subflow_ids) except Exception, e: self._log("Exception while parsing message '%s' %s:" % (message, e), logging.ERROR)
[docs] def control_subflow_submission(self, path, runs, subflow_ids): """ Give SubflowStarter only a limited amount of tasks The MAX_RUNNING variable specifies how many loadleveler subflow jobs should maximal run in parallel. With respect to this number the SubflowStarter gets only tasks if there are available slots. """ free_slots = self.MAX_RUNNING - (len(self.subflow_ids_running) / \ self.batch_size) if free_slots < (len(subflow_ids) / self.batch_size): # cannot submit all jobs at once end_ind = free_slots * self.batch_size self.subflow_msg.put((path, runs, subflow_ids[:end_ind])) self.subflows_waiting.append((path, runs, subflow_ids[end_ind:])) self.subflow_ids_running.update(subflow_ids[:end_ind]) else: self.subflow_msg.put((path, runs, subflow_ids)) self.subflow_ids_running.update(subflow_ids)
[docs]class SubflowStarter(threading.Thread): """ Submit subflows to LoadLeveler A helper class for :class:`~pySPACE.environments.backends.ll_backend.LoadLevelerComHandler`, which submits subflows as independent jobs to LoadLeveler. The reason behind it is a better (more fair) scheduling of jobs when several users run the software on a cluster due to the fact that each single job has a short computation time. Since LoadLeveler does not know that Processes have nothing to do than wait when their subflows have to be executed, another job class is used for the subflow that preempt the Processes. The name of this class ends with '_child'. **Parameters** :task_queue: This queue is filled with jobs from incoming requests by the :class:`LoadLevelerComHandler<pySPACE.environments.backends.ll_backend.LoadLevelerComHandler>`. Each job is specified by a tuple: (*path*,*runs*,*subflow_ids*). :sock: The listener socket to be able to get information about it. Needed to fill out the LoadLeveler command file. :loadl_temp: LoadLeveler command file template that has been partly filled out by :class:`LoadLevelerBackend<pySPACE.environments.backends.ll_backend.LoadLevelerBackend>` but needs to be further adjusted. .. todo:: create parallel Loadleveler jobs to reduce request from processes; its better to execute all subflows of one process first than one subflow of every process .. todo:: actually the server socket is not needed here, only the server port """
[docs] def __init__(self, task_queue, sock, loadl_temp, log_func, operation_dir=None): threading.Thread.__init__(self) self.task_queue = task_queue self.sock = sock self.loadl_temp = loadl_temp self.first_subflow = True self.batch_size = 1 self.operation_dir = operation_dir self._log = log_func
[docs] def submit(self, path, runs, subflow_ids): """ Submit one subflow job to LoadLeveler """ # TODO: Add path, runs and subflow_ids to docstring if self.operation_dir is None: operation_dir = path else: operation_dir = self.operation_dir llfile = self.loadl_temp % { "process_file_path": path, "server_port": self.sock.getsockname()[1], "op_result_dir": operation_dir, "runs": '"' + str(runs) + '"', "subflow_ids": str(subflow_ids)} llfilepath = os.path.join(path, "ll_call.cmd") with open(llfilepath, 'w') as f: f.write(llfile) # submit to LoadLeveler error_counter = 0 while True: outlog, errlog = sub.Popen(["llsubmit", llfilepath], stdout=sub.PIPE, stderr=sub.PIPE).communicate() if errlog == "": break elif error_counter < 100: self._log("Warning: Job submission to LoadLeveler failed" " with %s. Job will be resubmitted." % errlog, logging.WARNING) time.sleep(1) error_counter += 1 else: self._log("Warning: Job submission to LoadLeveler failed %d times" " with %s. skipping job" % (error_counter, errlog), logging.WARNING) break
[docs] def run(self): """ Portion and execute subflow tasks until *subflow_ids* equals 'False' """ rest_ids = [] path, runs, subflow_ids = self.task_queue.get(timeout=1e10) while subflow_ids: if self.first_subflow: # need this modification only once # modify template for subflows: class, ll_runner, params template = self.loadl_temp.split("\n") self.loadl_temp = "" for line in template: if line.startswith('# @ class'): assert ("forking" in line), "When using subflow " \ "paralellization a job class with forking " \ "functionality has to be used!" line = line.replace("forking", "child") elif line.startswith('# @ arguments'): line = line.replace('', ' ') line += ' %(runs)s "%(subflow_ids)s"' elif line.startswith('# @ output') or line.startswith('# @ error'): line = line.replace("/log/pySPACE_", "/sub_log/pySPACE_") self.loadl_temp += line + " \n" self.first_subflow = False # check whether nr of subflows is dividable by batch_size if len(subflow_ids) % self.batch_size: rest = len(subflow_ids) % self.batch_size rest_ids = subflow_ids[-1 * rest:] subflow_ids = subflow_ids[:len(subflow_ids) - rest] # iterate over loadlevler jobs for ind in range(0, len(subflow_ids), self.batch_size): self.submit(path, runs, subflow_ids[ind:ind + self.batch_size]) if rest_ids: # need to submit one more job self.submit(path, runs, rest_ids) path, runs, subflow_ids = self.task_queue.get(timeout=1e10)