sics/multichan.c

/*
 * M U L T I C H A N
 *
 * This module manages communications on a multi-channel controller.
 *
 * A multi-channel controller is one where multiple channels or units are on a
 * single controller and share a single command channel.
 *
 * Douglas Clowes, February 2007
 *
 */

#include <sys/time.h>
#include <arpa/inet.h>
#include <netinet/tcp.h>
#include <netdb.h>
#include <sics.h>
#include <rs232controller.h>
#include "network.h"
#include "multichan.h"
#include "nwatch.h"

typedef struct __MultiChanController MultiChanController, *pMultiChanController;

typedef struct __mcc_command MC_Cmd, *pMC_Cmd;

struct __mcc_command {
  pMC_Cmd   next;
  void*     cntx;
  MCC_Transmit tx;
  MCC_Receive rx;
  pMultiChan unit;
  int timeout;
  int retries;
};

struct __MultiChan {
  pMultiChan next;
  pMultiChanController mcc;
  MCC_Notify notify_func;
  void* notify_cntx;
};

struct __MultiChanController {
  pObjectDescriptor pDes;
  char*         mcc_name;
  char*         pHost;
  int           iPort;
  int           iDelay;       /* intercommand delay in milliseconds */
  int           timeout;
  int           retries;
  struct timeval tvLastCmd;   /* time of completion of last command */
  int           unit_count;   /* number of units connected */
  pMultiChan    units;        /* head of unit chain */
  prs232        rs232c;       /* associated RS232 controller object */
  pMC_Cmd       command_head; /* first/next command in queue */
  pMC_Cmd       command_tail; /* last command in queue */
  pNWContext    nw_ctx;       /* NetWait context handle */
  pNWTimer      nw_tmr;       /* NetWait timer handle */
  mkChannel*    pSock;        /* socket address */
};

static pMultiChanController mcc_array[FD_SETSIZE];
static int mcc_index = 0;

/* ---------------------------- Local ------------------------------------
     CreateSocketAdress stolen from Tcl. Thanks to John Ousterhout
*/

static int
CreateSocketAdress(
    struct sockaddr_in *sockaddrPtr,	/* Socket address */
    char *host,				/* Host.  NULL implies INADDR_ANY */
    int port)				/* Port number */
{
    struct hostent *hostent;		/* Host database entry */
    struct in_addr addr;		/* For 64/32 bit madness */

    (void) memset((char *) sockaddrPtr, '\0', sizeof(struct sockaddr_in));
    sockaddrPtr->sin_family = AF_INET;
    sockaddrPtr->sin_port = htons((unsigned short) (port & 0xFFFF));
    if (host == NULL) {
	addr.s_addr = INADDR_ANY;
    } else {
	hostent = gethostbyname(host);
	if (hostent != NULL) {
	    memcpy((char *) &addr,
		(char *) hostent->h_addr_list[0], (size_t) hostent->h_length);
	} else {
	    addr.s_addr = inet_addr(host);
	    if (addr.s_addr == (unsigned long)-1) {
		return 0;	/* error */
	    }
	}
    }
    /*
     * There is a rumor that this assignment may require care on
     * some 64 bit machines.
     */

    sockaddrPtr->sin_addr.s_addr = addr.s_addr;
    return 1;
}

static void MC_Notify(pMultiChanController self, int event)
{
  pMultiChan unit;
  for (unit = self->units; unit; unit = unit->next)
    if (unit->notify_func != NULL)
      unit->notify_func(unit->notify_cntx, event);
}

static int MC_Reconnect(pMultiChanController self)
{
  int iRet;
  int sock;
  int flag = 1;
  char line[132];

  iRet = NETReconnect(self->pSock);
  if (iRet <= 0) {
    snprintf(line, 132, "Disconnect on MultiChan '%s'", self->mcc_name);
    SICSLogWrite(line, eStatus);
    MC_Notify(self, MCC_DISCONNECT);
    return iRet;
  }
  snprintf(line, 132, "Reconnect on MultiChan '%s'", self->mcc_name);
  SICSLogWrite(line, eStatus);
  MC_Notify(self, MCC_RECONNECT);
  return 1;
}


static int CommandTimeout(void* cntx, int mode);

