def Reload(self):
    """Load events from every detected run."""
    logger.info('Beginning DbImportMultiplexer.Reload()')
    # Defer event sink creation until needed; this ensures it will only exist in
    # the thread that calls Reload(), since DB connections must be thread-local.
    if not self._event_sink:
      self._event_sink = self._CreateEventSink()
    # Use collections.deque() for speed when we don't need blocking since it
    # also has thread-safe appends/pops.
    loader_queue = collections.deque(six.itervalues(self._run_loaders))
    loader_delete_queue = collections.deque()

    def batch_generator():
      while True:
        try:
          loader = loader_queue.popleft()
        except IndexError:
          return
        try:
          for batch in loader.load_batches():
            yield batch
        except directory_watcher.DirectoryDeletedError:
          loader_delete_queue.append(loader)
        except (OSError, IOError) as e:
          logger.error('Unable to load run %r: %s', loader.subdir, e)

    num_threads = min(self._max_reload_threads, len(self._run_loaders))
    if num_threads <= 1:
      logger.info('Importing runs serially on a single thread')
      for batch in batch_generator():
        self._event_sink.write_batch(batch)
    else:
      output_queue = queue.Queue()
      sentinel = object()
      def producer():
        try:
          for batch in batch_generator():
            output_queue.put(batch)
        finally:
          output_queue.put(sentinel)
      logger.info('Starting %d threads to import runs', num_threads)
      for i in xrange(num_threads):
        thread = threading.Thread(target=producer, name='Loader %d' % i)
        thread.daemon = True
        thread.start()
      num_live_threads = num_threads
      while num_live_threads > 0:
        output = output_queue.get()
        if output == sentinel:
          num_live_threads -= 1
          continue
        self._event_sink.write_batch(output)
    for loader in loader_delete_queue:
      logger.warn('Deleting loader %r', loader.subdir)
      del self._run_loaders[loader.subdir]
    logger.info('Finished with DbImportMultiplexer.Reload()')
def Reload ( self ) : logger . info ( 'Beginning DbImportMultiplexer.Reload()' ) if not self . _event_sink : self . _event_sink = self . _CreateEventSink ( ) loader_queue = collections . deque ( six . itervalues ( self . _run_loaders ) ) loader_delete_queue = collections . deque ( ) def batch_generator ( ) : while True : try : loader = loader_queue . popleft ( ) except IndexError : return try : for batch in loader . load_batches ( ) : yield batch except directory_watcher . DirectoryDeletedError : loader_delete_queue . append ( loader ) except ( OSError , IOError ) as e : logger . error ( 'Unable to load run %r: %s' , loader . subdir , e ) num_threads = min ( self . _max_reload_threads , len ( self . _run_loaders ) ) if num_threads <= 1 : logger . info ( 'Importing runs serially on a single thread' ) for batch in batch_generator ( ) : self . _event_sink . write_batch ( batch ) else : output_queue = queue . Queue ( ) sentinel = object ( ) def producer ( ) : try : for batch in batch_generator ( ) : output_queue . put ( batch ) finally : output_queue . put ( sentinel ) logger . info ( 'Starting %d threads to import runs' , num_threads ) for i in xrange ( num_threads ) : thread = threading . Thread ( target = producer , name = 'Loader %d' % i ) thread . daemon = True thread . start ( ) num_live_threads = num_threads while num_live_threads > 0 : output = output_queue . get ( ) if output == sentinel : num_live_threads -= 1 continue self . _event_sink . write_batch ( output ) for loader in loader_delete_queue : logger . warn ( 'Deleting loader %r' , loader . subdir ) del self . _run_loaders [ loader . subdir ] logger . info ( 'Finished with DbImportMultiplexer.Reload()' )
Load events from every detected run.