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

This is the first working version of the new logging system. Some work

Browse Source 
in fine tuning still needs to be done. But is reasonably OK now.
This commit is contained in:
koennecke
2016-02-11 13:40:31 +01:00
parent b1cda3f579
commit 66466c1c0f
56 changed files with 420 additions and 1845 deletions
Download Patch File Download Diff File Expand all files Collapse all files
asyncqueue.c
+35 -54
View File
@@ -207,7 +207,7 @@ static void AQ_Notify(pAsyncQueue self, int event)
{
pAsyncUnit unit;
if (self->state != eAsyncConnected)
SICSLogPrintf(eStatus, "Function: %s:%s\n", self->queue_name,
Log(DEBUG,"ASQUIO", "Function:%s:%s", self->queue_name,
__func__);
for (unit = self->units; unit; unit = unit->next)
if (unit->notify_func != NULL)
@@ -222,7 +222,7 @@ static int TimedReconnect(void *cntx, int mode)
self->nw_tmr = 0;
if (self->state != eAsyncConnected)
SICSLogPrintf(eStatus, "Function: %s:%s\n", self->queue_name,
Log(DEBUG,"ASQUIO", "Function: %s:%s\n", self->queue_name,
__func__);
AQ_Purge(self);
@@ -240,7 +240,7 @@ static int TimedReconnect(void *cntx, int mode)
if (iRet < 0) {
snprintf(line, 132, "Failed reconnect on AsyncQueue '%s'",
self->queue_name);
SICSLogWrite(line, eStatus);
Log(DEBUG,"ASQUIO","%s",line);
/* Timer for retry */
NetWatchSetMode(self->nw_ctx, 0);
/* implement an exponential backoff within limits */
@@ -251,12 +251,12 @@ static int TimedReconnect(void *cntx, int mode)
self->retryTimer = 16000;
AQ_SetTimer(self, self->retryTimer,
TimedReconnect, self);
SICSLogPrintf(eStatus, "In %s:%s: state %s => eAsyncWaiting\n",
Log(DEBUG,"ASQUIO", "In %s:%s: state %s => eAsyncWaiting",
self->queue_name, __func__, state_name(self->state));
self->state = eAsyncWaiting;
} else {
NetWatchSetMode(self->nw_ctx, nwatch_write);
SICSLogPrintf(eStatus, "In %s:%s: state %s => eAsyncConnecting\n",
Log(DEBUG,"ASQUIO", "In %s:%s: state %s => eAsyncConnecting\n",
self->queue_name, __func__, state_name(self->state));
self->state = eAsyncConnecting;
/* await reconnect result in MyCallback */
@@ -264,11 +264,11 @@ static int TimedReconnect(void *cntx, int mode)
return 1;
}
NetWatchSetMode(self->nw_ctx, nwatch_read);
SICSLogPrintf(eStatus, "In %s:%s: state %s => eAsyncConnected\n",
Log(DEBUG,"ASQUIO", "In %s:%s: state %s => eAsyncConnected\n",
self->queue_name, __func__, state_name(self->state));
self->state = eAsyncConnected;
snprintf(line, 132, "Reconnect on AsyncQueue '%s'", self->queue_name);
SICSLogWrite(line, eStatus);
Log(DEBUG,"ASQUIO","%s",line);
AQ_Purge(self);
AQ_Notify(self, AQU_RECONNECT);
return 1;
@@ -280,7 +280,7 @@ static int AQ_Reconnect(pAsyncQueue self)
char line[132];
if (self->state != eAsyncConnected)
SICSLogPrintf(eStatus, "Function: %s:%s\n", self->queue_name,
Log(DEBUG,"ASQUIO", "Function: %s:%s\n", self->queue_name,
__func__);
/*
* Remove any old timer
@@ -290,7 +290,7 @@ static int AQ_Reconnect(pAsyncQueue self)
if (self->state == eAsyncConnected) {
self->state = eAsyncIdle;
SICSLogPrintf(eStatus, "Disconnect on AsyncQueue '%s'", self->queue_name);
Log(DEBUG,"ASQUIO", "Disconnect on AsyncQueue '%s'", self->queue_name);
AQ_Notify(self, AQU_DISCONNECT);
AQ_Purge(self);
}
@@ -310,12 +310,12 @@ static int AQ_Reconnect(pAsyncQueue self)
self->retryTimer = 125; /* initial delay */
AQ_SetTimer(self, self->retryTimer,
TimedReconnect, self);
SICSLogPrintf(eStatus, "In %s:%s: state %s => eAsyncWaiting\n",
Log(DEBUG,"ASQUIO", "In %s:%s: state %s => eAsyncWaiting\n",
self->queue_name, __func__, state_name(self->state));
self->state = eAsyncWaiting;
} else {
NetWatchSetMode(self->nw_ctx, nwatch_write);
SICSLogPrintf(eStatus, "In %s:%s: state %s => eAsyncConnecting\n",
Log(DEBUG,"ASQUIO", "In %s:%s: state %s => eAsyncConnecting\n",
self->queue_name, __func__, state_name(self->state));
self->state = eAsyncConnecting;
/* await reconnect result in MyCallback */
@@ -323,11 +323,11 @@ static int AQ_Reconnect(pAsyncQueue self)
return iRet;
}
NetWatchSetMode(self->nw_ctx, nwatch_read);
SICSLogPrintf(eStatus, "In %s:%s: state %s => eAsyncConnected\n",
Log(DEBUG,"ASQUIO", "In %s:%s: state %s => eAsyncConnected\n",
self->queue_name, __func__, state_name(self->state));
self->state = eAsyncConnected;
snprintf(line, 132, "Reconnect on AsyncQueue '%s'", self->queue_name);
SICSLogWrite(line, eStatus);
Log(DEBUG,"ASQUIO",line);
AQ_Purge(self);
AQ_Notify(self, AQU_RECONNECT);
return 1;
@@ -344,7 +344,7 @@ static int StartCommand(pAsyncQueue self)
int iRet = 0;
if (self->state != eAsyncConnected)
SICSLogPrintf(eStatus, "Function: %s:%s\n", self->queue_name,
Log(DEBUG,"ASQUIO", "Function: %s:%s\n", self->queue_name,
__func__);
if (myCmd == NULL)
return OKOK;
@@ -393,10 +393,7 @@ static int StartCommand(pAsyncQueue self)
} else if (iRet > 0) {
struct timeval tv;
gettimeofday(&tv, NULL);
SICSLogTimePrintf(eError, &tv,
"ERROR: %d unsolicited chars in AsyncQueue %s",
iRet, self->queue_name);
SICSLogWriteHexTime(reply, iRet, eError, &tv);
Log(ERROR, "ASQUIO","%d unsolicited chars in AsyncQueue %s",iRet, self->queue_name);
}
}
}
@@ -504,9 +501,7 @@ static int CommandTimeout(void *cntx, int mode)
if (self->trace) {
struct timeval tv;
gettimeofday(&tv, NULL);
SICSLogTimePrintf(eLog, &tv,
"Timeout Trace on AsyncQueue %s", self->queue_name);
SICSLogWriteHexTime(myCmd->tran->inp_buf, myCmd->tran->inp_idx, eLog, &tv);
Log(DEBUG,"ASQUIO", "Timeout Trace on AsyncQueue %s", self->queue_name);
}
if (myCmd->retries > 0) {
--myCmd->retries;
@@ -530,7 +525,7 @@ static int DelayedStart(void *cntx, int mode)
{
pAsyncQueue self = (pAsyncQueue) cntx;
if (self->state != eAsyncConnected)
SICSLogPrintf(eStatus, "Function: %s:%s\n", self->queue_name,
Log(DEBUG,"ASQUIO", "Function: %s:%s\n", self->queue_name,
__func__);
self->nw_tmr = 0;
StartCommand(self);
@@ -542,7 +537,7 @@ static int MyCallback(void *context, int mode)
pAsyncQueue self = (pAsyncQueue) context;
if (self->state != eAsyncConnected)
SICSLogPrintf(eStatus, "Function: %s:%s\n", self->queue_name,
Log(DEBUG,"ASQUIO", "Function: %s:%s\n", self->queue_name,
__func__);
if (mode & nwatch_read) {
int iRet;
@@ -572,9 +567,8 @@ static int MyCallback(void *context, int mode)
if (self->trace) {
struct timeval tv;
gettimeofday(&tv, NULL);
SICSLogTimePrintf(eLog, &tv,
"Input Trace on AsyncQueue %s", self->queue_name);
SICSLogWriteHexTime(myCmd->tran->inp_buf, myCmd->tran->inp_idx, eLog, &tv);
Log(INFO, "ASQUIO",
"Received:%s:%s", self->queue_name,myCmd->tran->inp_buf);
}
PopCommand(self);
break;
@@ -582,18 +576,10 @@ static int MyCallback(void *context, int mode)
int excess = nchars - 1 - i;
struct timeval tv;
gettimeofday(&tv, NULL);
SICSLogTimePrintf(eError, &tv,
"ERROR: Protocol error %d in AsyncQueue %s",
Log(ERROR,"ASQUIO",
"Protocol error %d in AsyncQueue %s",
iRet, self->queue_name);
SICSLogWriteTime("Sent:", eError, &tv);
SICSLogWriteHexTime(myCmd->tran->out_buf, myCmd->tran->out_len, eError, &tv);
SICSLogWriteTime("Received:", eError, &tv);
SICSLogWriteHexTime(myCmd->tran->inp_buf, myCmd->tran->inp_idx, eError, &tv);
SICSLogWriteTime("Processed:", eError, &tv);
SICSLogWriteHexTime(&reply[0], i, eError, &tv);
SICSLogWriteTime("Unprocessed:", eError, &tv);
SICSLogWriteHexTime(&reply[i], excess, eError, &tv);
/* TODO: error */
Log(ERROR,"ASQUIO","SENT:%s:Received:%s",myCmd->tran->out_buf, myCmd->tran->inp_buf);
break;
}
}
@@ -601,30 +587,26 @@ static int MyCallback(void *context, int mode)
int excess = nchars - 1 - i;
struct timeval tv;
gettimeofday(&tv, NULL);
SICSLogTimePrintf(eError, &tv,
"ERROR: %d excess chars in AsyncQueue %s",
Log(ERROR, "ASQUIO", "%d excess chars in AsyncQueue %s",
excess, self->queue_name);
SICSLogWriteHexTime(&reply[i], excess, eError, &tv);
/* TODO: handle unsolicited */
}
} else {
int excess = nchars - 1 - i;
struct timeval tv;
gettimeofday(&tv, NULL);
SICSLogTimePrintf(eError, &tv,
"ERROR: %d unsolicited chars in AsyncQueue %s",
Log(ERROR, "ASQUIO", "%d unsolicited chars in AsyncQueue %s",
excess, self->queue_name);
SICSLogWriteHexTime(&reply[i], excess, eError, &tv);
/* TODO: handle unsolicited input */
}
}
}
if (mode & nwatch_write) {
char line[132];
SICSLogPrintf(eStatus, "Writeable socket callback on AsyncQueue %s",
Log(DEBUG,"ASQUIO", "Writeable socket callback on AsyncQueue %s",
self->queue_name);
NetWatchSetMode(self->nw_ctx, nwatch_read);
SICSLogPrintf(eStatus, "In %s:%s: state %s => eAsyncConnected\n",
Log(DEBUG,"ASQUIO", "In %s:%s: state %s => eAsyncConnected\n",
self->queue_name, __func__, state_name(self->state));
self->state = eAsyncConnected;
}
@@ -638,7 +620,7 @@ int AsyncUnitEnqueueHead(pAsyncUnit unit, pAsyncTxn context)
assert(unit && unit->queue && unit->queue->protocol);
myCmd = (pAQ_Cmd) malloc(sizeof(AQ_Cmd));
if (myCmd == NULL) {
SICSLogWrite("ERROR: Out of memory in AsyncUnitEnqueHead", eError);
Log(ERROR,"ASQUIO","Out of memory in AsyncUnitEnqueHead", eError);
return 0;
}
memset(myCmd, 0, sizeof(AQ_Cmd));
@@ -657,7 +639,7 @@ int AsyncUnitEnqueueTxn(pAsyncUnit unit, pAsyncTxn pTxn)
assert(unit && unit->queue && unit->queue->protocol);
myCmd = (pAQ_Cmd) malloc(sizeof(AQ_Cmd));
if (myCmd == NULL) {
SICSLogWrite("ERROR: Out of memory in AsyncUnitEnqueueTxn", eError);
Log(ERROR,"ASQUIO","%s","Out of memory in AsyncUnitEnqueueTxn");
return 0;
}
memset(myCmd, 0, sizeof(AQ_Cmd));
@@ -679,7 +661,7 @@ pAsyncTxn AsyncUnitPrepareTxn(pAsyncUnit unit,
assert(unit);
myTxn = (pAsyncTxn) malloc(sizeof(AsyncTxn));
if (myTxn == NULL) {
SICSLogWrite("ERROR: Out of memory in AsyncUnitPrepareTxn", eError);
Log(ERROR,"ASQUIO","%s","Out of memory in AsyncUnitPrepareTxn", eError);
return NULL;
}
memset(myTxn, 0, sizeof(AsyncTxn));
@@ -714,7 +696,7 @@ pAsyncTxn AsyncUnitPrepareTxn(pAsyncUnit unit,
} else {
myTxn->out_buf = (char *) malloc(cmd_len + 5);
if (myTxn->out_buf == NULL) {
SICSLogWrite("ERROR: Out of memory in AsyncUnitPrepareTxn", eError);
Log(ERROR,"ASQUIO","%s","Out of memory in AsyncUnitPrepareTxn");
free(myTxn);
return NULL;
}
@@ -733,7 +715,7 @@ pAsyncTxn AsyncUnitPrepareTxn(pAsyncUnit unit,
else {
myTxn->inp_buf = malloc(rsp_len + 1);
if (myTxn->inp_buf == NULL) {
SICSLogWrite("ERROR: Out of memory in AsyncUnitPrepareTxn", eError);
Log(ERROR,"ASQUIO","%s","Out of memory in AsyncUnitPrepareTxn", eError);
free(myTxn->out_buf);
free(myTxn);
return NULL;
@@ -831,9 +813,8 @@ int AsyncUnitWrite(pAsyncUnit unit, void *buffer, int buflen)
if (unit->queue->trace) {
struct timeval tv;
gettimeofday(&tv, NULL);
SICSLogTimePrintf(eLog, &tv,
"Output Trace on AsyncQueue %s", unit->queue->queue_name);
SICSLogWriteHexTime(buffer, buflen, eLog, &tv);
Log(DEBUG,"ASQUIO"
"Output Trace on AsyncQueue %s:%s", unit->queue->queue_name, buffer);
}
sock = AsyncUnitGetSocket(unit);
iRet = NETWrite(sock, buffer, buflen);
@@ -1399,7 +1380,7 @@ int AsyncUnitCreateHost(const char *host, const char *port,
unit = (pAsyncUnit) malloc(sizeof(AsyncUnit));
if (unit == NULL) {
SICSLogWrite("ERROR: Out of memory in AsyncUnitCreateHost", eError);
Log(ERROR,"ASQUIO","%s","Out of memory in AsyncUnitCreateHost", eError);
*handle = NULL;
return 0;
}