timing.C

Go to the documentation of this file.

// Copyright (C) 2004, Kenneth C. Schalk
// 
// This file is part of Vesta.
// 
// Vesta is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 2.1 of the License, or (at your option) any later version.
// 
// Vesta is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
// Lesser General Public License for more details.
// 
// You should have received a copy of the GNU Lesser General Public
// License along with Vesta; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

// timing.C - implementation of fine-grained NFS call timing
// recording, one of the performance debugging features

// Last modified on Tue Dec 13 16:23:39 EST 2005 by ken@xorian.net

#include <BufStream.H>
#include <iomanip>

#include <Basics.H>

#include "VestaSource.H"
#include <VestaConfig.H>

#include "logging.H"
#include "timing.H"

#include <assert.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>

using std::setw;
using std::setfill;
using Basics::OBufStream;

static pthread_key_t timing_key;
static pthread_once_t timing_once = PTHREAD_ONCE_INIT;

// Should we be recording right now?  (Note: *not* protected by a
// mutex, which makes for a race condition.  We don't really care
// though.  A timing recorder should see its value on the next NFS
// call anyway.)
static bool recording_timing = false;

Basics::mutex popen_mu;

static unsigned int timing_file_minutes = 60;

extern "C"
{
  static void timing_init()
  {
    int err = pthread_key_create(&timing_key, NULL);
    assert(err == 0);

    try
      {
        Text unused;
        if(VestaConfig::get("Repository",
                            "timing_file_minutes", unused))
          timing_file_minutes = VestaConfig::get_int("Repository",
                                                     "timing_file_minutes");
      }
    catch (VestaConfig::failure f)
      {
        Repos::dprintf(DBG_ALWAYS,
                       "VestaConfig::failure in VestaSourceServerExport: %s\n",
                       f.msg.cchars());
        abort();
      }
  
  }
}

void timing_start_call(unsigned int proc)
{
  pthread_once(&timing_once, timing_init);
  Timing_Recorder *recorder =
    (Timing_Recorder *) pthread_getspecific(timing_key);

  if(recorder != 0)
    {
      recorder->start(proc);
    }
}

void timing_dupe_status(int new_call)
{
  pthread_once(&timing_once, timing_init);
  Timing_Recorder *recorder =
    (Timing_Recorder *) pthread_getspecific(timing_key);

  if(recorder != 0)
    {
      recorder->dupe_status(new_call != 0);
    }
}

void timing_record_time_point(const char *file, unsigned int line)
{
  pthread_once(&timing_once, timing_init);
  Timing_Recorder *recorder =
    (Timing_Recorder *) pthread_getspecific(timing_key);

  if(recorder != 0)
    {
      recorder->time_point(file, line);
    }
}

void timing_end_call(/*OUT*/ struct timeval *delta_time)
{
  pthread_once(&timing_once, timing_init);
  Timing_Recorder *recorder =
    (Timing_Recorder *) pthread_getspecific(timing_key);

  if(recorder != 0)
    {
      struct timeval delta = recorder->end();
      if(delta_time != 0)
        {
          *delta_time = delta;
        }
    }
  else if(delta_time != 0)
    {
      delta_time->tv_sec = 0;
      delta_time->tv_usec = 0;
    }
}

int timing_in_call()
{
  pthread_once(&timing_once, timing_init);
  Timing_Recorder *recorder =
    (Timing_Recorder *) pthread_getspecific(timing_key);

  if(recorder != 0)
    {
      return (recorder->in_call() ? 1 : 0);
    }
  return 0;
}

void timing_control(bool enable)
{
  recording_timing = enable;
}

void timing_record_LongId(const LongId &lid)
{
  pthread_once(&timing_once, timing_init);
  Timing_Recorder *recorder =
    (Timing_Recorder *) pthread_getspecific(timing_key);

  if(recorder != 0)
    {
      recorder->record_LongId(lid);
    }
}


static unsigned long thread_index()
{
#if defined(__linux__)
  return pthread_self()/1024;
#elif defined(__osf__)
  return pthread_getsequence_np(pthread_self());
#else
#error Need way to get thread index for this platform
#endif
}

#define USECS_PER_SEC 1000000

static struct timeval delta_time(const struct timeval &time_a,
                                 const struct timeval &time_b)
{
  struct timeval result;

  if(time_a.tv_sec == time_b.tv_sec)
    {
      result.tv_sec = 0;
      // Need to handle time moving backwards thanks to ntp
      result.tv_usec = ((time_b.tv_usec >= time_a.tv_usec)
                        ? time_b.tv_usec - time_a.tv_usec
                        : 0);
    }
  else if(time_b.tv_sec > time_a.tv_sec)
    {
      result.tv_sec = (time_b.tv_sec - time_a.tv_sec) - 1;
      result.tv_usec = (time_b.tv_usec + USECS_PER_SEC) - time_a.tv_usec;
      if(result.tv_usec > USECS_PER_SEC)
        {
          result.tv_sec += result.tv_usec/USECS_PER_SEC;
          result.tv_usec %= USECS_PER_SEC;
        }
    }
  else
    {
      // Need to handle time moving backwards thanks to ntp
      result.tv_sec = 0;
      result.tv_usec = 0;
    }

  return result;
}

