sics/nwatch.c

/*
 * N E T   W A T C H E R
 *
 * This module watches network connections for sockets becoming readable or
 * writeable and invokes callbacks. It also provides a timer mechanism.
 *
 * Douglas Clowes, February 2007
 *
 */

#include <stdlib.h>
#include <assert.h>
#ifdef CYGNUS
#include <sys/socket.h>
#else
#include <sys/select.h>
#endif
#include <sys/time.h>
#include "fortify.h"
#include "nwatch.h"
#include "uselect.h"
#include "sics.h"

#define NWMAGIC 51966

/* Net Watcher control structure */
typedef struct __netwatcher_s {
  pNWContext cq_head;           /* head of socket context queue */
  pNWContext cq_tail;           /* tail of socket context queue */
  int nInvalid;                 /* number of invalidated entries */
  pNWTimer tq_head;             /* head of timer context queue */
  pNWTimer tq_tail;             /* tail of timer context queue */
  long lMagic;                  /* integrity check */
} NetWatch, *pNetWatch;

/* Singleton pattern */
static pNetWatch instance = NULL;

/**
 * \brief Initialises the Net Watcher singleton and starts the task
 *
 * \return 1=success, 0=failure
 */
static pNetWatch NetWatchGetInstance(void)
{
  /*
   * If the singleton has not yet been created, do so now
   */
  if (instance == NULL) {
    instance = (pNetWatch) malloc(sizeof(NetWatch));
    if (instance == NULL)
      return 0;
    memset(instance, 0, sizeof(NetWatch));
    instance->lMagic = NWMAGIC;
  }
  return instance;
}

/*
 * The timer context object private definition
 */
typedef struct __netwatchtimer {
  pNWTimer next;                /* chain to next event */
  struct timeval tv;            /* time when event is due */
  pNWCallback func;             /* function to call */
  void *cntx;                   /* abstract context to pass to callback */
  int msec;                     /* millisecond delay time */
  int tick;                     /* millisecond repeat rate */
  long int vrfy;                /* integrity check */
} NWTimer;

static pNWTimer activeTimer = NULL;

/*
 * \brief private function to insert an entry into the sorted timer queue.
 *
 * \param self singleton
 * \param handle new timer to insert
 */
static int NetWatchTimerInsQue(pNetWatch self, pNWTimer handle)
{
  /* if the queue is empty, just stick new one in */
  if (self->tq_head == NULL) {
    self->tq_head = self->tq_tail = handle;
    handle->next = NULL;
    handle->vrfy = NWMAGIC;
    return 1;
  }
  /* if new one is not earlier than latest one, insert after latest */
  if (handle->tv.tv_sec > self->tq_tail->tv.tv_sec ||
      (handle->tv.tv_sec == self->tq_tail->tv.tv_sec &&
       handle->tv.tv_usec >= self->tq_tail->tv.tv_usec)) {
    self->tq_tail->next = handle;
    self->tq_tail = handle;
    handle->next = NULL;
    handle->vrfy = NWMAGIC;
    return 1;
  }
  /* if new one is not later than earliest one, insert before earliest */
  if (handle->tv.tv_sec < self->tq_head->tv.tv_sec ||
      (handle->tv.tv_sec == self->tq_head->tv.tv_sec &&
       handle->tv.tv_usec <= self->tq_head->tv.tv_usec)) {
    handle->next = self->tq_head;
    self->tq_head = handle;
    handle->vrfy = NWMAGIC;
    return 1;
  } else {
    /* must be in between two so start at the first entry */
    pNWTimer pNxt = self->tq_head;
    /* follow chain until the one after this one is greater than new one */
    while (pNxt->next &&
           (handle->tv.tv_sec > pNxt->next->tv.tv_sec ||
            (handle->tv.tv_sec == pNxt->next->tv.tv_sec &&
             handle->tv.tv_usec > pNxt->next->tv.tv_usec)))
      pNxt = pNxt->next;
    /* slip new one in between this one and the next one */
    handle->next = pNxt->next;
    pNxt->next = handle;
    handle->vrfy = NWMAGIC;
    return 1;
  }
  return 0;
}

/*
 * \brief private function to remove an entry from the sorted timer queue.
 *
 * \param self singleton
 * \param handle existing timer to remove
 */
static int NetWatchTimerRemQue(pNetWatch self, pNWTimer handle)
{
  assert(handle->vrfy == NWMAGIC);
  /* handle the case of first and possibly only */
  if (handle == self->tq_head) {
    self->tq_head = self->tq_head->next;        /* may be NULL */
    if (handle == self->tq_tail)
      self->tq_tail = NULL;
  }
  /* handle general case */
  else {
    pNWTimer pNxt = self->tq_head;
    while (pNxt) {
      if (handle == pNxt->next) {
        pNxt->next = pNxt->next->next;
        break;
      }
      pNxt = pNxt->next;
    }
    /* It it was the last entry, point tail to its predecessor */
    if (handle == self->tq_tail)
      self->tq_tail = pNxt;
  }
  handle->vrfy = 0;
  return 1;
}

