Subversion Repositories SmartDukaan

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

#include <sys/time.h>

#include "FacebookBase.h"
#include "ServiceTracker.h"
#include "concurrency/ThreadManager.h"

using namespace std;
using namespace facebook::fb303;
using namespace apache::thrift::concurrency;


uint64_t ServiceTracker::CHECKPOINT_MINIMUM_INTERVAL_SECONDS = 60;
int ServiceTracker::LOG_LEVEL = 5;


ServiceTracker::ServiceTracker(facebook::fb303::FacebookBase *handler,
                               void (*logMethod)(int, const string &),
                               bool featureCheckpoint,
                               bool featureStatusCheck,
                               bool featureThreadCheck,
                               Stopwatch::Unit stopwatchUnit)
  : handler_(handler), logMethod_(logMethod),
    featureCheckpoint_(featureCheckpoint),
    featureStatusCheck_(featureStatusCheck),
    featureThreadCheck_(featureThreadCheck),
    stopwatchUnit_(stopwatchUnit),
    checkpointServices_(0)
{
  if (featureCheckpoint_) {
    time_t now = time(NULL);
    checkpointTime_ = now;
  } else {
    checkpointTime_ = 0;
  }
}

/**
 * Registers the beginning of a "service method": basically, any of
 * the implementations of Thrift remote procedure calls that a
 * FacebookBase handler is handling.  Controls concurrent
 * services and reports statistics (via log and via fb303 counters).
 * Throws an exception if the server is not ready to handle service
 * methods yet.
 *
 * note: The relationship between startService() and finishService()
 * is currently defined so that a call to finishService() should only
 * be matched to this call to startService() if this method returns
 * without exception.  It wouldn't be a problem to implement things
 * the other way, so that *every* start needed a finish, but this
 * convention was chosen to match the way an object's constructor and
 * destructor work together, i.e. to work well with ServiceMethod
 * objects.
 *
 * @param const ServiceMethod &serviceMethod A reference to the ServiceMethod
 *                                           object instantiated at the start
 *                                           of the service method.
 */
void
ServiceTracker::startService(const ServiceMethod &serviceMethod)
{
  // note: serviceMethod.timer_ automatically starts at construction.

  // log service start
  logMethod_(5, serviceMethod.signature_);

  // check handler ready
  if (featureStatusCheck_ && !serviceMethod.featureLogOnly_) {
    // note: Throwing exceptions before counting statistics.  See note
    // in method header.
    // note: A STOPPING server is not accepting new connections, but it
    // is still handling any already-connected threads -- so from the
    // service method's point of view, a status of STOPPING is a green
    // light.
    facebook::fb303::fb_status status = handler_->getStatus();
    if (status != facebook::fb303::ALIVE
        && status != facebook::fb303::STOPPING) {
      if (status == facebook::fb303::STARTING) {
        throw ServiceException("Server starting up; please try again later");
      } else {
        throw ServiceException("Server not alive; please try again later");
      }
    }
  }

  // check server threads
  if (featureThreadCheck_ && !serviceMethod.featureLogOnly_) {
    // note: Might want to put these messages in reportCheckpoint() if
    // log is getting spammed.
    if (threadManager_ != NULL) {
      size_t idle_count = threadManager_->idleWorkerCount();
      if (idle_count == 0) {
        stringstream message;
        message << "service " << serviceMethod.signature_
                << ": all threads (" << threadManager_->workerCount()
                << ") in use";
        logMethod_(3, message.str());
      }
    }
  }
}

/**
 * Logs a significant step in the middle of a "service method"; see
 * startService.
 *
 * @param const ServiceMethod &serviceMethod A reference to the ServiceMethod
 *                                           object instantiated at the start
 *                                           of the service method.
 * @return int64_t Elapsed units (see stopwatchUnit_) since ServiceMethod
 *                 instantiation.
 */
int64_t
ServiceTracker::stepService(const ServiceMethod &serviceMethod,
                            const string &stepName)
{
  stringstream message;
  string elapsed_label;
  int64_t elapsed = serviceMethod.timer_.elapsedUnits(stopwatchUnit_,
                                                      &elapsed_label);
  message << serviceMethod.signature_
          << ' ' << stepName
          << " [" << elapsed_label << ']';
  logMethod_(5, message.str());
  return elapsed;
}

/**
 * Registers the end of a "service method"; see startService().
 *
 * @param const ServiceMethod &serviceMethod A reference to the ServiceMethod
 *                                           object instantiated at the start
 *                                           of the service method.
 */