static int StartCommand(pMultiChanController self)
{
  pMC_Cmd myCmd = self->command_head;
  mkChannel* sock = self->pSock;

  if (myCmd == NULL)
    return OKOK;

  /*
   * Remove any old command timeout timer
   */
  if (self->nw_tmr)
    NetWatchRemoveTimer(self->nw_tmr);

  /*
   * Implement the inter-command delay
   */
  if (self->iDelay) {
    struct timeval now, when;
    gettimeofday(&now, NULL);
    if (self->tvLastCmd.tv_sec == 0)
      self->tvLastCmd = now;
    when.tv_sec = self->tvLastCmd.tv_sec;
    when.tv_usec = self->tvLastCmd.tv_usec + 1000 * self->iDelay;
    if (when.tv_usec >= 1000000) {
      when.tv_sec += when.tv_usec / 1000000;
      when.tv_usec %= 1000000;
    }
    if (when.tv_sec > now.tv_sec ||
        (when.tv_sec == now.tv_sec && when.tv_usec > now.tv_usec)) {
      int delay = when.tv_sec - now.tv_sec;
      delay *= 1000;
      delay += (when.tv_usec - now.tv_usec + (1000 - 1)) / 1000;
      NetWatchRegisterTimer(&self->nw_tmr, delay,
          CommandTimeout, self);
      return OKOK;
    }
  }

  /*
   * Discard any input before sending command
   */
  while (NETAvailable(sock, 0)) {
    char reply[1];
    int iRet;
    iRet = NETRead(sock, reply, 1, 0);
    if (iRet < 0) { /* TODO: EOF */
      iRet = MC_Reconnect(self);
      if (iRet == 0)
        return 0;
    }
  }
  /*
   * Add a new command timeout timer
   */
  if (myCmd->timeout > 0)
    NetWatchRegisterTimer(&self->nw_tmr, myCmd->timeout,
        CommandTimeout, self);
  else
    NetWatchRegisterTimer(&self->nw_tmr, 30000,
        CommandTimeout, self);
  return myCmd->tx(myCmd->cntx);
}

static int QueCommand(pMultiChanController self, pMC_Cmd cmd)
{
  /*
   * If the command queue is empty, start transmission
   */
  if (self->command_head == NULL) {
    self->command_head = self->command_tail = cmd;
    StartCommand(self);
  } else {
    self->command_tail->next = cmd;
    self->command_tail = cmd;
  }
  return 1;
}

static int PopCommand(pMultiChanController self)
{
  pMC_Cmd myCmd = self->command_head;
  if (self->nw_tmr)
    NetWatchRemoveTimer(self->nw_tmr);
  self->nw_tmr = 0;
  gettimeofday(&self->tvLastCmd, NULL);
  /*
   * If this is not the last in queue, start transmission
   */
  if (myCmd->next) {
    pMC_Cmd pNew = myCmd->next;
    self->command_head = pNew;
    StartCommand(self);
  }
  else
    self->command_head = self->command_tail = NULL;
  free(myCmd);
  return 1;
}

static int CommandTimeout(void* cntx, int mode)
{
  pMultiChanController self = (pMultiChanController) cntx;
  pMC_Cmd myCmd = self->command_head;
  self->nw_tmr = 0;
  if (myCmd->retries > 0) {
    --myCmd->retries;
    StartCommand(self);
  }
  else {
    int iRet;
    iRet = myCmd->rx(myCmd->cntx, MCC_TIMEOUT);
    if (iRet == MCC_POP_CMD)
      PopCommand(self); /* remove command */
    else if (iRet == MCC_RETRY_CMD)
      StartCommand(self); /* restart command */
    else if (iRet == MCC_RECONNECT)
      MC_Reconnect(self);
  }
  return 1;
}

static int MyCallback(void* context, int mode)
{
  pMultiChanController self = (pMultiChanController) context;

  if (mode & nwatch_read) {
    int iRet;
    char reply[1];

    iRet = NETRead(self->pSock, reply, 1, 0);
    if (iRet < 0) { /* TODO: EOF */
      iRet = MC_Reconnect(self);
      if (iRet <= 0)
        return iRet;
      /* restart the command */
      StartCommand(self);
      return 1;
    }
    if (iRet == 0) { /* TODO: timeout or error */
      return 0;
    } else {
      pMC_Cmd myCmd = self->command_head;
      if (myCmd) {
        iRet = myCmd->rx(myCmd->cntx, reply[0]);
        if (iRet == MCC_POP_CMD) /* end of command */
          PopCommand(self);
        else if (iRet == 0) /* end of command */
          PopCommand(self);
        else if (iRet < 0) /* TODO: error */
          ;
      }
    }
  }
  return 1;
}

