original_docstring: Processes the source in a multiple process.
  
      Muliprocessing is used to start up separate processes.
  
      Args:
        options: the command line arguments (instance of argparse.Namespace).
      
docstring: Processes the source in a multiple process.
  Muliprocessing is used to start up separate processes.
docstring_tokens: Processes the source in a multiple process . Muliprocessing is used to start up separate processes .
code: def _ProcessSourceMultiProcessMode(self, options):
      """Processes the source in a multiple process.
  
      Muliprocessing is used to start up separate processes.
  
      Args:
        options: the command line arguments (instance of argparse.Namespace).
      """
      # TODO: replace by an option.
      start_collection_process = True
  
      self._number_of_worker_processes = getattr(options, 'workers', 0)
      if self._number_of_worker_processes < 1:
        # One worker for each "available" CPU (minus other processes).
        # The number three here is derived from the fact that the engine starts
        # up:
        #   + A collector process.
        #   + A storage process.
        # If we want to utilize all CPUs on the system we therefore need to start
        # up workers that amounts to the total number of CPUs - 3 (these two plus
        # the main process). Thus the number three.
        cpus = multiprocessing.cpu_count() - 3
  
        if cpus <= self.MINIMUM_WORKERS:
          cpus = self.MINIMUM_WORKERS
        elif cpus >= self.MAXIMUM_WORKERS:
          # Let's have a maximum amount of workers.
          cpus = self.MAXIMUM_WORKERS
  
        self._number_of_worker_processes = cpus
  
      logging.info(u'Starting extraction in multi process mode.')
  
      collection_queue = queue.MultiThreadedQueue()
      storage_queue = queue.MultiThreadedQueue()
      parse_error_queue = queue.MultiThreadedQueue()
      self._engine = engine.Engine(
          collection_queue, storage_queue, parse_error_queue)
  
      # TODO: add support to handle multiple partitions.
      self._engine.SetSource(
          self.GetSourcePathSpec(), resolver_context=self._resolver_context)
  
      logging.debug(u'Starting preprocessing.')
      pre_obj = self.PreprocessSource(options)
  
      output_module = getattr(options, 'output_module', None)
      if output_module:
        storage_writer = storage.BypassStorageWriter(
            storage_queue, self._storage_file_path,
            output_module_string=output_module, pre_obj=pre_obj)
      else:
        storage_writer = storage.StorageFileWriter(
            storage_queue, self._storage_file_path, self._buffer_size, pre_obj)
  
      logging.debug(u'Preprocessing done.')
  
      # TODO: make sure parsers option is not set by preprocessing.
      parser_filter_string = getattr(options, 'parsers', '')
  
      self._parser_names = []
      for _, parser_class in parsers_manager.ParsersManager.GetParsers(
          parser_filter_string=parser_filter_string):
        self._parser_names.append(parser_class.NAME)
  
      self._PreprocessSetCollectionInformation(options, pre_obj)
  
      if 'filestat' in self._parser_names:
        include_directory_stat = True
      else:
        include_directory_stat = False
  
      filter_file = getattr(options, 'file_filter', None)
      if filter_file:
        filter_find_specs = engine_utils.BuildFindSpecsFromFile(
            filter_file, pre_obj=pre_obj)
      else:
        filter_find_specs = None
  
      if start_collection_process:
        resolver_context = context.Context()
      else:
        resolver_context = self._resolver_context
  
      engine_proxy = None
      rpc_proxy_client = None
  
      if self._run_foreman:
        worker_foreman = foreman.Foreman(
            show_memory_usage=self._show_worker_memory_information)
  
        # Start a proxy server (only needed when a foreman is started).
        engine_proxy = rpc_proxy.StandardRpcProxyServer(os.getpid())
        try:
          engine_proxy.Open()
          engine_proxy.RegisterFunction(
              'signal_end_of_collection', worker_foreman.SignalEndOfProcessing)
  
          proxy_thread = threading.Thread(
              name='rpc_proxy', target=engine_proxy.StartProxy)
          proxy_thread.start()
  
          rpc_proxy_client = rpc_proxy.StandardRpcProxyClient(
              engine_proxy.listening_port)
        except errors.ProxyFailedToStart as exception:
          proxy_thread = None
          logging.error((
              u'Unable to setup a RPC server for the engine with error '
              u'{0:s}').format(exception))
      else:
        worker_foreman = None
  
      self._collector = self._engine.CreateCollector(
          include_directory_stat, vss_stores=self._vss_stores,
          filter_find_specs=filter_find_specs, resolver_context=resolver_context)
  
      if rpc_proxy_client:
        self._collector.SetProxy(rpc_proxy_client)
  
      self._DebugPrintCollector(options)
  
      logging.info(u'Starting storage process.')
      self._storage_process = multiprocessing.Process(
          name='StorageThread', target=storage_writer.WriteEventObjects)
      self._storage_process.start()
  
      if start_collection_process:
        logging.info(u'Starting collection process.')
        self._collection_process = multiprocessing.Process(
            name='Collection', target=self._collector.Collect)
        self._collection_process.start()
  
      logging.info(u'Starting worker processes to extract events.')
  
      for worker_nr in range(self._number_of_worker_processes):
        extraction_worker = self._CreateExtractionWorker(worker_nr, options)
  
        logging.debug(u'Starting worker: {0:d} process'.format(worker_nr))
        worker_name = u'Worker_{0:d}'.format(worker_nr)
        # TODO: Test to see if a process pool can be a better choice.
        self._worker_processes[worker_name] = multiprocessing.Process(
            name=worker_name, target=extraction_worker.Run,
            kwargs={'parser_filter_string': parser_filter_string})
  
        self._worker_processes[worker_name].start()
        pid = self._worker_processes[worker_name].pid
        if worker_foreman:
          worker_foreman.MonitorWorker(pid=pid, name=worker_name)
  
      logging.info(u'Collecting and processing files.')
      if self._collection_process:
        while self._collection_process.is_alive():
          self._collection_process.join(10)
          # Check the worker status regularly while collection is still ongoing.
          if worker_foreman:
            worker_foreman.CheckStatus()
            # TODO: We get a signal when collection is done, which might happen
            # before the collection thread joins. Look at the option of speeding
            # up the process of the collector stopping by potentially killing it.
      else:
        self._collector.Collect()
  
      logging.info(u'Collection is done, waiting for processing to complete.')
      if worker_foreman:
        worker_foreman.SignalEndOfProcessing()
  
      # Close the RPC server since the collection thread is done.
      if engine_proxy:
        # Close the proxy, free up resources so we can shut down the thread.
        engine_proxy.Close()
  
        if proxy_thread.isAlive():
          proxy_thread.join()
  
      # Run through the running workers, one by one.
      # This will go through a list of all active worker processes and check it's
      # status. If a worker has completed it will be removed from the list.
      # The process will not wait longer than five seconds for each worker to
      # complete, if longer time passes it will simply check it's status and
      # move on. That ensures that worker process is monitored and status is
      # updated.
      while self._worker_processes:
        for process_name, process_obj in sorted(self._worker_processes.items()):
          if worker_foreman:
            worker_label = worker_foreman.GetLabel(
                name=process_name, pid=process_obj.pid)
          else:
            worker_label = None
  
          if not worker_label:
            if process_obj.is_alive():
              logging.info((
                  u'Process {0:s} [{1:d}] is not monitored by the foreman. Most '
                  u'likely due to a worker having completed it\'s processing '
                  u'while waiting for another worker to complete.').format(
                      process_name, process_obj.pid))
              logging.info(
                  u'Waiting for worker {0:s} to complete.'.format(process_name))
              process_obj.join()
              logging.info(u'Worker: {0:s} [{1:d}] has completed.'.format(
                  process_name, process_obj.pid))
  
            del self._worker_processes[process_name]
            continue
  
          if process_obj.is_alive():
            # Check status of worker.
            worker_foreman.CheckStatus(label=worker_label)
            process_obj.join(5)
          # Note that we explicitly must test against exitcode 0 here since
          # process.exitcode will be None if there is no exitcode.
          elif process_obj.exitcode != 0:
            logging.warning((
                u'Worker process: {0:s} already exited with code: '
                u'{1:d}.').format(process_name, process_obj.exitcode))
            process_obj.terminate()
            worker_foreman.TerminateProcess(label=worker_label)
  
          else:
            # Process is no longer alive, no need to monitor.
            worker_foreman.StopMonitoringWorker(label=worker_label)
            # Remove it from our list of active workers.
            del self._worker_processes[process_name]
  
      logging.info(u'Processing is done, waiting for storage to complete.')
  
      self._engine.SignalEndOfInputStorageQueue()
      self._storage_process.join()
      logging.info(u'Storage is done.')