void
ServiceTracker::finishService(const ServiceMethod &serviceMethod)
{
  // log end of service
  stringstream message;
  string duration_label;
  int64_t duration = serviceMethod.timer_.elapsedUnits(stopwatchUnit_,
                                                       &duration_label);
  message << serviceMethod.signature_
          << " finish [" << duration_label << ']';
  logMethod_(5, message.str());

  // count, record, and maybe report service statistics
  if (!serviceMethod.featureLogOnly_) {

    if (!featureCheckpoint_) {

      // lifetime counters
      // (note: No need to lock statisticsMutex_ if not doing checkpoint;
      // FacebookService::incrementCounter() is already thread-safe.)
      handler_->incrementCounter("lifetime_services");

    } else {

      statisticsMutex_.lock();
      // note: No exceptions expected from this code block.  Wrap in a try
      // just to be safe.
      try {

        // lifetime counters
        // note: Good to synchronize this with the increment of
        // checkpoint services, even though incrementCounter() is
        // already thread-safe, for the sake of checkpoint reporting
        // consistency (i.e.  since the last checkpoint,
        // lifetime_services has incremented by checkpointServices_).
        handler_->incrementCounter("lifetime_services");

        // checkpoint counters
        checkpointServices_++;
        checkpointDuration_ += duration;

        // per-service timing
        // note kjv: According to my tests it is very slightly faster to
        // call insert() once (and detect not-found) than calling find()
        // and then maybe insert (if not-found).  However, the difference
        // is tiny for small maps like this one, and the code for the
        // faster solution is slightly less readable.  Also, I wonder if
        // the instantiation of the (often unused) pair to insert makes
        // the first algorithm slower after all.
        map<string, pair<uint64_t, uint64_t> >::iterator iter;
        iter = checkpointServiceDuration_.find(serviceMethod.name_);
        if (iter != checkpointServiceDuration_.end()) {
          iter->second.first++;
          iter->second.second += duration;
        } else {
          checkpointServiceDuration_.insert(make_pair(serviceMethod.name_,
                                                      make_pair(1, duration)));
        }

        // maybe report checkpoint
        // note: ...if it's been long enough since the last report.
        time_t now = time(NULL);
        uint64_t check_interval = now - checkpointTime_;
        if (check_interval >= CHECKPOINT_MINIMUM_INTERVAL_SECONDS) {
          reportCheckpoint();
        }

      } catch (...) {
        statisticsMutex_.unlock();
        throw;
      }
      statisticsMutex_.unlock();

    }
  }
}

/**
 * Logs some statistics gathered since the last call to this method.
 *
 * note: Thread race conditions on this method could cause
 * misreporting and/or undefined behavior; the caller must protect
 * uses of the object variables (and calls to this method) with a
 * mutex.
 *
 */
void
ServiceTracker::reportCheckpoint()
{
  time_t now = time(NULL);

  uint64_t check_count = checkpointServices_;
  uint64_t check_interval = now - checkpointTime_;
  uint64_t check_duration = checkpointDuration_;

  // export counters for timing of service methods (by service name)
  handler_->setCounter("checkpoint_time", check_interval);
  map<string, pair<uint64_t, uint64_t> >::iterator iter;
  uint64_t count;
  for (iter = checkpointServiceDuration_.begin();
       iter != checkpointServiceDuration_.end();
       iter++) {
    count = iter->second.first;
    handler_->setCounter(string("checkpoint_count_") + iter->first, count);
    if (count == 0) {
      handler_->setCounter(string("checkpoint_speed_") + iter->first,
                           0);
    } else {
      handler_->setCounter(string("checkpoint_speed_") + iter->first,
                           iter->second.second / count);
    }
  }

  // reset checkpoint variables
  // note: Clearing the map while other threads are using it might
  // cause undefined behavior.
  checkpointServiceDuration_.clear();
  checkpointTime_ = now;
  checkpointServices_ = 0;
  checkpointDuration_ = 0;

  // get lifetime variables
  uint64_t life_count = handler_->getCounter("lifetime_services");
  uint64_t life_interval = now - handler_->aliveSince();

  // log checkpoint
  stringstream message;
  message << "checkpoint_time:" << check_interval
          << " checkpoint_services:" << check_count
          << " checkpoint_speed_sum:" << check_duration
          << " lifetime_time:" << life_interval
          << " lifetime_services:" << life_count;
  if (featureThreadCheck_ && threadManager_ != NULL) {
    size_t worker_count = threadManager_->workerCount();
    size_t idle_count = threadManager_->idleWorkerCount();
    message << " total_workers:" << worker_count
            << " active_workers:" << (worker_count - idle_count);
  }
  logMethod_(4, message.str());
}

/**
 * Remembers the thread manager used in the server, for monitoring thread
 * activity.
 *
 * @param shared_ptr<ThreadManager> threadManager The server's thread manager.
 */
void
ServiceTracker::setThreadManager(boost::shared_ptr<ThreadManager>
                                 threadManager)
{
  threadManager_ = threadManager;
}