int MultiChanReconnect(pMultiChan unit)
{
  int iRet;
  assert(unit);
  assert(unit->mcc);
  iRet = MC_Reconnect(unit->mcc);
  /* TODO: handle in-progress */
  return iRet;
}

mkChannel* MultiChanGetSocket(pMultiChan unit)
{
  assert(unit);
  assert(unit->mcc);
  return unit->mcc->pSock;
}

int MultiChanEnque(pMultiChan unit, void* context, MCC_Transmit tx, MCC_Receive rx)
{
  pMC_Cmd myCmd = NULL;

  myCmd = (pMC_Cmd) malloc(sizeof(MC_Cmd));
  /* TODO: check malloc */
  memset(myCmd, 0, sizeof(MC_Cmd));
  myCmd->cntx = context;
  myCmd->tx = tx;
  myCmd->rx = rx;
  myCmd->unit = unit;
  myCmd->timeout = unit->mcc->timeout;
  myCmd->retries = unit->mcc->retries;
  return QueCommand(unit->mcc, myCmd);
}

int MultiChanWrite(pMultiChan unit, void* buffer, int buflen)
{
  int iRet;
  mkChannel* sock;
  assert(unit);
  assert(unit->mcc);
  sock = MultiChanGetSocket(unit);
  iRet = NETWrite(sock, buffer, buflen);
  /* TODO handle errors */
  if (iRet < 0) { /* TODO: EOF */
    iRet = MC_Reconnect(unit->mcc);
    if (iRet == 0)
      return 0;
  }
  return 1;
}

void MultiChanSetNotify(pMultiChan unit, void* context, MCC_Notify notify)
{
  assert(unit);
  unit->notify_func = notify;
  unit->notify_cntx = context;
}

int MultiChanGetDelay(pMultiChan unit)
{
  assert(unit);
  return unit->mcc->iDelay;
}

int MultiChanSetDelay(pMultiChan unit, int iDelay)
{
  int old_iDelay;
  assert(unit);
  old_iDelay = unit->mcc->iDelay;
  unit->mcc->iDelay = iDelay;
  return old_iDelay;
}

int MultiChanGetTimeout(pMultiChan unit)
{
  assert(unit);
  return unit->mcc->timeout;
}

int MultiChanSetTimeout(pMultiChan unit, int timeout)
{
  int old_timeout;
  assert(unit);
  old_timeout = unit->mcc->timeout;
  unit->mcc->timeout = timeout;
  return old_timeout;
}

int MultiChanGetRetries(pMultiChan unit)
{
  assert(unit);
  return unit->mcc->retries;
}

int MultiChanSetRetries(pMultiChan unit, int retries)
{
  int old_retries;
  assert(unit);
  old_retries = unit->mcc->retries;
  unit->mcc->retries = retries;
  return old_retries;
}

