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);

}