def Reload(self):
    """Call `Reload` on every `EventAccumulator`."""
    logger.info('Beginning EventMultiplexer.Reload()')
    self._reload_called = True
    # Build a list so we're safe even if the list of accumulators is modified
    # even while we're reloading.
    with self._accumulators_mutex:
      items = list(self._accumulators.items())
    items_queue = queue.Queue()
    for item in items:
      items_queue.put(item)

    # Methods of built-in python containers are thread-safe so long as the GIL
    # for the thread exists, but we might as well be careful.
    names_to_delete = set()
    names_to_delete_mutex = threading.Lock()

    def Worker():
      """Keeps reloading accumulators til none are left."""
      while True:
        try:
          name, accumulator = items_queue.get(block=False)
        except queue.Empty:
          # No more runs to reload.
          break

        try:
          accumulator.Reload()
        except (OSError, IOError) as e:
          logger.error('Unable to reload accumulator %r: %s', name, e)
        except directory_watcher.DirectoryDeletedError:
          with names_to_delete_mutex:
            names_to_delete.add(name)
        finally:
          items_queue.task_done()

    if self._max_reload_threads > 1:
      num_threads = min(
          self._max_reload_threads, len(items))
      logger.info('Starting %d threads to reload runs', num_threads)
      for i in xrange(num_threads):
        thread = threading.Thread(target=Worker, name='Reloader %d' % i)
        thread.daemon = True
        thread.start()
      items_queue.join()
    else:
      logger.info(
          'Reloading runs serially (one after another) on the main '
          'thread.')
      Worker()

    with self._accumulators_mutex:
      for name in names_to_delete:
        logger.warn('Deleting accumulator %r', name)
        del self._accumulators[name]
    logger.info('Finished with EventMultiplexer.Reload()')
    return self
def Reload ( self ) : logger . info ( 'Beginning EventMultiplexer.Reload()' ) self . _reload_called = True with self . _accumulators_mutex : items = list ( self . _accumulators . items ( ) ) items_queue = queue . Queue ( ) for item in items : items_queue . put ( item ) names_to_delete = set ( ) names_to_delete_mutex = threading . Lock ( ) def Worker ( ) : while True : try : name , accumulator = items_queue . get ( block = False ) except queue . Empty : break try : accumulator . Reload ( ) except ( OSError , IOError ) as e : logger . error ( 'Unable to reload accumulator %r: %s' , name , e ) except directory_watcher . DirectoryDeletedError : with names_to_delete_mutex : names_to_delete . add ( name ) finally : items_queue . task_done ( ) if self . _max_reload_threads > 1 : num_threads = min ( self . _max_reload_threads , len ( items ) ) logger . info ( 'Starting %d threads to reload runs' , num_threads ) for i in xrange ( num_threads ) : thread = threading . Thread ( target = Worker , name = 'Reloader %d' % i ) thread . daemon = True thread . start ( ) items_queue . join ( ) else : logger . info ( 'Reloading runs serially (one after another) on the main ' 'thread.' ) Worker ( ) with self . _accumulators_mutex : for name in names_to_delete : logger . warn ( 'Deleting accumulator %r' , name ) del self . _accumulators [ name ] logger . info ( 'Finished with EventMultiplexer.Reload()' ) return self
Call `Reload` on every `EventAccumulator`.