def _StartWorkerProcess(self, process_name, storage_writer):
    """Creates, starts, monitors and registers a worker process.

    Args:
      process_name (str): process name.
      storage_writer (StorageWriter): storage writer for a session storage used
          to create task storage.

    Returns:
      MultiProcessWorkerProcess: extraction worker process or None if the
          process could not be started.
    """
    process_name = 'Worker_{0:02d}'.format(self._last_worker_number)
    logger.debug('Starting worker process {0:s}'.format(process_name))

    if self._use_zeromq:
      queue_name = '{0:s} task queue'.format(process_name)
      task_queue = zeromq_queue.ZeroMQRequestConnectQueue(
          delay_open=True, linger_seconds=0, name=queue_name,
          port=self._task_queue_port,
          timeout_seconds=self._TASK_QUEUE_TIMEOUT_SECONDS)
    else:
      task_queue = self._task_queue

    process = worker_process.WorkerProcess(
        task_queue, storage_writer, self._artifacts_filter_helper,
        self.knowledge_base, self._session_identifier,
        self._processing_configuration,
        enable_sigsegv_handler=self._enable_sigsegv_handler, name=process_name)

    # Remove all possible log handlers to prevent a child process from logging
    # to the main process log file and garbling the log. The log handlers are
    # recreated after the worker process has been started.
    for handler in logging.root.handlers:
      logging.root.removeHandler(handler)
      handler.close()

    process.start()

    loggers.ConfigureLogging(
        debug_output=self._debug_output, filename=self._log_filename,
        mode='a', quiet_mode=self._quiet_mode)

    try:
      self._StartMonitoringProcess(process)

    except (IOError, KeyError) as exception:
      pid = process.pid
      logger.error((
          'Unable to monitor replacement worker process: {0:s} '
          '(PID: {1:d}) with error: {2!s}').format(
              process_name, pid, exception))

      self._TerminateProcess(process)
      return None

    self._RegisterProcess(process)

    self._last_worker_number += 1

    return process
def _StartWorkerProcess ( self , process_name , storage_writer ) : process_name = 'Worker_{0:02d}' . format ( self . _last_worker_number ) logger . debug ( 'Starting worker process {0:s}' . format ( process_name ) ) if self . _use_zeromq : queue_name = '{0:s} task queue' . format ( process_name ) task_queue = zeromq_queue . ZeroMQRequestConnectQueue ( delay_open = True , linger_seconds = 0 , name = queue_name , port = self . _task_queue_port , timeout_seconds = self . _TASK_QUEUE_TIMEOUT_SECONDS ) else : task_queue = self . _task_queue process = worker_process . WorkerProcess ( task_queue , storage_writer , self . _artifacts_filter_helper , self . knowledge_base , self . _session_identifier , self . _processing_configuration , enable_sigsegv_handler = self . _enable_sigsegv_handler , name = process_name ) for handler in logging . root . handlers : logging . root . removeHandler ( handler ) handler . close ( ) process . start ( ) loggers . ConfigureLogging ( debug_output = self . _debug_output , filename = self . _log_filename , mode = 'a' , quiet_mode = self . _quiet_mode ) try : self . _StartMonitoringProcess ( process ) except ( IOError , KeyError ) as exception : pid = process . pid logger . error ( ( 'Unable to monitor replacement worker process: {0:s} ' '(PID: {1:d}) with error: {2!s}' ) . format ( process_name , pid , exception ) ) self . _TerminateProcess ( process ) return None self . _RegisterProcess ( process ) self . _last_worker_number += 1 return process
Creates, starts, monitors and registers a worker process.

    Args:
      process_name (str): process name.
      storage_writer (StorageWriter): storage writer for a session storage used
          to create task storage.

    Returns:
      MultiProcessWorkerProcess: extraction worker process or None if the
          process could not be started.