int NetWatchRegisterTimer(pNWTimer * handle, int mSec,
                          pNWCallback callback, void *context)
{
  assert(callback);
  pNetWatch self = NetWatchGetInstance();
  if (!self || self->lMagic != NWMAGIC)
    return 0;
  pNWTimer pNew = (pNWTimer) malloc(sizeof(NWTimer));
  if (pNew == NULL)
    return 0;
  memset(pNew, 0, sizeof(NWTimer));
  gettimeofday(&pNew->tv, NULL);
  pNew->tv.tv_sec += mSec / 1000;
  pNew->tv.tv_usec += 1000 * (mSec % 1000);
  if (pNew->tv.tv_usec > 1000000) {
    pNew->tv.tv_sec++;
    pNew->tv.tv_usec -= 1000000;
  }
  pNew->msec = mSec;
  pNew->tick = 0;
  pNew->func = callback;
  pNew->cntx = context;
  NetWatchTimerInsQue(self, pNew);
  *handle = pNew;
  return 1;
}

int NetWatchRegisterTimerPeriodic(pNWTimer * handle, int mSecInitial,
                                  int mSecPeriod, pNWCallback callback,
                                  void *context)
{
  assert(callback);
  if (NetWatchRegisterTimer(handle, mSecInitial, callback, context)) {
    pNWTimer pNew = *handle;
    if (pNew == NULL)
      return 0;
    if (mSecPeriod > 0)
      pNew->tick = mSecPeriod;
    return 1;
  }
  return 0;
}

pNWTimer NetWatchGetActiveTimer(void)
{
  return activeTimer;
}

int NetWatchGetTimerInitial(pNWTimer handle)
{
  if (handle == NULL
      || (handle->vrfy != NWMAGIC && handle->vrfy != ~NWMAGIC))
    return 0;
  return handle->msec;
}

int NetWatchGetTimerDelay(pNWTimer handle)
{
  return NetWatchGetTimerInitial(handle);
}

int NetWatchGetTimerPeriod(pNWTimer handle)
{
  if (handle == NULL
      || (handle->vrfy != NWMAGIC && handle->vrfy != ~NWMAGIC))
    return 0;
  return handle->tick;
}

int NetWatchSetTimerPeriod(pNWTimer handle, int mSecPeriod)
{
  if (handle == NULL
      || (handle->vrfy != NWMAGIC && handle->vrfy != ~NWMAGIC))
    return 0;
  handle->tick = mSecPeriod;
  return 1;
}

int NetWatchRemoveTimer(pNWTimer handle)
{
  pNetWatch self = NetWatchGetInstance();
  if (!self || self->lMagic != NWMAGIC)
    return 0;
  if (handle == NULL || handle->vrfy != NWMAGIC)
    return 0;
  NetWatchTimerRemQue(self, handle);
  handle->vrfy = 0;
  free(handle);
  return 1;
}

/* private data */
typedef struct __netwatchcontext {
  pNWContext next;              /* chain pointer */
  int sock;                     /* socket to watch */
  int mode;                     /* read or write */
  pNWCallback func;             /* user supplied callback function */
  void *cntx;                   /* user supplied callback context */
  long vrfy;                    /* integrity check */
} NWContext;

/**
 * \brief private function to insert entry into unsorted queue
 *
 * \param self singleton
 * \param handle entry to insert
 */
static int NetWatchContextInsQue(pNetWatch self, pNWContext handle)
{
  if (self->cq_head == NULL)    /* empty */
    self->cq_head = self->cq_tail = handle;
  else {
    self->cq_tail->next = handle;
    self->cq_tail = handle;
  }
  handle->vrfy = NWMAGIC;
  return 1;
}


/**
 * \brief private function to purge invalid entries
 *
 * \param self singleton
 */
static void NetWatchContextPrgQue(pNetWatch self)
{
  pNWContext pNxt = NULL;
  /* while the first entry is invalid remove it */
  while (self->cq_head && self->cq_head->sock < 0) {
    pNWContext tmp = NULL;
    tmp = self->cq_head;
    self->cq_head = self->cq_head->next;
    tmp->vrfy = 0;
    free(tmp);
  }
  pNxt = self->cq_head;
  while (pNxt && pNxt->next) {
    if (pNxt->next->sock < 0) {
      pNWContext tmp = NULL;
      tmp = pNxt->next;
      pNxt->next = pNxt->next->next;
      tmp->vrfy = 0;
      free(tmp);
      continue;
    }
    pNxt = pNxt->next;
  }
  /* if the queue is empty then pNxt is NULL else pNxt points to the tail */
  self->cq_tail = pNxt;
  self->nInvalid = 0;
  return;
}

