dnxTimer.c

Go to the documentation of this file.

/*--------------------------------------------------------------------------

   Copyright (c) 2006-2010, Intellectual Reserve, Inc. All rights reserved.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License version 2 as
   published by the Free Software Foundation.

   This program 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 General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

  --------------------------------------------------------------------------*/

#include "dnxTimer.h"

#include "dnxServerMain.h"
#include "dnxError.h"
#include "dnxDebug.h"
#include "dnxProtocol.h"
#include "dnxJobList.h"
#include "dnxLogging.h"
#include "dnxSleep.h"
#include "dnxStats.h"

#if HAVE_CONFIG_H
# include "config.h"
#endif

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <assert.h>

#define DNX_DEF_TIMER_SLEEP   5000  
#define MAX_EXPIRED           10    
typedef struct iDnxTimer_
{
   DnxJobList * joblist;   
   pthread_t tid;          
   int sleepms;            
} iDnxTimer;

/*--------------------------------------------------------------------------
                              IMPLEMENTATION
  --------------------------------------------------------------------------*/

static void dnxTimerCleanup(void * data)
{
   iDnxTimer * itimer = (iDnxTimer *)data;
   assert(data);
}

//----------------------------------------------------------------------------

static void * dnxTimer(void * data)
{
   iDnxTimer * itimer = (iDnxTimer *)data;
   DnxNewJob ExpiredList[MAX_EXPIRED];
   int i, totalExpired;
   int ret = 0;

   assert(data);

   pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, 0);
   pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, 0);
   pthread_cleanup_push(dnxTimerCleanup, data);

   dnxLog("dnxTimer[%lx]: Watching for expired jobs...", pthread_self());

   while (1)
   {
      pthread_testcancel();

      dnxCancelableSleep(itimer->sleepms);

      // search for expired jobs in the pending queue
      totalExpired = MAX_EXPIRED;
      if ((ret = dnxJobListExpire(itimer->joblist, ExpiredList, 
            &totalExpired)) == DNX_OK && totalExpired > 0)
      {
         for (i = 0; i < totalExpired; i++)
         {
            char msg[256];
            char addrstr[DNX_MAX_ADDRSTR];
            DnxNewJob * job = &ExpiredList[i];

            dnxDebug(1, "dnxTimer[%lx]: Expiring Job [%lu,%lu]: %s.",
                  pthread_self(), job->xid.objSerial, job->xid.objSlot, job->cmd);

            dnxStatsInc(job->pNode->address, RESULTS_TIMED_OUT);
            dnxAuditJob(job, "EXPIRE");

//          if (job->ack)
               snprintf(msg, sizeof msg, 
                     "(DNX: Service Check [%lu,%lu] Timed Out - "
                     "Node: %s - Failed to return job response in time allowed)",
                     job->xid.objSerial, job->xid.objSlot, addrstr);
//          else
//             snprintf(msg, sizeof msg, 
//                   "(DNX: Service Check [%lu,%lu] Timed Out - "
//                   "Node: %s - Failed to acknowledge job receipt)",
//                   job->xid.objSerial, job->xid.objSlot, addrstr);

            dnxDebug(2, msg);

            // report the expired job to Nagios
            ret = dnxPostResult(job->payload, job->xid.objSerial, job->start_time, 
                  time(0) - job->start_time, 1, 0, msg);
            dnxJobCleanup(job);
         }
      }

      if (totalExpired > 0 || ret != DNX_OK)
         dnxDebug(2, "dnxTimer[%lx]: Expired job count: %d  Retcode=%d: %s.",
               pthread_self(), totalExpired, ret, dnxErrorString(ret));
   }

   dnxLog("dnxTimer[%lx]: Terminating: %s.", pthread_self(), dnxErrorString(ret));

   pthread_cleanup_pop(1);
   return 0;
}

/*--------------------------------------------------------------------------
                                 INTERFACE
  --------------------------------------------------------------------------*/

int dnxTimerCreate(DnxJobList * joblist, int sleeptime, DnxTimer ** ptimer)
{
   iDnxTimer * itimer;
   int ret;

   assert(joblist && ptimer);

   // don't allow sleep times outside the range 1/10th sec to 5 minutes
   if (sleeptime < 100 || sleeptime > 300000)
      sleeptime = DNX_DEF_TIMER_SLEEP;

   if ((itimer = (iDnxTimer *)xmalloc(sizeof *itimer)) == 0)
      return DNX_ERR_MEMORY;

   // initialize the itimer
   memset(itimer, 0, sizeof *itimer);
   itimer->joblist = joblist;
   itimer->sleepms = sleeptime;

   // create the timer thread
   if ((ret = pthread_create(&itimer->tid, 0, dnxTimer, itimer)) != 0)
   {
      dnxLog("Timer thread creation failed: %s.", dnxErrorString(ret));
      xfree(itimer);
      return DNX_ERR_THREAD;
   }

   *ptimer = (DnxTimer *)itimer;

   return DNX_OK;
}

