src/net/capture_engine.cpp
/**
 * Process packets found in the packet queue.
 * This method pulls packets off of the packets queue, parses them into memcache
 * commands, and enqueues them via CaptureEngine::enqueue(MemcacheCommand) if
 * the parsed command is a response.
 *
 * This method is run in a thread and continues running until isShutdown()
 * returns true.
 *
 * @TODO create several worker threads to process packets in parallel
 * @protected
 */
Process packets found in the packet queue.
This method pulls packets off of the packets queue, parses them into memcache
commands, and enqueues them via CaptureEngine::enqueue(MemcacheCommand) if
the parsed command is a response.

This method is run in a thread and continues running until isShutdown()
returns true.

@TODO create several worker threads to process packets in parallel
@protected
Process packets found in the packet queue . This method pulls packets off of the packets queue parses them into memcache commands and enqueues them via CaptureEngine :: enqueue ( MemcacheCommand ) if the parsed command is a response . This method is run in a thread and continues running until isShutdown () returns true . @TODO create several worker threads to process packets in parallel @protected
void CaptureEngine::processPackets(int worker_id, mqueue<Packet>* work_queue) {
  static int64_t pktCount = 0;
  static llui_t resCount = 0;
  static bool isDebug = logger->isDebug();
  Backoff backoff;
  backoff.setMaxElapsedTimeMillis(10);
  backoff.setMaxIntervalMillis(10);
  uint64_t backoffMs = 0;
  struct timespec waitTime;
  logger->info(CONTEXT, "Worker %d starting capture processing", worker_id);
  while(!isShutdown()) {
    Packet packet;
    if (work_queue->consume(packet)) {
      pktCount += 1;
      if (backoffMs > 0) {
        backoffMs = 0;
        backoff.reset();
      }
#ifdef _DEVEL
      logger->trace(CONTEXT,
                    "worker %d Consumed packet %ld", worker_id, packet.id());
#endif
      MemcacheCommand mc = parse(packet);
      if (mc.isResponse()) {
        enqueue(mc);
        resCount += 1;
#ifdef _DEBUG
        logger->trace(CONTEXT,
                      "worker %d, packet %ld, key %s", worker_id, packet.id(),
                      mc.getObjectKey().c_str());
#endif
      } else {
#ifdef _DEVEL
        logger->trace(CONTEXT, "worker %d not a memcache response", worker_id);
#endif
      }
      if ((pktCount % 10000) == 0 && isDebug) {
        string out = getStatsString();
        uint64_t now = UtilTime::currentTimeMillis();
        llui_t tdiff = now - packet.timestamp();
        out.append(", memcache replies = ");
        out.append(to_string(resCount));
        out.append(", time diff = ");
        out.append(to_string(tdiff));
        logger->debug(out);
      }
    } else {
      backoffMs = backoff.getNextBackOffMillis();
      waitTime = UtilTime::millisToTimespec(backoffMs);
#ifdef _DEVEL
      logger->trace(CONTEXT,
                    "worker %d no packet to consume, will sleep %lu ms",
                    worker_id, backoffMs);
#endif
    }
    if (backoffMs > 0) {
      nanosleep(&waitTime, NULL);
    }
  }
  logger->info(CONTEXT, "Worker %d stopped processing packets", worker_id);
}
void CaptureEngine :: processPackets ( int worker_id , mqueue < Packet > * work_queue ) { static int64_t pktCount = 0 ; static llui_t resCount = 0 ; static bool isDebug = logger -> isDebug ( ) ; Backoff backoff ; backoff . setMaxElapsedTimeMillis ( 10 ) ; backoff . setMaxIntervalMillis ( 10 ) ; uint64_t backoffMs = 0 ; struct timespec waitTime ; logger -> info ( CONTEXT , " " , worker_id ) ; while ( ! isShutdown ( ) ) { Packet packet ; if ( work_queue -> consume ( packet ) ) { pktCount += 1 ; if ( backoffMs > 0 ) { backoffMs = 0 ; backoff . reset ( ) ; } #ifdef _DEVEL logger -> trace ( CONTEXT , " " , worker_id , packet . id ( ) ) ; #endif MemcacheCommand mc = parse ( packet ) ; if ( mc . isResponse ( ) ) { enqueue ( mc ) ; resCount += 1 ; #ifdef _DEBUG logger -> trace ( CONTEXT , " " , worker_id , packet . id ( ) , mc . getObjectKey ( ) . c_str ( ) ) ; #endif } else { #ifdef _DEVEL logger -> trace ( CONTEXT , " " , worker_id ) ; #endif } if ( ( pktCount % 10000 ) == 0 && isDebug ) { string out = getStatsString ( ) ; uint64_t now = UtilTime :: currentTimeMillis ( ) ; llui_t tdiff = now - packet . timestamp ( ) ; out . append ( " " ) ; out . append ( to_string ( resCount ) ) ; out . append ( " " ) ; out . append ( to_string ( tdiff ) ) ; logger -> debug ( out ) ; } } else { backoffMs = backoff . getNextBackOffMillis ( ) ; waitTime = UtilTime :: millisToTimespec ( backoffMs ) ; #ifdef _DEVEL logger -> trace ( CONTEXT , " " , worker_id , backoffMs ) ; #endif } if ( backoffMs > 0 ) { nanosleep ( & waitTime , NULL ) ; } } logger -> info ( CONTEXT , " " , worker_id ) ; }
Process packets found in the packet queue.
Process packets found in the packet queue .