code_tokens: def _ProcessSourceMultiProcessMode ( self , options ) : start_collection_process = True self . _number_of_worker_processes = getattr ( options , 'workers' , 0 ) if self . _number_of_worker_processes < 1 : cpus = multiprocessing . cpu_count ( ) - 3 if cpus <= self . MINIMUM_WORKERS : cpus = self . MINIMUM_WORKERS elif cpus >= self . MAXIMUM_WORKERS : cpus = self . MAXIMUM_WORKERS self . _number_of_worker_processes = cpus logging . info ( u'Starting extraction in multi process mode.' ) collection_queue = queue . MultiThreadedQueue ( ) storage_queue = queue . MultiThreadedQueue ( ) parse_error_queue = queue . MultiThreadedQueue ( ) self . _engine = engine . Engine ( collection_queue , storage_queue , parse_error_queue ) self . _engine . SetSource ( self . GetSourcePathSpec ( ) , resolver_context = self . _resolver_context ) logging . debug ( u'Starting preprocessing.' ) pre_obj = self . PreprocessSource ( options ) output_module = getattr ( options , 'output_module' , None ) if output_module : storage_writer = storage . BypassStorageWriter ( storage_queue , self . _storage_file_path , output_module_string = output_module , pre_obj = pre_obj ) else : storage_writer = storage . StorageFileWriter ( storage_queue , self . _storage_file_path , self . _buffer_size , pre_obj ) logging . debug ( u'Preprocessing done.' ) parser_filter_string = getattr ( options , 'parsers' , '' ) self . _parser_names = [ ] for _ , parser_class in parsers_manager . ParsersManager . GetParsers ( parser_filter_string = parser_filter_string ) : self . _parser_names . append ( parser_class . NAME ) self . _PreprocessSetCollectionInformation ( options , pre_obj ) if 'filestat' in self . _parser_names : include_directory_stat = True else : include_directory_stat = False filter_file = getattr ( options , 'file_filter' , None ) if filter_file : filter_find_specs = engine_utils . BuildFindSpecsFromFile ( filter_file , pre_obj = pre_obj ) else : filter_find_specs = None if start_collection_process : resolver_context = context . Context ( ) else : resolver_context = self . _resolver_context engine_proxy = None rpc_proxy_client = None if self . _run_foreman : worker_foreman = foreman . Foreman ( show_memory_usage = self . _show_worker_memory_information ) engine_proxy = rpc_proxy . StandardRpcProxyServer ( os . getpid ( ) ) try : engine_proxy . Open ( ) engine_proxy . RegisterFunction ( 'signal_end_of_collection' , worker_foreman . SignalEndOfProcessing ) proxy_thread = threading . Thread ( name = 'rpc_proxy' , target = engine_proxy . StartProxy ) proxy_thread . start ( ) rpc_proxy_client = rpc_proxy . StandardRpcProxyClient ( engine_proxy . listening_port ) except errors . ProxyFailedToStart as exception : proxy_thread = None logging . error ( ( u'Unable to setup a RPC server for the engine with error ' u'{0:s}' ) . format ( exception ) ) else : worker_foreman = None self . _collector = self . _engine . CreateCollector ( include_directory_stat , vss_stores = self . _vss_stores , filter_find_specs = filter_find_specs , resolver_context = resolver_context ) if rpc_proxy_client : self . _collector . SetProxy ( rpc_proxy_client ) self . _DebugPrintCollector ( options ) logging . info ( u'Starting storage process.' ) self . _storage_process = multiprocessing . Process ( name = 'StorageThread' , target = storage_writer . WriteEventObjects ) self . _storage_process . start ( ) if start_collection_process : logging . info ( u'Starting collection process.' ) self . _collection_process = multiprocessing . Process ( name = 'Collection' , target = self . _collector . Collect ) self . _collection_process . start ( ) logging . info ( u'Starting worker processes to extract events.' ) for worker_nr in range ( self . _number_of_worker_processes ) : extraction_worker = self . _CreateExtractionWorker ( worker_nr , options ) logging . debug ( u'Starting worker: {0:d} process' . format ( worker_nr ) ) worker_name = u'Worker_{0:d}' . format ( worker_nr ) self . _worker_processes [ worker_name ] = multiprocessing . Process ( name = worker_name , target = extraction_worker . Run , kwargs = { 'parser_filter_string' : parser_filter_string } ) self . _worker_processes [ worker_name ] . start ( ) pid = self . _worker_processes [ worker_name ] . pid if worker_foreman : worker_foreman . MonitorWorker ( pid = pid , name = worker_name ) logging . info ( u'Collecting and processing files.' ) if self . _collection_process : while self . _collection_process . is_alive ( ) : self . _collection_process . join ( 10 ) if worker_foreman : worker_foreman . CheckStatus ( ) else : self . _collector . Collect ( ) logging . info ( u'Collection is done, waiting for processing to complete.' ) if worker_foreman : worker_foreman . SignalEndOfProcessing ( ) if engine_proxy : engine_proxy . Close ( ) if proxy_thread . isAlive ( ) : proxy_thread . join ( ) while self . _worker_processes : for process_name , process_obj in sorted ( self . _worker_processes . items ( ) ) : if worker_foreman : worker_label = worker_foreman . GetLabel ( name = process_name , pid = process_obj . pid ) else : worker_label = None if not worker_label : if process_obj . is_alive ( ) : logging . info ( ( u'Process {0:s} [{1:d}] is not monitored by the foreman. Most ' u'likely due to a worker having completed it\'s processing ' u'while waiting for another worker to complete.' ) . format ( process_name , process_obj . pid ) ) logging . info ( u'Waiting for worker {0:s} to complete.' . format ( process_name ) ) process_obj . join ( ) logging . info ( u'Worker: {0:s} [{1:d}] has completed.' . format ( process_name , process_obj . pid ) ) del self . _worker_processes [ process_name ] continue if process_obj . is_alive ( ) : worker_foreman . CheckStatus ( label = worker_label ) process_obj . join ( 5 ) elif process_obj . exitcode != 0 : logging . warning ( ( u'Worker process: {0:s} already exited with code: ' u'{1:d}.' ) . format ( process_name , process_obj . exitcode ) ) process_obj . terminate ( ) worker_foreman . TerminateProcess ( label = worker_label ) else : worker_foreman . StopMonitoringWorker ( label = worker_label ) del self . _worker_processes [ process_name ] logging . info ( u'Processing is done, waiting for storage to complete.' ) self . _engine . SignalEndOfInputStorageQueue ( ) self . _storage_process . join ( ) logging . info ( u'Storage is done.' )
short_docstring: Processes the source in a multiple process.
short_docstring_tokens: Processes the source in a multiple process .
comment: """Processes the source in a multiple process.
  
      Muliprocessing is used to start up separate processes.
  
      Args:
        options: the command line arguments (instance of argparse.Namespace).
      """$SEP$# TODO: replace by an option.$SEP$# One worker for each "available" CPU (minus other processes).$SEP$# The number three here is derived from the fact that the engine starts$SEP$# up:$SEP$#   + A collector process.$SEP$#   + A storage process.$SEP$# If we want to utilize all CPUs on the system we therefore need to start$SEP$# up workers that amounts to the total number of CPUs - 3 (these two plus$SEP$# the main process). Thus the number three.$SEP$# Let's have a maximum amount of workers.$SEP$# TODO: add support to handle multiple partitions.$SEP$# TODO: make sure parsers option is not set by preprocessing.$SEP$# Start a proxy server (only needed when a foreman is started).$SEP$# TODO: Test to see if a process pool can be a better choice.$SEP$# Check the worker status regularly while collection is still ongoing.$SEP$# TODO: We get a signal when collection is done, which might happen$SEP$# before the collection thread joins. Look at the option of speeding$SEP$# up the process of the collector stopping by potentially killing it.$SEP$# Close the RPC server since the collection thread is done.$SEP$# Close the proxy, free up resources so we can shut down the thread.$SEP$# Run through the running workers, one by one.$SEP$# This will go through a list of all active worker processes and check it's$SEP$# status. If a worker has completed it will be removed from the list.$SEP$# The process will not wait longer than five seconds for each worker to$SEP$# complete, if longer time passes it will simply check it's status and$SEP$# move on. That ensures that worker process is monitored and status is$SEP$# updated.$SEP$# Check status of worker.$SEP$# Note that we explicitly must test against exitcode 0 here since$SEP$# process.exitcode will be None if there is no exitcode.$SEP$# Process is no longer alive, no need to monitor.$SEP$# Remove it from our list of active workers.