def RunOnce(self):
    """Processes one set of messages from Task Scheduler.

    The worker processes new jobs from the task master. For each job
    we retrieve the session from the Task Scheduler.

    Returns:
        Total number of messages processed by this call.
    """
    start_time = time.time()
    processed = 0

    queue_manager = queue_manager_lib.QueueManager(token=self.token)
    for queue in self.queues:
      # Freezeing the timestamp used by queue manager to query/delete
      # notifications to avoid possible race conditions.
      queue_manager.FreezeTimestamp()

      fetch_messages_start = time.time()
      notifications = queue_manager.GetNotifications(queue)
      stats_collector_instance.Get().RecordEvent(
          "worker_time_to_retrieve_notifications",
          time.time() - fetch_messages_start)

      stuck_flows = []
      for n in notifications:
        if n.in_progress:
          stuck_flows.append(n)

      # Process stuck flows first
      if stuck_flows:
        self.ProcessStuckFlows(stuck_flows, queue_manager)

      notifications_available = []
      for notification in notifications:
        # Filter out session ids we already tried to lock but failed.
        if notification.session_id not in self.queued_flows:
          notifications_available.append(notification)

      try:
        # If we spent too much time processing what we have so far, the
        # active_sessions list might not be current. We therefore break here
        # so we can re-fetch a more up to date version of the list, and try
        # again later. The risk with running with an old active_sessions list
        # is that another worker could have already processed this message,
        # and when we try to process it, there is nothing to do - costing us a
        # lot of processing time. This is a tradeoff between checking the data
        # store for current information and processing out of date
        # information.
        processed += self.ProcessMessages(
            notifications_available, queue_manager,
            self.RUN_ONCE_MAX_SECONDS - (time.time() - start_time))

      # We need to keep going no matter what.
      except Exception as e:  # pylint: disable=broad-except
        logging.error("Error processing message %s. %s.", e,
                      traceback.format_exc())
        stats_collector_instance.Get().IncrementCounter("grr_worker_exceptions")
        if flags.FLAGS.pdb_post_mortem:
          pdb.post_mortem()

      queue_manager.UnfreezeTimestamp()
      # If we have spent too much time, stop.
      if (time.time() - start_time) > self.RUN_ONCE_MAX_SECONDS:
        return processed
    return processed
def RunOnce ( self ) : start_time = time . time ( ) processed = 0 queue_manager = queue_manager_lib . QueueManager ( token = self . token ) for queue in self . queues : queue_manager . FreezeTimestamp ( ) fetch_messages_start = time . time ( ) notifications = queue_manager . GetNotifications ( queue ) stats_collector_instance . Get ( ) . RecordEvent ( "worker_time_to_retrieve_notifications" , time . time ( ) - fetch_messages_start ) stuck_flows = [ ] for n in notifications : if n . in_progress : stuck_flows . append ( n ) if stuck_flows : self . ProcessStuckFlows ( stuck_flows , queue_manager ) notifications_available = [ ] for notification in notifications : if notification . session_id not in self . queued_flows : notifications_available . append ( notification ) try : processed += self . ProcessMessages ( notifications_available , queue_manager , self . RUN_ONCE_MAX_SECONDS - ( time . time ( ) - start_time ) ) except Exception as e : logging . error ( "Error processing message %s. %s." , e , traceback . format_exc ( ) ) stats_collector_instance . Get ( ) . IncrementCounter ( "grr_worker_exceptions" ) if flags . FLAGS . pdb_post_mortem : pdb . post_mortem ( ) queue_manager . UnfreezeTimestamp ( ) if ( time . time ( ) - start_time ) > self . RUN_ONCE_MAX_SECONDS : return processed return processed
Processes one set of messages from Task Scheduler.

    The worker processes new jobs from the task master. For each job
    we retrieve the session from the Task Scheduler.

    Returns:
        Total number of messages processed by this call.