linse/sics
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

timeouts, retries and reconnection handling improvements

Browse Source 
r1660 | dcl | 2007-03-16 09:03:42 +1100 (Fri, 16 Mar 2007) | 2 lines
This commit is contained in:
Douglas Clowes
2007-03-16 09:03:42 +11:00
parent 3b7496b7ce
commit d1500ef0b4
2 changed files with 161 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

208
multichan.c
View File

@@ -28,6 +28,8 @@ struct __mcc_command {
MCC_Transmit tx;
MCC_Receive rx;
pMultiChan unit;
int timeout;
int retries;
};
struct __MultiChan {
@@ -35,14 +37,16 @@ struct __MultiChan {
pMultiChanController mcc;
MCC_Notify notify_func;
void* notify_cntx;
int timeout;
};
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 */
@@ -61,7 +65,7 @@ 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, event);
unit->notify_func(unit->notify_cntx, event);
}
static int MC_Reconnect(pMultiChanController self)
@@ -69,18 +73,17 @@ static int MC_Reconnect(pMultiChanController self)
int iRet;
int sock;
int flag = 1;
char line[132];
iRet = NETReconnect(self->controller->pSock);
if (iRet <= 0) {
snprintf(line, 132, "Disconnect on MultiChan '%s'", self->mcc_name);
SICSLogWrite(line, eStatus);
MC_Notify(self, MCC_DISCONNECT);
return iRet;
}
sock = self->controller->pSock->sockid;
iRet = setsockopt(sock, /* socket */
IPPROTO_TCP, /* set option at TCP level */
TCP_NODELAY, /* name of option */
(char *) &flag, /* the cast is historical cruft */
sizeof(int)); /* length of option value */
snprintf(line, 132, "Reconnect on MultiChan '%s'", self->mcc_name);
SICSLogWrite(line, eStatus);
MC_Notify(self, MCC_RECONNECT);
return 1;
}
@@ -93,6 +96,9 @@ static int StartCommand(pMultiChanController self)
pMC_Cmd myCmd = self->command_head;
mkChannel* sock = self->controller->pSock;
if (myCmd == NULL)
return OKOK;
/*
* Remove any old command timeout timer
*/
@@ -140,23 +146,15 @@ static int StartCommand(pMultiChanController self)
/*
* Add a new command timeout timer
*/
if (myCmd->unit->timeout > 0)
NetWatchRegisterTimer(&self->nw_tmr, myCmd->unit->timeout,
if (myCmd->timeout > 0)
NetWatchRegisterTimer(&self->nw_tmr, myCmd->timeout,
CommandTimeout, self);
else
NetWatchRegisterTimer(&self->nw_tmr, 1000,
NetWatchRegisterTimer(&self->nw_tmr, 30000,
CommandTimeout, self);
return myCmd->tx(myCmd->cntx);
}
static int CommandTimeout(void* cntx, int mode)
{
pMultiChanController self = (pMultiChanController) cntx;
self->nw_tmr = 0;
StartCommand(self);
return 1;
}
static int QueCommand(pMultiChanController self, pMC_Cmd cmd)
{
/*
@@ -193,6 +191,28 @@ static int PopCommand(pMultiChanController self)
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;
@@ -216,10 +236,12 @@ static int MyCallback(void* context, int mode)
pMC_Cmd myCmd = self->command_head;
if (myCmd) {
iRet = myCmd->rx(myCmd->cntx, reply[0]);
if (iRet < 0) /* TODO: error */
;
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 */
;
}
}
}
@@ -254,6 +276,8 @@ int MultiChanEnque(pMultiChan unit, void* context, MCC_Transmit tx, MCC_Receive
myCmd->tx = tx;
myCmd->rx = rx;
myCmd->unit = unit;
myCmd->timeout = unit->mcc->timeout;
myCmd->retries = unit->mcc->retries;
return QueCommand(unit->mcc, myCmd);
}
@@ -299,56 +323,131 @@ int MultiChanSetDelay(pMultiChan unit, int iDelay)
int MultiChanGetTimeout(pMultiChan unit)
{
assert(unit);
return unit->timeout;
return unit->mcc->timeout;
}
int MultiChanSetTimeout(pMultiChan unit, int timeout)
{
int old_timeout;
assert(unit);
old_timeout = unit->timeout;
unit->timeout = timeout;
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 (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);
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;
}
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, int port)
static pMultiChanController MC_Create(const char* host)
{
int i;
pMultiChanController self = NULL;
@@ -410,6 +509,8 @@ static void MC_Kill(void* pData)
NetWatchRemoveCallback(self->nw_ctx);
if (self->nw_tmr)
NetWatchRemoveTimer(self->nw_tmr);
if (self->mcc_name)
free(self->mcc_name);
free(self);
return;
}
@@ -422,7 +523,7 @@ int MultiChanFactory(SConnection *pCon, SicsInterp *pSics,
int iRet, status;
char pError[256];
if(argc < 4)
if(argc < 3)
{
SCWrite(pCon,"ERROR: insufficient no of arguments to MultiChanFactory",
eError);
@@ -432,13 +533,16 @@ int MultiChanFactory(SConnection *pCon, SicsInterp *pSics,
/*
create data structure and open port
*/
pNew = MC_Create(argv[2], atoi(argv[3]));
pNew = MC_Create(argv[2]);
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)
@@ -470,7 +574,7 @@ int MultiChanCreate(const char* name, pMultiChan* handle)
*handle = NULL;
self = MC_Create(name, 0);
self = MC_Create(name);
if (self == NULL)
return 0;
status = MC_Init(self);

7
multichan.h
View File

@@ -12,8 +12,11 @@
#ifndef SICSMULTICHAN
#define SICSMULTICHAN
#define MCC_DISCONNECT 101
#define MCC_RECONNECT 102
#define MCC_TIMEOUT -1
#define MCC_DISCONNECT -2
#define MCC_RECONNECT -3
#define MCC_RETRY_CMD -4
#define MCC_POP_CMD -5
typedef struct __MultiChan MultiChan, *pMultiChan;