//----------------------------------------------------------------------------

void dnxTimerDestroy(DnxTimer * timer)
{
   iDnxTimer * itimer = (iDnxTimer *)timer;

   pthread_cancel(itimer->tid);
   pthread_join(itimer->tid, 0);

   xfree(itimer);
}

/*--------------------------------------------------------------------------
                                 UNIT TEST

   From within dnx/server, compile with GNU tools using this command line:

      gcc -DDEBUG -DDNX_TIMER_TEST -DHAVE_NANOSLEEP -g -O0 -I../common \
         -I../nagios/nagios-2.7/include dnxTimer.c ../common/dnxSleep.c \
         ../common/dnxError.c -lpthread -lgcc_s -lrt -o dnxTimerTest

   Note: Leave out -DHAVE_NANOSLEEP if your system doesn't have nanosleep.

  --------------------------------------------------------------------------*/

#ifdef DNX_TIMER_TEST

#include "utesthelp.h"
#include <time.h>

#define elemcount(x) (sizeof(x)/sizeof(*(x)))

static int verbose;
static DnxNewJob fakejob;
static DnxJobList fakejoblist;
static int fakepayload;
static DnxNodeRequest fakenode;
static int entered_dnxJobListExpire;

// functional stubs
IMPLEMENT_DNX_SYSLOG(verbose);
IMPLEMENT_DNX_DEBUG(verbose);

int dnxJobListExpire(DnxJobList * pJobList, DnxNewJob * pExpiredJobs, int * totalJobs)
{
   CHECK_TRUE(pJobList == &fakejoblist);
   CHECK_TRUE(*totalJobs > 0);
   memcpy(&pExpiredJobs[0], &fakejob, sizeof *pExpiredJobs);
   *totalJobs = 1;
   entered_dnxJobListExpire = 1;
   return 0;
}

int dnxPostResult(void * payload, unsigned long serial, 
      time_t start_time, unsigned delta, int early_timeout, 
      int res_code, char * res_data)
{
   CHECK_TRUE(payload == &fakepayload);
   CHECK_TRUE(start_time == 100);
   CHECK_TRUE(early_timeout != 0);
   CHECK_TRUE(res_code == 0);
   CHECK_ZERO(memcmp(res_data, "(DNX: Service", 13));
   return 0;
}

int dnxAuditJob(DnxNewJob * pJob, char * action) { return 0; }

void dnxJobCleanup(DnxNewJob * pJob) { }

void dnxStatsInc(char * addr, DnxStatsIndex member) {}

int main(int argc, char ** argv)
{
   DnxTimer * timer;
   iDnxTimer * itimer;

   verbose = argc > 1? 1: 0;

   // setup test harness
   fakenode.xid.objType    = DNX_OBJ_JOB;
   fakenode.xid.objSerial  = 1;
   fakenode.xid.objSlot    = 2;
   fakenode.reqType        = DNX_REQ_DEREGISTER;
   fakenode.jobCap         = 1;
   fakenode.ttl            = 2;
   fakenode.expires        = 3;
   strcpy(fakenode.address, "fake address");

   fakejob.state           = DNX_JOB_INPROGRESS;
   fakejob.xid.objType     = DNX_OBJ_JOB;
   fakejob.xid.objSerial   = 1;
   fakejob.xid.objSlot     = 2;
   fakejob.cmd             = "fake command line";
   fakejob.start_time      = 100;
   fakejob.timeout         = 10;
   fakejob.expires         = fakejob.start_time + fakejob.timeout;
   fakejob.payload         = &fakepayload;
   fakejob.pNode           = &fakenode;

   entered_dnxJobListExpire = 0;

   // create a short timer and reference it as a concrete object for testing
   CHECK_ZERO(dnxTimerCreate(&fakejoblist, 100, &timer));
   itimer = (iDnxTimer *)timer;

   // check internal state
   CHECK_TRUE(itimer->joblist == &fakejoblist);
   CHECK_TRUE(itimer->tid != 0);
   CHECK_TRUE(itimer->sleepms == 100);

   // wait for timer to have made one pass though timer thread loop
   while (!entered_dnxJobListExpire)
      dnxCancelableSleep(10);

   // shut down
   dnxTimerDestroy(timer);

   return 0;
}

#endif   /* DNX_TIMER_TEST */

/*--------------------------------------------------------------------------*/