int MultiChanAction(SConnection *pCon, SicsInterp *pSics,
                        void *pData, int argc, char *argv[])
{
  char line[132];
  pMultiChanController self = (pMultiChanController) pData;
  if (argc > 1) {
    if (strcasecmp(argv[1], "reconnect") == 0) {
      MC_Reconnect(self);
      return OKOK;
    }
    if (strcasecmp(argv[1], "delay") == 0) {
      if (argc > 2) {
        int delay;
        int iRet;
        iRet = sscanf(argv[2], "%d", &delay);
        if (iRet != 1) {
          snprintf(line, 132, "Invalid argument: %s", argv[2]);
          SCWrite(pCon, line, eError);
          return 0;
        }
        else {
          if (delay < 0 || delay > 30000) {
            snprintf(line, 132, "Value out of range: %d", delay);
            SCWrite(pCon, line, eError);
            return 0;
          }
          self->iDelay = delay;
          return OKOK;
        }
      }
      else {
        snprintf(line, 132, "%s.delay = %d", argv[0], self->iDelay);
        SCWrite(pCon, line, eStatus);
        return OKOK;
      }
      return OKOK;
    }
    if (strcasecmp(argv[1], "timeout") == 0) {
      if (argc > 2) {
        int timeout;
        int iRet;
        iRet = sscanf(argv[2], "%d", &timeout);
        if (iRet != 1) {
          snprintf(line, 132, "Invalid argument: %s", argv[2]);
          SCWrite(pCon, line, eError);
          return 0;
        }
        else {
          if (timeout < 0 || timeout > 30000) {
            snprintf(line, 132, "Value out of range: %d", timeout);
            SCWrite(pCon, line, eError);
            return 0;
          }
          self->timeout = timeout;
          return OKOK;
        }
      }
      else {
        snprintf(line, 132, "%s.timeout = %d", argv[0], self->timeout);
        SCWrite(pCon, line, eStatus);
        return OKOK;
      }
      return OKOK;
    }
    if (strcasecmp(argv[1], "retries") == 0) {
      if (argc > 2) {
        int retries;
        int iRet;
        iRet = sscanf(argv[2], "%d", &retries);
        if (iRet != 1) {
          snprintf(line, 132, "Invalid argument: %s", argv[2]);
          SCWrite(pCon, line, eError);
          return 0;
        }
        else {
          if (retries < 0 || retries > 30000) {
            snprintf(line, 132, "Value out of range: %d", retries);
            SCWrite(pCon, line, eError);
            return 0;
          }
          self->retries = retries;
          return OKOK;
        }
      }
      else {
        snprintf(line, 132, "%s.retries = %d", argv[0], self->retries);
        SCWrite(pCon, line, eStatus);
        return OKOK;
      }
      return OKOK;
    }
  }
  snprintf(line, 132, "%s does not understand %s", argv[0], argv[1]);
  SCWrite(pCon, line, eError);
  return 0;
}

static pMultiChanController MC_Create(const char* host, const char* port)
{
  int i;
  pMultiChanController self = NULL;
  prs232 ctlr = NULL;
  mkChannel* channel = NULL;

  if (host == NULL)
    return NULL;

  /* try the MCC with this name */
  self = (pMultiChanController) FindCommandData(pServ->pSics, host, "MultiChan");

  /* try the RS232 Controller with this name */
  if (self == NULL && port == NULL) {
    ctlr = (prs232) FindCommandData(pServ->pSics, host, "RS232 Controller");
    if(ctlr == NULL) {
      char line[132];
      snprintf(line, 132, "ERROR: RS232 Controller '%s' is not found in MultiChan", host);
      SICSLogWrite(line, eError);
      return NULL;
    }
    channel = ctlr->pSock;
    if (channel == NULL) {
      char line[132];
      snprintf(line, 132, "ERROR: RS232 Controller '%s' is not connected in MultiChan", host);
      SICSLogWrite(line, eError);
      return NULL;
    }
    /* look for mcc with same controller */
    for (i = 0; i < mcc_index; ++i)
      if (mcc_array[i]->rs232c == ctlr) {
        self = mcc_array[i];
        break;
      }
  }

  /* try host and port */
  if (self == NULL && ctlr == NULL) {
    int port_no = atoi(port);
    if (port_no == 0) {
      struct servent *sp=NULL;
      sp = getservbyname(port, NULL);
      if (sp)
        port_no = ntohs(sp->s_port);
    }
    if (port_no > 0) {
      struct sockaddr_in sa;
      if (CreateSocketAdress(&sa, host, port_no)) {
        /* look for mcc with same address */
        for (i = 0; i < mcc_index; ++i)
          if (mcc_array[i]->pSock->adresse.sin_port == sa.sin_port
              && mcc_array[i]->pSock->adresse.sin_addr.s_addr == sa.sin_addr.s_addr) {
            self = mcc_array[i];
            break;
          }
      }
      if (self == NULL) {
        channel = NETConnectWithFlags(host, port_no, 0);
        /* TODO */
      }
    }
  }

  if (self == NULL) {
    if (ctlr == NULL && channel == NULL)
      return NULL;

    self = (pMultiChanController) malloc(sizeof(MultiChanController));
    if (self == NULL)
      return NULL;
    memset(self, 0, sizeof(MultiChanController));
    self->rs232c = ctlr;
    self->pSock = channel;
    self->pDes = CreateDescriptor("MultiChan");
    mcc_array[mcc_index++] = self;
  }

  for (i = 0; i < mcc_index; ++i)
    if (mcc_array[i] == self) {
      break;
    }
  if (i == mcc_index)
    mcc_array[mcc_index++] = self;

  return self;
}