static pthread_once_t timing_cleanup_once = PTHREAD_ONCE_INIT;
static Basics::mutex recorder_list_mu;
Timing_Recorder *Timing_Recorder::first = 0;

extern "C"
{
  void Timing_Recorder_exit_cleanup() throw()
  {
    recorder_list_mu.lock();
    Timing_Recorder *cur = Timing_Recorder::first;
    while(cur)
      {
        Timing_Recorder *next = cur->next;
        delete cur;
        cur = next;
      }
    Timing_Recorder::first = 0;
    recorder_list_mu.unlock();
  }

  void Timing_Recorder_exit_cleanup_setup() throw()
  {
    atexit(Timing_Recorder_exit_cleanup);
  }
}

Timing_Recorder::Timing_Recorder()
  : current_byte_count(0), timing_record_stream(0), file_end_time(0)
{
  pthread_once(&timing_once, timing_init);

  // Only one timing recorder per thread
  assert(pthread_getspecific(timing_key) == 0);

  // This is the timing recorder for this thread.
  int err = pthread_setspecific(timing_key, (void *) this);
  assert(err == 0);

  // Make sure everything gets cleaned up at exit time.
  pthread_once(&timing_cleanup_once, Timing_Recorder_exit_cleanup_setup);

  // Insert this recorder into the list
  recorder_list_mu.lock();
  this->next = Timing_Recorder::first;
  Timing_Recorder::first = this;
  recorder_list_mu.unlock();
}

void Timing_Recorder::start_new_file(struct timeval *current_time)
{
  // Close existing stream, if any.
  if(timing_record_stream != 0)
    {
      fflush(timing_record_stream);
      popen_mu.lock();
      pclose(timing_record_stream);
      popen_mu.unlock();
    }

  // Get the year/month/day/hour/minute for use in the timing filename

  // If this is the first file we're opening, use the current time
  // rounded down to the specified number of minutes.
  struct tm time_detail;
  if(file_end_time == 0)
    {
      // Get the current time if the caller didn't provide it.
      struct timeval temp_time;
      if(current_time == 0)
        {
          struct timezone unused_tz;

          gettimeofday(&temp_time, &unused_tz);

          current_time = &temp_time;
        }

      localtime_r(&(current_time->tv_sec), &time_detail);
      time_detail.tm_sec = 0;
      time_detail.tm_min -= (time_detail.tm_min%timing_file_minutes);
    }
  // Use the end time for the last file to label the next file
  else
    {
      time_t file_label_time = file_end_time;
      localtime_r(&file_label_time, &time_detail);
    }

  // Generate a name for a directory to put the timings into, and
  // create it if it doesn't exist.
  OBufStream l_dirname;
  l_dirname << "timings."
            << (time_detail.tm_year+1900) << "-"
            << setw(2) << setfill('0') << (time_detail.tm_mon + 1) << "-"
            << setw(2) << setfill('0') << time_detail.tm_mday;
  if(!FS::Exists(l_dirname.str()))
    {
      mkdir(l_dirname.str(), 0755);
    }

  // Open the stream we'll record our results into.  We start a gzip
  // process to compress them as we write them.
  OBufStream l_cmd;
  l_cmd << "exec gzip - > " << l_dirname.str()
        << "/nfs_calls.pid_" << getpid()
        << ".thread_" << thread_index()
        << "."
        << setw(2) << setfill('0') << time_detail.tm_hour << "_"
        << setw(2) << setfill('0') << time_detail.tm_min;
  popen_mu.lock();
  timing_record_stream = popen(l_cmd.str(), "w");
  popen_mu.unlock();
  assert(timing_record_stream != 0);

  // Compute the time at which we should start a new file (the
  // beginning of the next interval).
  file_end_time = mktime(&time_detail) + 60*timing_file_minutes;
}

Timing_Recorder::~Timing_Recorder()
{
  if(timing_record_stream != 0)
    {
      fflush(timing_record_stream);
      popen_mu.lock();
      pclose(timing_record_stream);
      popen_mu.unlock();
    }
}

