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

add some more checks, defer context deletion, don't double scan

Browse Source 
r1784 | dcl | 2007-04-02 17:35:44 +1000 (Mon, 02 Apr 2007) | 2 lines
This commit is contained in:
Douglas Clowes
2007-04-02 17:35:44 +10:00
parent 62aa98edf7
commit 105cc5650c
1 changed files with 128 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

219
nwatch.c
View File

@@ -26,6 +26,7 @@
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 */
@@ -228,7 +229,6 @@ typedef struct __netwatchcontext {
pNWContext next; /* chain pointer */
int sock; /* socket to watch */
int mode; /* read or write */
int actn; /* action being invoked on this pass */
pNWCallback func; /* user supplied callback function */
void* cntx; /* user supplied callback context */
long vrfy; /* integrity check */
@@ -257,7 +257,7 @@ static int NetWatchContextInsQue(pNetWatch self, pNWContext handle)
* \param self singleton
* \param handle entry to insert
*/
static int NetWatchContextRemQue(pNetWatch self, pNWContext handle)
static void NetWatchContextRemQue(pNetWatch self, pNWContext handle)
{
if (handle == self->cq_head) { /* if first */
self->cq_head = self->cq_head->next;
@@ -279,6 +279,40 @@ static int NetWatchContextRemQue(pNetWatch self, pNWContext handle)
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) {
if (pNxt->next && pNxt->next->sock < 0) {
pNWContext tmp = NULL;
tmp = pNxt->next;
pNxt->next = pNxt->next->next;
tmp->vrfy = 0;
free(tmp);
}
pNxt = pNxt->next;
}
/* if the queue is empty clear the tail */
if (self->cq_head == NULL)
self->cq_tail = pNxt;
self->nInvalid = 0;
return;
}
int NetWatchRegisterCallback(pNWContext* handle, int iSocket,
pNWCallback callback, void* context)
{
@@ -286,6 +320,8 @@ int NetWatchRegisterCallback(pNWContext* handle, int iSocket,
pNetWatch self = instance;
if(!self || self->lMagic != NWMAGIC)
return 0;
if (iSocket < 0 || iSocket > 65535)
return 0;
pNew = (pNWContext) malloc(sizeof(NWContext));
if (pNew == NULL)
return 0;
@@ -303,9 +339,12 @@ int NetWatchRegisterCallback(pNWContext* handle, int iSocket,
int NetWatchRemoveCallback(pNWContext handle)
{
pNetWatch self = instance;
if (handle == NULL || handle->vrfy != NWMAGIC)
return 0;
if(!self || self->lMagic != NWMAGIC)
return 0;
NetWatchContextRemQue(self, handle);
handle->sock = -1;
self->nInvalid++;
return 1;
}
@@ -329,26 +368,30 @@ int NetWatchSetMode(pNWContext handle, int mode)
*/
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;
pNWContext action[FD_SETSIZE];
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 = (pNetWatch) instance;
if(!self || self->lMagic != NWMAGIC)
return 0;
/* build the select mask */
FD_ZERO(&rMask);
FD_ZERO(&wMask);
pNWC = self->cq_head;
iCount = -1;
while(pNWC) {
/* Check the singleton */
self = (pNetWatch) instance;
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)
@@ -356,80 +399,74 @@ int NetWatchTask (void* pData)
if(pNWC->sock > iCount) {
iCount = pNWC->sock;
}
}
pNWC = pNWC->next;
}
iRet = 0;
if (iCount >= 0)
iRet = select(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;
}
}
iRet = 0;
if (iCount >= 0)
iRet = select(iCount+1, &rMask, &wMask, NULL, &tmo);
if(iRet > 0) {
/* now go through all registered things and note the active callbacks */
iCount = 0;
pNWC = self->cq_head;
while(pNWC)
{
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;
pNWC->actn = action_mode;
if (action_mode != 0)
action[iCount++] = pNWC;
pNWC = pNWC->next;
}
/* now process the callbacks */
while (iCount > 0) {
pNWC = action[--iCount];
if (pNWC->vrfy == NWMAGIC) {
int iStatus;
iStatus = (*pNWC->func)(pNWC->cntx, pNWC->actn);
pNWC->actn = 0;
}
}
}
/* 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 ||
/* 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);
iStatus = pNew->func(pNew->cntx, 0);
/*
* 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
free(pNew);
tv.tv_usec < pNew->tv.tv_usec)) {
break;
}
NetWatchTimerRemQue(self, pNew);
iStatus = pNew->func(pNew->cntx, 0);
/*
* 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
free(pNew);
}
}
/* done, finally */
return 1;
/* done, finally */
return 1;
}