static int MC_Init(pMultiChanController self)
{
  /* TODO: Init the controller */
  if (self->nw_ctx == NULL)
    NetWatchRegisterCallback(&self->nw_ctx,
        self->pSock->sockid,
        MyCallback,
        self);
  return 1;
}

static void MC_Kill(void* pData)
{
  int i;
  pMultiChanController self = (pMultiChanController) pData;
  for (i = 0; i < mcc_index; ++i)
    if (mcc_array[i] == self) {
      --mcc_index;
      if (mcc_index > 0)
        mcc_array[i] = mcc_array[mcc_index];
      if (self->nw_ctx)
        NetWatchRemoveCallback(self->nw_ctx);
      if (self->nw_tmr)
        NetWatchRemoveTimer(self->nw_tmr);
      if (self->mcc_name)
        free(self->mcc_name);
      if (self->rs232c == NULL) {
        NETClosePort(self->pSock);
        free(self->pSock);
      }
      free(self);
      return;
    }
}

/*
 * \brief make a multichannel controller from the command line
 */
int MultiChanFactory(SConnection *pCon, SicsInterp *pSics,
    void *pData, int argc, char *argv[])
{
  pMultiChanController pNew = NULL;
  int iRet, status;
  char pError[256];

  if(argc < 3)
  {
    SCWrite(pCon,"ERROR: insufficient no of arguments to MultiChanFactory",
            eError);
    return 0;
  }

  /* try to find an existing mcc with this name */
  pNew = (pMultiChanController) FindCommandData(pServ->pSics, argv[1], "MultiChan");
  if (pNew != NULL) {
    char line[132];
    snprintf(line, 132, "ERROR: MultiChan '%s' already exists", argv[1]);
    SCWrite(pCon, line, eError);
    return 1;
  }

  /*
   create data structure and open port 
  */
  if (argc > 3)
    pNew = MC_Create(argv[2], argv[3]);
  else
    pNew = MC_Create(argv[2], NULL) ;
  
  if(!pNew)
  {
    SCWrite(pCon,"ERROR: failed to create MultiChan in MultiChanFactory",eError);
    return 0;
  }
  if (pNew->mcc_name)
    free(pNew->mcc_name);
  pNew->mcc_name = strdup(argv[1]);

  status = MC_Init(pNew);
  if(status != 1)
  {
     sprintf(pError,"ERROR: failed to connect to %s at port %d",
                    pNew->pHost, pNew->iPort);
     SCWrite(pCon,pError,eError);
  }
  
  /*
    create the command
  */
  iRet = AddCommand(pSics, argv[1], MultiChanAction, MC_Kill, pNew);
  if(!iRet)
  {
        sprintf(pError,"ERROR: duplicate command %s not created", argv[1]);
        SCWrite(pCon,pError,eError);
        MC_Kill(pNew);
        return 0;
 }
 return 1;
}

/*
 * \brief make a multichannel controller from a named rs232 controller
 *
 * \param name the name of the SICS "RS232 Controller" object
 * \param handle the handle to the multichannel controller object
 * \return 0 for FAILURE, 1 for SUCCESS
 */
int MultiChanCreateHost(const char* host, const char* port, pMultiChan* handle)
{
  int status;
  pMultiChanController self = NULL;
  pMultiChan unit = NULL;

  *handle = NULL;

  self = MC_Create(host, port);
  if (self == NULL)
    return 0;
  status = MC_Init(self);

  unit = (pMultiChan) malloc(sizeof(MultiChan));
  /* TODO: check malloc failure */
  memset(unit, 0, sizeof(MultiChan));
  ++self->unit_count;
  unit->mcc = self;
  unit->next = self->units;
  self->units = unit;
  *handle = unit;
  return 1;
}

int MultiChanCreate(const char* host, pMultiChan* handle) {
  return MultiChanCreateHost(host, NULL, handle);
}

int MultiChanDestroy(pMultiChan unit)
{
  assert(unit);
  assert(unit->mcc);
  pMultiChanController self = unit->mcc;
  pMultiChan* pNxt = &self->units;
  while (*pNxt) {
    if (*pNxt == unit) {
      *pNxt = (*pNxt)->next;
      break;
    }
    pNxt = &(*pNxt)->next;
  }
  --self->unit_count;
  if (self->unit_count <= 0) {
    MC_Kill(self);
  }
  free(unit);
  return 1;
}