int NetWatchRegisterCallback(pNWContext * handle, int iSocket,
                             pNWCallback callback, void *context)
{
  assert(callback);
  pNWContext pNew = NULL;
  pNetWatch self = NetWatchGetInstance();
  if (!self || self->lMagic != NWMAGIC)
    return 0;
  if (iSocket < 0 || iSocket > 65535)
    return 0;
  pNew = (pNWContext) malloc(sizeof(NWContext));
  if (pNew == NULL)
    return 0;
  memset(pNew, 0, sizeof(NWContext));
  pNew->sock = iSocket;
  pNew->mode = nwatch_read;
  pNew->func = callback;
  pNew->cntx = context;
  *handle = pNew;
  NetWatchContextInsQue(self, pNew);
  return 1;
}

int NetWatchRemoveCallback(pNWContext handle)
{
  pNetWatch self = NetWatchGetInstance();
  if (handle == NULL || handle->vrfy != NWMAGIC)
    return 0;
  if (!self || self->lMagic != NWMAGIC)
    return 0;
  /* mark as invalid */
  handle->sock = -1;
  /* increment count of invalid */
  self->nInvalid++;
  /* leave for garbage collection */
  return 1;
}

int NetWatchGetMode(pNWContext handle)
{
  if (handle == NULL || handle->vrfy != NWMAGIC)
    return 0;
  return handle->mode;
}

int NetWatchSetMode(pNWContext handle, int mode)
{
  if (handle == NULL || handle->vrfy != NWMAGIC)
    return 0;
  handle->mode = mode;
  return 1;
}

/**
 * \brief the task to drive all this
 * Should be called periodically
 */
int NetWatchTask(void *pData)
{
  pNetWatch self = NULL;
  pNWContext pNWC = NULL;
  fd_set rMask;
  fd_set wMask;
  struct timeval tmo = { 0, 0 };
  int iRet;
  int iCount;

  /* Check the singleton */
  self = NetWatchGetInstance();
  if (!self || self->lMagic != NWMAGIC)
    return 0;

  /* Purge the invalidated */
  if (self->nInvalid > 0)
    NetWatchContextPrgQue(self);

  /* build the select mask */
  FD_ZERO(&rMask);
  FD_ZERO(&wMask);
  pNWC = self->cq_head;
  iCount = -1;
  while (pNWC) {
    if (pNWC->sock >= 0 && pNWC->sock <= 65535) {
      if (pNWC->mode & nwatch_read)
        FD_SET(pNWC->sock, &rMask);
      if (pNWC->mode & nwatch_write)
        FD_SET(pNWC->sock, &wMask);
      if (pNWC->sock > iCount) {
        iCount = pNWC->sock;
      }
    }
    pNWC = pNWC->next;
  }

  iRet = 0;
  if (iCount >= 0)
    iRet = uselect(iCount + 1, &rMask, &wMask, NULL, &tmo);

  if (iRet > 0) {
    /* invoke the active callbacks */
    iCount = 0;
    pNWC = self->cq_head;
    while (pNWC) {
      if (pNWC->sock >= 0 && pNWC->sock <= 65535) {
        int action_mode = 0;
        if ((pNWC->mode & nwatch_read) && FD_ISSET(pNWC->sock, &rMask))
          action_mode |= nwatch_read;
        if ((pNWC->mode & nwatch_write) && FD_ISSET(pNWC->sock, &wMask))
          action_mode |= nwatch_write;
        if (action_mode != 0) {
          int iStatus;
          iStatus = (*pNWC->func) (pNWC->cntx, action_mode);
        }
      }
      pNWC = pNWC->next;
    }
  }

  /* Now do the timers */
  if (self->tq_head) {
    int iStatus;
    struct timeval tv;
    gettimeofday(&tv, NULL);
    while (self->tq_head) {
      pNWTimer pNew = self->tq_head;
      if (tv.tv_sec < pNew->tv.tv_sec ||
          (tv.tv_sec == pNew->tv.tv_sec &&
           tv.tv_usec < pNew->tv.tv_usec)) {
        break;
      }
      NetWatchTimerRemQue(self, pNew);
      activeTimer = pNew;
      activeTimer->vrfy = ~NWMAGIC;
      iStatus = pNew->func(pNew->cntx, 0);
      activeTimer->vrfy = 0;
      activeTimer = NULL;
      /*
       * If this is a recurrent timer and the function
       * indicates to keep it going, put it back in
       */
      if (pNew->tick && iStatus == 1) {
        /*
         * While the expiration time is in the past, increment
         */
        gettimeofday(&tv, NULL);
        while (tv.tv_sec > pNew->tv.tv_sec ||
               (tv.tv_sec == pNew->tv.tv_sec &&
                tv.tv_usec > pNew->tv.tv_usec)) {
          pNew->tv.tv_usec += 1000 * pNew->tick;
          if (pNew->tv.tv_usec > 1000000) {
            pNew->tv.tv_sec += pNew->tv.tv_usec / 1000000;
            pNew->tv.tv_usec %= 1000000;
          }
        }
        NetWatchTimerInsQue(self, pNew);
      } else {
        pNew->vrfy = 0;
        free(pNew);
      }
    }
  }

  /* done, finally */
  return 1;
}