void Timing_Recorder::start(unsigned int proc)
{
  // Can't start a new call until we finish the last one.
  assert(current_byte_count == 0);

  // Get the start time
  struct timezone unused_tz;
  int err = gettimeofday(&(this->start_time), &unused_tz);
  assert(err == 0);

  // If we're not supposed to be recording timing, do nothing else.
  if(!recording_timing)
    return;

  // Start a new timing file if we should
  if((timing_record_stream == 0) || (this->start_time.tv_sec >= file_end_time))
    {
      start_new_file(&(this->start_time));

      // Get the time again to avoid counting the change of file for
      // this call.
      err = gettimeofday(&(this->start_time), &unused_tz);
      assert(err == 0);
    }

  // ------------------------------------------------------------
  // Place the start record into the buffer
  // ------------------------------------------------------------

  // Start marker byte
  write_buffer[0] = 1;

  // NFS procedure
  write_buffer[1] = proc;
  assert(write_buffer[1] == proc);

  // Start time seconds
  unsigned int start_sec = this->start_time.tv_sec;
  memcpy(&(write_buffer[2]), (void *) &start_sec, sizeof(unsigned int));

  // Start time microseconds
  unsigned int start_usec = this->start_time.tv_usec;
  memcpy(&(write_buffer[6]), (void *) &start_usec, sizeof(unsigned int));

  // Set initial byte count
  current_byte_count = 2+(sizeof(unsigned int)*2);
  assert(current_byte_count == 10);
}

struct timeval Timing_Recorder::record_delta()
{
  // Get the current time
  struct timeval current_time;
  struct timezone unused_tz;
  int err = gettimeofday(&current_time, &unused_tz);
  assert(err == 0);

  // Find the time delta
  struct timeval delta = delta_time(this->start_time, current_time);

  // Seconds delta from start
  unsigned short sec_delta = delta.tv_sec;
  assert(sec_delta == delta.tv_sec);
  memcpy(&(write_buffer[current_byte_count]),
         (void *) &sec_delta, sizeof(unsigned short));
  assert(sizeof(unsigned short) == 2);
  current_byte_count += sizeof(unsigned short);

  // Microseconds delta from start
  unsigned int usec_delta = delta.tv_usec;
  memcpy(&(write_buffer[current_byte_count]),
         (void *) &usec_delta, sizeof(unsigned int));
  assert(sizeof(unsigned int) == 4);
  current_byte_count += sizeof(unsigned int);

  return delta;
}

void Timing_Recorder::dupe_status(bool new_call)
{
  // If we're not in a call, do nothing.
  if(!in_call())
    return;

  // ------------------------------------------------------------
  // Place the duplicate record into the buffer
  // ------------------------------------------------------------

  // Duplicate status marker byte
  write_buffer[current_byte_count++] = (new_call ? 4 : 5);

  // Time delta from start
  (void) record_delta();

  // Make sure we haven't run out of buffer space.
  assert(current_byte_count < TIMING_RECORDER_BUF_SIZE);
}

void Timing_Recorder::time_point(const char *file, unsigned short line)
{
  // If we're not in a call, do nothing.
  if(!in_call())
    return;

  // ------------------------------------------------------------
  // Place the time point record into the buffer
  // ------------------------------------------------------------

  // Time point marker byte
  write_buffer[current_byte_count++] = 2;

  // Time delta from start
  (void) record_delta();

  // Source file
  strcpy(&(write_buffer[current_byte_count]), file);
  current_byte_count += strlen(file)+1;

  // Source line number
  memcpy(&(write_buffer[current_byte_count]),
         (void *) &line, sizeof(unsigned short));
  current_byte_count += sizeof(unsigned short);

  // Make sure we haven't run out of buffer space.
  assert(current_byte_count < TIMING_RECORDER_BUF_SIZE);
}

struct timeval Timing_Recorder::end()
{
  // If we're not in a call, just compute and return the call delta.
  if(!in_call())
    {
      // Get the current time
      struct timeval current_time;
      struct timezone unused_tz;
      int err = gettimeofday(&current_time, &unused_tz);
      assert(err == 0);

      // Find the time delta
      return delta_time(this->start_time, current_time);
    }

  // ------------------------------------------------------------
  // Place the end record into the buffer
  // ------------------------------------------------------------

  // End record marker byte
  write_buffer[current_byte_count++] = 3;

  // Time delta from start
  struct timeval delta = record_delta();

  // Make sure we haven't run out of buffer space.
  assert(current_byte_count < TIMING_RECORDER_BUF_SIZE);

  // Write the entire record for this call
  size_t written = fwrite(write_buffer, current_byte_count, 1,
                          timing_record_stream);
  assert(written == 1);
  fflush(timing_record_stream);

  // Reset the byte count
  current_byte_count = 0;

  // Return the delta since the start of call.
  return delta;
}

void Timing_Recorder::record_LongId(const LongId &lid)
{
  // If we're not in a call, do nothing.
  if(!in_call())
    return;

  // ------------------------------------------------------------
  // Place the LongId record into the buffer
  // ------------------------------------------------------------

  // LongId marker byte
  write_buffer[current_byte_count++] = 6;

  // Store the LongId
  memcpy(&(write_buffer[current_byte_count]), &(lid.value.byte[0]),
         sizeof(lid.value));
  current_byte_count += sizeof(lid.value);

  // Make sure we haven't run out of buffer space.
  assert(current_byte_count < TIMING_RECORDER_BUF_SIZE);
}

---