/**
 * Logs messages to stdout; the passed message will be logged if the
 * passed level is less than or equal to LOG_LEVEL.
 *
 * This is the default logging method used by the ServiceTracker.  An
 * alternate logging method (that accepts the same parameters) may be
 * specified to the constructor.
 *
 * @param int level A level associated with the message: higher levels
 *                  are used to indicate higher levels of detail.
 * @param string message The message to log.
 */
void
ServiceTracker::defaultLogMethod(int level, const string &message)
{
  if (level <= LOG_LEVEL) {
    string level_string;
    time_t now = time(NULL);
    char now_pretty[26];
    ctime_r(&now, now_pretty);
    now_pretty[24] = '\0';
    switch (level) {
    case 1:
      level_string = "CRITICAL";
      break;
    case 2:
      level_string = "ERROR";
      break;
    case 3:
      level_string = "WARNING";
      break;
    case 5:
      level_string = "DEBUG";
      break;
    case 4:
    default:
      level_string = "INFO";
      break;
    }
    cout << '[' << level_string << "] [" << now_pretty << "] "
         << message << endl;
  }
}


/**
 * Creates a Stopwatch, which can report the time elapsed since its
 * creation.
 *
 */
Stopwatch::Stopwatch()
{
  gettimeofday(&startTime_, NULL);
}

void
Stopwatch::reset()
{
  gettimeofday(&startTime_, NULL);
}

uint64_t
Stopwatch::elapsedUnits(Stopwatch::Unit unit, string *label) const
{
  timeval now_time;
  gettimeofday(&now_time, NULL);
  time_t duration_secs = now_time.tv_sec - startTime_.tv_sec;

  uint64_t duration_units;
  switch (unit) {
  case UNIT_SECONDS:
    duration_units = duration_secs
      + (now_time.tv_usec - startTime_.tv_usec + 500000) / 1000000;
    if (NULL != label) {
      stringstream ss_label;
      ss_label << duration_units << " secs";
      label->assign(ss_label.str());
    }
    break;
  case UNIT_MICROSECONDS:
    duration_units = duration_secs * 1000000
      + now_time.tv_usec - startTime_.tv_usec;
    if (NULL != label) {
      stringstream ss_label;
      ss_label << duration_units << " us";
      label->assign(ss_label.str());
    }
    break;
  case UNIT_MILLISECONDS:
  default:
    duration_units = duration_secs * 1000
      + (now_time.tv_usec - startTime_.tv_usec + 500) / 1000;
    if (NULL != label) {
      stringstream ss_label;
      ss_label << duration_units << " ms";
      label->assign(ss_label.str());
    }
    break;
  }
  return duration_units;
}

/**
 * Creates a ServiceMethod, used for tracking a single service method
 * invocation (via the ServiceTracker).  The passed name of the
 * ServiceMethod is used to group statistics (e.g. counts and durations)
 * for similar invocations; the passed signature is used to uniquely
 * identify the particular invocation in the log.
 *
 * note: A version of this constructor is provided that automatically
 * forms a signature the name and a passed numeric id.  Silly, sure,
 * but commonly used, since it often saves the caller a line or two of
 * code.
 *
 * @param ServiceTracker *tracker The service tracker that will track this
 *                                ServiceMethod.
 * @param const string &name The service method name (usually independent
 *                           of service method parameters).
 * @param const string &signature A signature uniquely identifying the method
 *                                invocation (usually name plus parameters).
 */
ServiceMethod::ServiceMethod(ServiceTracker *tracker,
                             const string &name,
                             const string &signature,
                             bool featureLogOnly)
  : tracker_(tracker), name_(name), signature_(signature),
    featureLogOnly_(featureLogOnly)
{
  // note: timer_ automatically starts at construction.

  // invoke tracker to start service
  // note: Might throw.  If it throws, then this object's destructor
  // won't be called, which is according to plan: finishService() is
  // only supposed to be matched to startService() if startService()
  // returns without error.
  tracker_->startService(*this);
}

ServiceMethod::ServiceMethod(ServiceTracker *tracker,
                             const string &name,
                             uint64_t id,
                             bool featureLogOnly)
  : tracker_(tracker), name_(name), featureLogOnly_(featureLogOnly)
{
  // note: timer_ automatically starts at construction.
  stringstream ss_signature;
  ss_signature << name << " (" << id << ')';
  signature_ = ss_signature.str();

  // invoke tracker to start service
  // note: Might throw.  If it throws, then this object's destructor
  // won't be called, which is according to plan: finishService() is
  // only supposed to be matched to startService() if startService()
  // returns without error.
  tracker_->startService(*this);
}

ServiceMethod::~ServiceMethod()
{
  // invoke tracker to finish service
  // note: Not expecting an exception from this code, but
  // finishService() might conceivably throw an out-of-memory
  // exception.
  try {
    tracker_->finishService(*this);
  } catch (...) {
    // don't throw
  }
}

uint64_t
ServiceMethod::step(const std::string &stepName)
{
  return tracker_->stepService(*this, stepName);
}