diff options
author | Rainer Gerhards <rgerhards@adiscon.com> | 2009-08-21 10:37:59 +0200 |
---|---|---|
committer | Rainer Gerhards <rgerhards@adiscon.com> | 2009-08-21 10:37:59 +0200 |
commit | 596383988792635f128cb645c9b7c8aae50453e6 (patch) | |
tree | 5e5ba6dc03b6e9b687b09f49828b5a1fcdda1cda /runtime | |
parent | 8624f937523c1ca78cc1d78b8eba18f628ae9ab5 (diff) | |
parent | daa76ad94428599336ddafdd6854dc0b71356180 (diff) | |
download | rsyslog-596383988792635f128cb645c9b7c8aae50453e6.tar.gz rsyslog-596383988792635f128cb645c9b7c8aae50453e6.tar.xz rsyslog-596383988792635f128cb645c9b7c8aae50453e6.zip |
Merge branch 'v4-devel' into v4-beta
Diffstat (limited to 'runtime')
-rw-r--r-- | runtime/apc.c | 2 | ||||
-rw-r--r-- | runtime/datetime.c | 56 | ||||
-rw-r--r-- | runtime/debug.c | 4 | ||||
-rw-r--r-- | runtime/debug.h | 3 | ||||
-rw-r--r-- | runtime/msg.c | 319 | ||||
-rw-r--r-- | runtime/msg.h | 4 | ||||
-rw-r--r-- | runtime/rule.c | 7 | ||||
-rw-r--r-- | runtime/stream.c | 445 | ||||
-rw-r--r-- | runtime/stream.h | 19 | ||||
-rw-r--r-- | runtime/wti.c | 1 | ||||
-rw-r--r-- | runtime/wtp.c | 7 |
11 files changed, 547 insertions, 320 deletions
diff --git a/runtime/apc.c b/runtime/apc.c index 5919191d..bc330e39 100644 --- a/runtime/apc.c +++ b/runtime/apc.c @@ -335,9 +335,11 @@ CancelApc(apc_id_t id) { DEFVARS_mutexProtection_uncond; + BEGINfunc BEGIN_MTX_PROTECTED_OPERATIONS_UNCOND(&listMutex); deleteApc(id); END_MTX_PROTECTED_OPERATIONS_UNCOND(&listMutex); + ENDfunc return RS_RET_OK; } diff --git a/runtime/datetime.c b/runtime/datetime.c index 2db1d3c5..dfa56b4f 100644 --- a/runtime/datetime.c +++ b/runtime/datetime.c @@ -335,6 +335,10 @@ ParseTIMESTAMP3164(struct syslogTime *pTime, uchar** ppszTS) * We will use this for parsing, as it probably is the * fastest way to parse it. * + * 2009-08-17: we now do case-insensitive comparisons, as some devices obviously do not + * obey to the RFC-specified case. As we need to guess in any case, we can ignore case + * in the first place -- rgerhards + * * 2005-07-18, well sometimes it pays to be a bit more verbose, even in C... * Fixed a bug that lead to invalid detection of the data. The issue was that * we had an if(++pszTS == 'x') inside of some of the consturcts below. However, @@ -346,20 +350,21 @@ ParseTIMESTAMP3164(struct syslogTime *pTime, uchar** ppszTS) */ switch(*pszTS++) { + case 'j': case 'J': - if(*pszTS == 'a') { + if(*pszTS == 'a' || *pszTS == 'A') { ++pszTS; - if(*pszTS == 'n') { + if(*pszTS == 'n' || *pszTS == 'N') { ++pszTS; month = 1; } else ABORT_FINALIZE(RS_RET_INVLD_TIME); - } else if(*pszTS == 'u') { + } else if(*pszTS == 'u' || *pszTS == 'U') { ++pszTS; - if(*pszTS == 'n') { + if(*pszTS == 'n' || *pszTS == 'N') { ++pszTS; month = 6; - } else if(*pszTS == 'l') { + } else if(*pszTS == 'l' || *pszTS == 'L') { ++pszTS; month = 7; } else @@ -367,10 +372,11 @@ ParseTIMESTAMP3164(struct syslogTime *pTime, uchar** ppszTS) } else ABORT_FINALIZE(RS_RET_INVLD_TIME); break; + case 'f': case 'F': - if(*pszTS == 'e') { + if(*pszTS == 'e' || *pszTS == 'E') { ++pszTS; - if(*pszTS == 'b') { + if(*pszTS == 'b' || *pszTS == 'B') { ++pszTS; month = 2; } else @@ -378,13 +384,14 @@ ParseTIMESTAMP3164(struct syslogTime *pTime, uchar** ppszTS) } else ABORT_FINALIZE(RS_RET_INVLD_TIME); break; + case 'm': case 'M': - if(*pszTS == 'a') { + if(*pszTS == 'a' || *pszTS == 'A') { ++pszTS; - if(*pszTS == 'r') { + if(*pszTS == 'r' || *pszTS == 'R') { ++pszTS; month = 3; - } else if(*pszTS == 'y') { + } else if(*pszTS == 'y' || *pszTS == 'Y') { ++pszTS; month = 5; } else @@ -392,17 +399,18 @@ ParseTIMESTAMP3164(struct syslogTime *pTime, uchar** ppszTS) } else ABORT_FINALIZE(RS_RET_INVLD_TIME); break; + case 'a': case 'A': - if(*pszTS == 'p') { + if(*pszTS == 'p' || *pszTS == 'P') { ++pszTS; - if(*pszTS == 'r') { + if(*pszTS == 'r' || *pszTS == 'R') { ++pszTS; month = 4; } else ABORT_FINALIZE(RS_RET_INVLD_TIME); - } else if(*pszTS == 'u') { + } else if(*pszTS == 'u' || *pszTS == 'U') { ++pszTS; - if(*pszTS == 'g') { + if(*pszTS == 'g' || *pszTS == 'G') { ++pszTS; month = 8; } else @@ -410,10 +418,11 @@ ParseTIMESTAMP3164(struct syslogTime *pTime, uchar** ppszTS) } else ABORT_FINALIZE(RS_RET_INVLD_TIME); break; + case 's': case 'S': - if(*pszTS == 'e') { + if(*pszTS == 'e' || *pszTS == 'E') { ++pszTS; - if(*pszTS == 'p') { + if(*pszTS == 'p' || *pszTS == 'P') { ++pszTS; month = 9; } else @@ -421,10 +430,11 @@ ParseTIMESTAMP3164(struct syslogTime *pTime, uchar** ppszTS) } else ABORT_FINALIZE(RS_RET_INVLD_TIME); break; + case 'o': case 'O': - if(*pszTS == 'c') { + if(*pszTS == 'c' || *pszTS == 'C') { ++pszTS; - if(*pszTS == 't') { + if(*pszTS == 't' || *pszTS == 'T') { ++pszTS; month = 10; } else @@ -432,10 +442,11 @@ ParseTIMESTAMP3164(struct syslogTime *pTime, uchar** ppszTS) } else ABORT_FINALIZE(RS_RET_INVLD_TIME); break; + case 'n': case 'N': - if(*pszTS == 'o') { + if(*pszTS == 'o' || *pszTS == 'O') { ++pszTS; - if(*pszTS == 'v') { + if(*pszTS == 'v' || *pszTS == 'V') { ++pszTS; month = 11; } else @@ -443,10 +454,11 @@ ParseTIMESTAMP3164(struct syslogTime *pTime, uchar** ppszTS) } else ABORT_FINALIZE(RS_RET_INVLD_TIME); break; + case 'd': case 'D': - if(*pszTS == 'e') { + if(*pszTS == 'e' || *pszTS == 'E') { ++pszTS; - if(*pszTS == 'c') { + if(*pszTS == 'c' || *pszTS == 'C') { ++pszTS; month = 12; } else diff --git a/runtime/debug.c b/runtime/debug.c index 4ee90226..807fd3f7 100644 --- a/runtime/debug.c +++ b/runtime/debug.c @@ -732,6 +732,8 @@ static void dbgGetThrdName(char *pszBuf, size_t lenBuf, pthread_t thrd, int bInc */ void dbgSetThrdName(uchar *pszName) { +return; + dbgThrdInfo_t *pThrd = dbgGetThrdInfo(); if(pThrd->pszThrdName != NULL) free(pThrd->pszThrdName); @@ -776,7 +778,7 @@ static void dbgCallStackPrint(dbgThrdInfo_t *pThrd) /* print all threads call stacks */ -static void dbgCallStackPrintAll(void) +void dbgCallStackPrintAll(void) { dbgThrdInfo_t *pThrd; /* stack info */ diff --git a/runtime/debug.h b/runtime/debug.h index 1375493d..dcbfb930 100644 --- a/runtime/debug.h +++ b/runtime/debug.h @@ -134,8 +134,7 @@ void dbgPrintAllDebugInfo(void); /* debug aides */ -//#ifdef RTINST -#if 0 // temporarily removed for helgrind +#ifdef RTINST #define d_pthread_mutex_lock(x) dbgMutexLock(x, pdbgFuncDB, __LINE__, dbgCALLStaCK_POP_POINT ) #define d_pthread_mutex_trylock(x) dbgMutexTryLock(x, pdbgFuncDB, __LINE__, dbgCALLStaCK_POP_POINT ) #define d_pthread_mutex_unlock(x) dbgMutexUnlock(x, pdbgFuncDB, __LINE__, dbgCALLStaCK_POP_POINT ) diff --git a/runtime/msg.c b/runtime/msg.c index d29da560..de298871 100644 --- a/runtime/msg.c +++ b/runtime/msg.c @@ -1159,16 +1159,16 @@ int getMSGLen(msg_t *pM) return((pM == NULL) ? 0 : pM->iLenMSG); } -char *getMSG(msg_t *pM) +uchar *getMSG(msg_t *pM) { - char *ret; + uchar *ret; if(pM == NULL) - ret = ""; + ret = UCHAR_CONSTANT(""); else { if(pM->offMSG == -1) - ret = ""; + ret = UCHAR_CONSTANT(""); else - ret = (char*)(pM->pszRawMsg + pM->offMSG); + ret = pM->pszRawMsg + pM->offMSG; } return ret; } @@ -1613,21 +1613,23 @@ static inline void tryEmulateTAG(msg_t *pM, bool bLockMutex) } -static inline char *getTAG(msg_t *pM) +static inline void +getTAG(msg_t *pM, uchar **ppBuf, int *piLen) { - char *ret; - - if(pM == NULL) - ret = ""; - else { + if(pM == NULL) { + *ppBuf = UCHAR_CONSTANT(""); + *piLen = 0; + } else { if(pM->iLenTAG == 0) tryEmulateTAG(pM, LOCK_MUTEX); - if(pM->iLenTAG == 0) - ret = ""; - else - ret = (char*) ((pM->iLenTAG < CONF_TAG_BUFSIZE) ? pM->TAG.szBuf : pM->TAG.pszTAG); + if(pM->iLenTAG == 0) { + *ppBuf = UCHAR_CONSTANT(""); + *piLen = 0; + } else { + *ppBuf = (pM->iLenTAG < CONF_TAG_BUFSIZE) ? pM->TAG.szBuf : pM->TAG.pszTAG; + *piLen = pM->iLenTAG; + } } - return(ret); } @@ -2130,14 +2132,14 @@ static uchar *getNOW(eNOWType eNow) * be used in selector line processing. * rgerhards 2005-09-15 */ -char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, +uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, propid_t propID, size_t *pPropLen, unsigned short *pbMustBeFreed) { - char *pRes; /* result pointer */ + uchar *pRes; /* result pointer */ int bufLen = -1; /* length of string or -1, if not known */ - char *pBufStart; - char *pBuf; + uchar *pBufStart; + uchar *pBuf; int iLen; short iOffs; @@ -2159,16 +2161,16 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, bufLen = getMSGLen(pMsg); break; case PROP_TIMESTAMP: - pRes = getTimeReported(pMsg, pTpe->data.field.eDateFormat); + pRes = (uchar*)getTimeReported(pMsg, pTpe->data.field.eDateFormat); break; case PROP_HOSTNAME: - pRes = getHOSTNAME(pMsg); + pRes = (uchar*)getHOSTNAME(pMsg); break; case PROP_SYSLOGTAG: - pRes = getTAG(pMsg); + getTAG(pMsg, &pRes, &bufLen); break; case PROP_RAWMSG: - pRes = getRawMsg(pMsg); + pRes = (uchar*)getRawMsg(pMsg); break; /* enable this, if someone actually uses UxTradMsg, delete after some time has * passed and nobody complained -- rgerhards, 2009-06-16 @@ -2177,120 +2179,120 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, break; */ case PROP_INPUTNAME: - getInputName(pMsg, ((uchar**) &pRes), &bufLen); + getInputName(pMsg, &pRes, &bufLen); break; case PROP_FROMHOST: - pRes = (char*) getRcvFrom(pMsg); + pRes = getRcvFrom(pMsg); break; case PROP_FROMHOST_IP: - pRes = (char*) getRcvFromIP(pMsg); + pRes = getRcvFromIP(pMsg); break; case PROP_PRI: - pRes = getPRI(pMsg); + pRes = (uchar*)getPRI(pMsg); break; case PROP_PRI_TEXT: - pBuf = malloc(20 * sizeof(char)); + pBuf = malloc(20 * sizeof(uchar)); if(pBuf == NULL) { *pbMustBeFreed = 0; - return "**OUT OF MEMORY**"; + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } else { *pbMustBeFreed = 1; - pRes = textpri(pBuf, 20, getPRIi(pMsg)); + pRes = (uchar*)textpri((char*)pBuf, 20, getPRIi(pMsg)); } break; case PROP_IUT: - pRes = "1"; /* always 1 for syslog messages (a MonitorWare thing;)) */ + pRes = UCHAR_CONSTANT("1"); /* always 1 for syslog messages (a MonitorWare thing;)) */ break; case PROP_SYSLOGFACILITY: - pRes = getFacility(pMsg); + pRes = (uchar*)getFacility(pMsg); break; case PROP_SYSLOGFACILITY_TEXT: - pRes = getFacilityStr(pMsg); + pRes = (uchar*)getFacilityStr(pMsg); break; case PROP_SYSLOGSEVERITY: - pRes = getSeverity(pMsg); + pRes = (uchar*)getSeverity(pMsg); break; case PROP_SYSLOGSEVERITY_TEXT: - pRes = getSeverityStr(pMsg); + pRes = (uchar*)getSeverityStr(pMsg); break; case PROP_TIMEGENERATED: - pRes = getTimeGenerated(pMsg, pTpe->data.field.eDateFormat); + pRes = (uchar*)getTimeGenerated(pMsg, pTpe->data.field.eDateFormat); break; case PROP_PROGRAMNAME: - pRes = getProgramName(pMsg, LOCK_MUTEX); + pRes = (uchar*)getProgramName(pMsg, LOCK_MUTEX); break; case PROP_PROTOCOL_VERSION: - pRes = getProtocolVersionString(pMsg); + pRes = (uchar*)getProtocolVersionString(pMsg); break; case PROP_STRUCTURED_DATA: - pRes = getStructuredData(pMsg); + pRes = (uchar*)getStructuredData(pMsg); break; case PROP_APP_NAME: - pRes = getAPPNAME(pMsg, LOCK_MUTEX); + pRes = (uchar*)getAPPNAME(pMsg, LOCK_MUTEX); break; case PROP_PROCID: - pRes = getPROCID(pMsg, LOCK_MUTEX); + pRes = (uchar*)getPROCID(pMsg, LOCK_MUTEX); break; case PROP_MSGID: - pRes = getMSGID(pMsg); + pRes = (uchar*)getMSGID(pMsg); break; case PROP_SYS_NOW: - if((pRes = (char*) getNOW(NOW_NOW)) == NULL) { - return "***OUT OF MEMORY***"; + if((pRes = getNOW(NOW_NOW)) == NULL) { + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } else *pbMustBeFreed = 1; /* all of these functions allocate dyn. memory */ break; case PROP_SYS_YEAR: - if((pRes = (char*) getNOW(NOW_YEAR)) == NULL) { - return "***OUT OF MEMORY***"; + if((pRes = getNOW(NOW_YEAR)) == NULL) { + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } else *pbMustBeFreed = 1; /* all of these functions allocate dyn. memory */ break; case PROP_SYS_MONTH: - if((pRes = (char*) getNOW(NOW_MONTH)) == NULL) { - return "***OUT OF MEMORY***"; + if((pRes = getNOW(NOW_MONTH)) == NULL) { + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } else *pbMustBeFreed = 1; /* all of these functions allocate dyn. memory */ break; case PROP_SYS_DAY: - if((pRes = (char*) getNOW(NOW_DAY)) == NULL) { - return "***OUT OF MEMORY***"; + if((pRes = getNOW(NOW_DAY)) == NULL) { + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } else *pbMustBeFreed = 1; /* all of these functions allocate dyn. memory */ break; case PROP_SYS_HOUR: - if((pRes = (char*) getNOW(NOW_HOUR)) == NULL) { - return "***OUT OF MEMORY***"; + if((pRes = getNOW(NOW_HOUR)) == NULL) { + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } else *pbMustBeFreed = 1; /* all of these functions allocate dyn. memory */ break; case PROP_SYS_HHOUR: - if((pRes = (char*) getNOW(NOW_HHOUR)) == NULL) { - return "***OUT OF MEMORY***"; + if((pRes = getNOW(NOW_HHOUR)) == NULL) { + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } else *pbMustBeFreed = 1; /* all of these functions allocate dyn. memory */ break; case PROP_SYS_QHOUR: - if((pRes = (char*) getNOW(NOW_QHOUR)) == NULL) { - return "***OUT OF MEMORY***"; + if((pRes = getNOW(NOW_QHOUR)) == NULL) { + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } else *pbMustBeFreed = 1; /* all of these functions allocate dyn. memory */ break; case PROP_SYS_MINUTE: - if((pRes = (char*) getNOW(NOW_MINUTE)) == NULL) { - return "***OUT OF MEMORY***"; + if((pRes = getNOW(NOW_MINUTE)) == NULL) { + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } else *pbMustBeFreed = 1; /* all of these functions allocate dyn. memory */ break; case PROP_SYS_MYHOSTNAME: - pRes = (char*) glbl.GetLocalHostName(); + pRes = glbl.GetLocalHostName(); break; default: /* there is no point in continuing, we may even otherwise render the * error message unreadable. rgerhards, 2007-07-10 */ dbgprintf("invalid property id: '%d'\n", propID); - return "**INVALID PROPERTY NAME**"; + return UCHAR_CONSTANT("**INVALID PROPERTY NAME**"); } @@ -2310,8 +2312,8 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, */ if(pTpe->data.field.has_fields == 1) { size_t iCurrFld; - char *pFld; - char *pFldEnd; + uchar *pFld; + uchar *pFldEnd; /* first, skip to the field in question. The field separator * is always one character and is stored in the template entry. */ @@ -2349,7 +2351,7 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, if(*pbMustBeFreed == 1) free(pRes); *pbMustBeFreed = 0; - return "**OUT OF MEMORY**"; + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } /* now copy */ memcpy(pBuf, pFld, iLen); @@ -2366,12 +2368,12 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, if(*pbMustBeFreed == 1) free(pRes); *pbMustBeFreed = 0; - return "**FIELD NOT FOUND**"; + return UCHAR_CONSTANT("**FIELD NOT FOUND**"); } } else if(pTpe->data.field.iFromPos != 0 || pTpe->data.field.iToPos != 0) { /* we need to obtain a private copy */ int iFrom, iTo; - char *pSb; + uchar *pSb; iFrom = pTpe->data.field.iFromPos; iTo = pTpe->data.field.iToPos; /* need to zero-base to and from (they are 1-based!) */ @@ -2379,44 +2381,55 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, --iFrom; if(iTo > 0) --iTo; - iLen = iTo - iFrom + 1; /* the +1 is for an actual char, NOT \0! */ - pBufStart = pBuf = malloc((iLen + 1) * sizeof(char)); - if(pBuf == NULL) { - if(*pbMustBeFreed == 1) - free(pRes); - *pbMustBeFreed = 0; - return "**OUT OF MEMORY**"; - } - pSb = pRes; - if(iFrom) { - /* skip to the start of the substring (can't do pointer arithmetic - * because the whole string might be smaller!!) - */ - while(*pSb && iFrom) { - --iFrom; + if(bufLen == -1) + bufLen = ustrlen(pRes); + if(iFrom == 0 && iTo >= bufLen) { + /* in this case, the requested string is a superset of what we already have, + * so there is no need to do any processing. This is a frequent case for size-limited + * fields like TAG in the default forwarding template (so it is a useful optimization + * to check for this condition ;)). -- rgerhards, 2009-07-09 + */ + ; /*DO NOTHING*/ + } else { + iLen = iTo - iFrom + 1; /* the +1 is for an actual char, NOT \0! */ + pBufStart = pBuf = malloc((iLen + 1) * sizeof(char)); + if(pBuf == NULL) { + if(*pbMustBeFreed == 1) + free(pRes); + *pbMustBeFreed = 0; + return UCHAR_CONSTANT("**OUT OF MEMORY**"); + } + pSb = pRes; + if(iFrom) { + /* skip to the start of the substring (can't do pointer arithmetic + * because the whole string might be smaller!!) + */ + while(*pSb && iFrom) { + --iFrom; + ++pSb; + } + } + /* OK, we are at the begin - now let's copy... */ + bufLen = iLen; + while(*pSb && iLen) { + *pBuf++ = *pSb; ++pSb; + --iLen; } + *pBuf = '\0'; + bufLen -= iLen; /* subtract remaining length if the string was smaller! */ + if(*pbMustBeFreed == 1) + free(pRes); + pRes = pBufStart; + *pbMustBeFreed = 1; } - /* OK, we are at the begin - now let's copy... */ - bufLen = iLen; - while(*pSb && iLen) { - *pBuf++ = *pSb; - ++pSb; - --iLen; - } - *pBuf = '\0'; - bufLen -= iLen; /* subtract remaining length if the string was smaller! */ - if(*pbMustBeFreed == 1) - free(pRes); - pRes = pBufStart; - *pbMustBeFreed = 1; #ifdef FEATURE_REGEXP } else { /* Check for regular expressions */ if (pTpe->data.field.has_regex != 0) { if (pTpe->data.field.has_regex == 2) /* Could not compile regex before! */ - return "**NO MATCH** **BAD REGULAR EXPRESSION**"; + return UCHAR_CONSTANT("**NO MATCH** **BAD REGULAR EXPRESSION**"); dbgprintf("string to match for regex is: %s\n", pRes); @@ -2429,7 +2442,7 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, */ while(!bFound) { int iREstat; - iREstat = regexp.regexec(&pTpe->data.field.re, pRes + iOffs, nmatch, pmatch, 0); + iREstat = regexp.regexec(&pTpe->data.field.re, (char*)(pRes + iOffs), nmatch, pmatch, 0); dbgprintf("regexec return is %d\n", iREstat); if(iREstat == 0) { if(pmatch[0].rm_so == -1) { @@ -2457,11 +2470,11 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, *pbMustBeFreed = 0; } if(pTpe->data.field.nomatchAction == TPL_REGEX_NOMATCH_USE_DFLTSTR) - return "**NO MATCH**"; + return UCHAR_CONSTANT("**NO MATCH**"); else if(pTpe->data.field.nomatchAction == TPL_REGEX_NOMATCH_USE_ZERO) - return "0"; + return UCHAR_CONSTANT("0"); else - return ""; + return UCHAR_CONSTANT(""); } } else { /* Match- but did it match the one we wanted? */ @@ -2473,24 +2486,24 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, *pbMustBeFreed = 0; } if(pTpe->data.field.nomatchAction == TPL_REGEX_NOMATCH_USE_DFLTSTR) - return "**NO MATCH**"; + return UCHAR_CONSTANT("**NO MATCH**"); else - return ""; + return UCHAR_CONSTANT(""); } } /* OK, we have a usable match - we now need to malloc pB */ int iLenBuf; - char *pB; + uchar *pB; iLenBuf = pmatch[pTpe->data.field.iSubMatchToUse].rm_eo - pmatch[pTpe->data.field.iSubMatchToUse].rm_so; - pB = (char *) malloc((iLenBuf + 1) * sizeof(char)); + pB = malloc((iLenBuf + 1) * sizeof(uchar)); if (pB == NULL) { if (*pbMustBeFreed == 1) free(pRes); *pbMustBeFreed = 0; - return "**OUT OF MEMORY ALLOCATING pBuf**"; + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } /* Lets copy the matched substring to the buffer */ @@ -2513,7 +2526,7 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, free(pRes); *pbMustBeFreed = 0; } - return "***REGEXP NOT AVAILABLE***"; + return UCHAR_CONSTANT("***REGEXP NOT AVAILABLE***"); } } #endif /* #ifdef FEATURE_REGEXP */ @@ -2525,7 +2538,7 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, uchar cFirst = *pRes; /* save first char */ if(*pbMustBeFreed == 1) free(pRes); - pRes = (cFirst == ' ') ? "" : " "; + pRes = (cFirst == ' ') ? UCHAR_CONSTANT("") : UCHAR_CONSTANT(" "); bufLen = (cFirst == ' ') ? 0 : 1; *pbMustBeFreed = 0; } @@ -2537,21 +2550,21 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, if(pTpe->data.field.eCaseConv != tplCaseConvNo) { /* we need to obtain a private copy */ if(bufLen == -1) - bufLen = strlen(pRes); - char *pBStart; - char *pB; - char *pSrc; + bufLen = ustrlen(pRes); + uchar *pBStart; + uchar *pB; + uchar *pSrc; pBStart = pB = malloc((bufLen + 1) * sizeof(char)); if(pB == NULL) { if(*pbMustBeFreed == 1) free(pRes); *pbMustBeFreed = 0; - return "**OUT OF MEMORY**"; + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } pSrc = pRes; while(*pSrc) { *pB++ = (pTpe->data.field.eCaseConv == tplCaseConvUpper) ? - (char)toupper((int)*pSrc) : (char)tolower((int)*pSrc); + (uchar)toupper((int)*pSrc) : (uchar)tolower((int)*pSrc); /* currently only these two exist */ ++pSrc; } @@ -2575,10 +2588,10 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, */ if(pTpe->data.field.options.bDropCC) { int iLenBuf = 0; - char *pSrc = pRes; - char *pDstStart; - char *pDst; - char bDropped = 0; + uchar *pSrc = pRes; + uchar *pDstStart; + uchar *pDst; + uchar bDropped = 0; while(*pSrc) { if(!iscntrl((int) *pSrc++)) @@ -2593,7 +2606,7 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, if(*pbMustBeFreed == 1) free(pRes); *pbMustBeFreed = 0; - return "**OUT OF MEMORY**"; + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } for(pSrc = pRes; *pSrc; pSrc++) { if(!iscntrl((int) *pSrc)) @@ -2607,9 +2620,9 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, *pbMustBeFreed = 1; } } else if(pTpe->data.field.options.bSpaceCC) { - char *pSrc; - char *pDstStart; - char *pDst; + uchar *pSrc; + uchar *pDstStart; + uchar *pDst; if(*pbMustBeFreed == 1) { /* in this case, we already work on dynamic @@ -2623,13 +2636,13 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, } } else { if(bufLen == -1) - bufLen = strlen(pRes); + bufLen = ustrlen(pRes); pDst = pDstStart = malloc(bufLen + 1); if(pDst == NULL) { if(*pbMustBeFreed == 1) free(pRes); *pbMustBeFreed = 0; - return "**OUT OF MEMORY**"; + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } for(pSrc = pRes; *pSrc; pSrc++) { if(iscntrl((int) *pSrc)) @@ -2649,7 +2662,7 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, */ int iNumCC = 0; int iLenBuf = 0; - char *pB; + uchar *pB; for(pB = pRes ; *pB ; ++pB) { ++iLenBuf; @@ -2659,21 +2672,21 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, if(iNumCC > 0) { /* if 0, there is nothing to escape, so we are done */ /* OK, let's do the escaping... */ - char *pBStart; - char szCCEsc[8]; /* buffer for escape sequence */ + uchar *pBStart; + uchar szCCEsc[8]; /* buffer for escape sequence */ int i; iLenBuf += iNumCC * 4; - pBStart = pB = malloc((iLenBuf + 1) * sizeof(char)); + pBStart = pB = malloc((iLenBuf + 1) * sizeof(uchar)); if(pB == NULL) { if(*pbMustBeFreed == 1) free(pRes); *pbMustBeFreed = 0; - return "**OUT OF MEMORY**"; + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } while(*pRes) { if(iscntrl((int) *pRes)) { - snprintf(szCCEsc, sizeof(szCCEsc), "#%3.3d", *pRes); + snprintf((char*)szCCEsc, sizeof(szCCEsc), "#%3.3d", *pRes); for(i = 0 ; i < 4 ; ++i) *pB++ = szCCEsc[i]; } else { @@ -2697,10 +2710,10 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, if(pTpe->data.field.options.bSecPathDrop || pTpe->data.field.options.bSecPathReplace) { if(pTpe->data.field.options.bSecPathDrop) { int iLenBuf = 0; - char *pSrc = pRes; - char *pDstStart; - char *pDst; - char bDropped = 0; + uchar *pSrc = pRes; + uchar *pDstStart; + uchar *pDst; + uchar bDropped = 0; while(*pSrc) { if(*pSrc++ != '/') @@ -2715,7 +2728,7 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, if(*pbMustBeFreed == 1) free(pRes); *pbMustBeFreed = 0; - return "**OUT OF MEMORY**"; + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } for(pSrc = pRes; *pSrc; pSrc++) { if(*pSrc != '/') @@ -2729,9 +2742,9 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, *pbMustBeFreed = 1; } } else { - char *pSrc; - char *pDstStart; - char *pDst; + uchar *pSrc; + uchar *pDstStart; + uchar *pDst; if(*pbMustBeFreed == 1) { /* here, again, we can modify the string as we already obtained @@ -2745,13 +2758,13 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, } } else { if(bufLen == -1) - bufLen = strlen(pRes); + bufLen = ustrlen(pRes); pDst = pDstStart = malloc(bufLen + 1); if(pDst == NULL) { if(*pbMustBeFreed == 1) free(pRes); *pbMustBeFreed = 0; - return "**OUT OF MEMORY**"; + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } for(pSrc = pRes; *pSrc; pSrc++) { if(*pSrc == '/') @@ -2771,19 +2784,19 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, /* check for "." and ".." (note the parenthesis in the if condition!) */ if((*pRes == '.') && (*(pRes + 1) == '\0' || (*(pRes + 1) == '.' && *(pRes + 2) == '\0'))) { - char *pTmp = pRes; + uchar *pTmp = pRes; if(*(pRes + 1) == '\0') - pRes = "_"; + pRes = UCHAR_CONSTANT("_"); else - pRes = "_.";; + pRes = UCHAR_CONSTANT("_.");; if(*pbMustBeFreed == 1) free(pTmp); *pbMustBeFreed = 0; } else if(*pRes == '\0') { if(*pbMustBeFreed == 1) free(pRes); - pRes = "_"; + pRes = UCHAR_CONSTANT("_"); bufLen = 1; *pbMustBeFreed = 0; } @@ -2794,19 +2807,19 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, */ if(pTpe->data.field.options.bDropLastLF && !pTpe->data.field.options.bEscapeCC) { int iLn; - char *pB; + uchar *pB; if(bufLen == -1) - bufLen = strlen(pRes); + bufLen = ustrlen(pRes); iLn = bufLen; if(iLn > 0 && *(pRes + iLn - 1) == '\n') { /* we have a LF! */ /* check if we need to obtain a private copy */ if(*pbMustBeFreed == 0) { /* ok, original copy, need a private one */ - pB = malloc((iLn + 1) * sizeof(char)); + pB = malloc((iLn + 1) * sizeof(uchar)); if(pB == NULL) { *pbMustBeFreed = 0; - return "**OUT OF MEMORY**"; + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } memcpy(pB, pRes, iLn - 1); pRes = pB; @@ -2825,19 +2838,19 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, if(pTpe->data.field.options.bCSV) { /* we need to obtain a private copy, as we need to at least add the double quotes */ int iBufLen; - char *pBStart; - char *pDst; - char *pSrc; + uchar *pBStart; + uchar *pDst; + uchar *pSrc; if(bufLen == -1) - bufLen = strlen(pRes); + bufLen = ustrlen(pRes); iBufLen = bufLen; /* the malloc may be optimized, we currently use the worst case... */ - pBStart = pDst = malloc((2 * iBufLen + 3) * sizeof(char)); + pBStart = pDst = malloc((2 * iBufLen + 3) * sizeof(uchar)); if(pDst == NULL) { if(*pbMustBeFreed == 1) free(pRes); *pbMustBeFreed = 0; - return "**OUT OF MEMORY**"; + return UCHAR_CONSTANT("**OUT OF MEMORY**"); } pSrc = pRes; *pDst++ = '"'; /* starting quote */ @@ -2856,7 +2869,7 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, } if(bufLen == -1) - bufLen = strlen(pRes); + bufLen = ustrlen(pRes); *pPropLen = bufLen; ENDfunc diff --git a/runtime/msg.h b/runtime/msg.h index c20fb005..0b346f7b 100644 --- a/runtime/msg.h +++ b/runtime/msg.h @@ -159,7 +159,7 @@ void MsgSetMSGoffs(msg_t *pMsg, short offs); void MsgSetRawMsgWOSize(msg_t *pMsg, char* pszRawMsg); void MsgSetRawMsg(msg_t *pMsg, char* pszRawMsg, size_t lenMsg); rsRetVal MsgReplaceMSG(msg_t *pThis, uchar* pszMSG, int lenMSG); -char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, +uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, propid_t propID, size_t *pPropLen, unsigned short *pbMustBeFreed); char *textpri(char *pRes, size_t pResLen, int pri); rsRetVal msgGetMsgVar(msg_t *pThis, cstr_t *pstrPropName, var_t **ppVar); @@ -168,7 +168,7 @@ uchar *getRcvFrom(msg_t *pM); /* TODO: remove these five (so far used in action.c) */ -char *getMSG(msg_t *pM); +uchar *getMSG(msg_t *pM); char *getHOSTNAME(msg_t *pM); char *getPROCID(msg_t *pM, bool bLockMutex); char *getAPPNAME(msg_t *pM, bool bLockMutex); diff --git a/runtime/rule.c b/runtime/rule.c index 3a257a90..182d616a 100644 --- a/runtime/rule.c +++ b/runtime/rule.c @@ -39,6 +39,7 @@ #include "vm.h" #include "var.h" #include "srUtils.h" +#include "unicode-helper.h" #include "dirty.h" /* for getFIOPName */ /* static data */ @@ -104,7 +105,7 @@ shouldProcessThisMessage(rule_t *pRule, msg_t *pMsg, int *bProcessMsg) { DEFiRet; unsigned short pbMustBeFreed; - char *pszPropVal; + uchar *pszPropVal; int bRet = 0; size_t propLen; vm_t *pVM = NULL; @@ -189,12 +190,12 @@ shouldProcessThisMessage(rule_t *pRule, msg_t *pMsg, int *bProcessMsg) break; case FIOP_ISEQUAL: if(rsCStrSzStrCmp(pRule->f_filterData.prop.pCSCompValue, - (uchar*) pszPropVal, strlen(pszPropVal)) == 0) + pszPropVal, ustrlen(pszPropVal)) == 0) bRet = 1; /* process message! */ break; case FIOP_STARTSWITH: if(rsCStrSzStrStartsWithCStr(pRule->f_filterData.prop.pCSCompValue, - (uchar*) pszPropVal, strlen(pszPropVal)) == 0) + pszPropVal, ustrlen(pszPropVal)) == 0) bRet = 1; /* process message! */ break; case FIOP_REGEX: diff --git a/runtime/stream.c b/runtime/stream.c index 00c726d9..605a9771 100644 --- a/runtime/stream.c +++ b/runtime/stream.c @@ -48,37 +48,27 @@ #include "stream.h" #include "unicode-helper.h" #include "module-template.h" -#include "apc.h" +#if HAVE_SYS_PRCTL_H +# include <sys/prctl.h> +#endif + +#define inline /* static data */ DEFobjStaticHelpers DEFobjCurrIf(zlibw) -DEFobjCurrIf(apc) /* forward definitions */ static rsRetVal strmFlush(strm_t *pThis); static rsRetVal strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf); static rsRetVal strmCloseFile(strm_t *pThis); +static void *asyncWriterThread(void *pPtr); +static rsRetVal doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf); +static rsRetVal strmPhysWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf); /* methods */ -/* async flush apc handler - */ -static void -flushApc(void *param1, void __attribute__((unused)) *param2) -{ - DEFVARS_mutexProtection_uncond; - strm_t *pThis = (strm_t*) param1; - ISOBJ_TYPE_assert(pThis, strm); - - BEGIN_MTX_PROTECTED_OPERATIONS_UNCOND(&pThis->mut); - strmFlush(pThis); - pThis->apcRequested = 0; - END_MTX_PROTECTED_OPERATIONS_UNCOND(&pThis->mut); -} - - /* Try to resolve a size limit situation. This is used to support custom-file size handlers * for omfile. It first runs the command, and then checks if we are still above the size * treshold. Note that this works only with single file names, NOT with circular names. @@ -139,10 +129,11 @@ resolveFileSizeLimit(strm_t *pThis, uchar *pszCurrFName) finalize_it: if(iRet != RS_RET_OK) { - if(iRet == RS_RET_SIZELIMITCMD_DIDNT_RESOLVE) - dbgprintf("file size limit cmd for file '%s' did no resolve situation\n", pszCurrFName); - else - dbgprintf("file size limit cmd for file '%s' failed with code %d.\n", pszCurrFName, iRet); + if(iRet == RS_RET_SIZELIMITCMD_DIDNT_RESOLVE) { + DBGPRINTF("file size limit cmd for file '%s' did no resolve situation\n", pszCurrFName); + } else { + DBGPRINTF("file size limit cmd for file '%s' failed with code %d.\n", pszCurrFName, iRet); + } pThis->bDisabled = 1; } @@ -280,6 +271,23 @@ finalize_it: } +/* wait for the output writer thread to be done. This must be called before actions + * that require data to be persisted. May be called in non-async mode and is a null + * operation than. Must be called with the mutex locked. + */ +static inline void +strmWaitAsyncWriterDone(strm_t *pThis) +{ + BEGINfunc + if(pThis->bAsyncWrite) { + /* awake writer thread and make it write out everything */ + pthread_cond_signal(&pThis->notEmpty); + d_pthread_cond_wait(&pThis->isEmpty, &pThis->mut); + } + ENDfunc +} + + /* close a strm file * Note that the bDeleteOnClose flag is honored. If it is set, the file will be * deleted after close. This is in support for the qRead thread. @@ -292,8 +300,15 @@ static rsRetVal strmCloseFile(strm_t *pThis) ASSERT(pThis->fd != -1); dbgoprint((obj_t*) pThis, "file %d closing\n", pThis->fd); - if(pThis->tOperationsMode != STREAMMODE_READ) - strmFlush(pThis); + if(!pThis->bInClose && pThis->tOperationsMode != STREAMMODE_READ) { + pThis->bInClose = 1; + if(pThis->bAsyncWrite) { + strmFlush(pThis); + } else { + strmWaitAsyncWriterDone(pThis); + } + pThis->bInClose = 0; + } close(pThis->fd); pThis->fd = -1; @@ -569,11 +584,11 @@ ENDobjConstruct(strm) static rsRetVal strmConstructFinalize(strm_t *pThis) { rsRetVal localRet; + int i; DEFiRet; ASSERT(pThis != NULL); - CHKmalloc(pThis->pIOBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize)); pThis->iBufPtrMax = 0; /* results in immediate read request */ if(pThis->iZipLevel) { /* do we need a zip buf? */ localRet = objUse(zlibw, LM_ZLIBW_FILENAME); @@ -601,9 +616,28 @@ static rsRetVal strmConstructFinalize(strm_t *pThis) } } - /* if we should call flush apc's, we need a mutex */ + /* if we have a flush interval, we need to do async writes in any case */ if(pThis->iFlushInterval != 0) { + pThis->bAsyncWrite = 1; + } + + /* if we work asynchronously, we need a couple of synchronization objects */ + if(pThis->bAsyncWrite) { pthread_mutex_init(&pThis->mut, 0); + pthread_cond_init(&pThis->notFull, 0); + pthread_cond_init(&pThis->notEmpty, 0); + pthread_cond_init(&pThis->isEmpty, 0); + pThis->iCnt = pThis->iEnq = pThis->iDeq = 0; + for(i = 0 ; i < STREAM_ASYNC_NUMBUFS ; ++i) { + CHKmalloc(pThis->asyncBuf[i].pBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize)); + } + pThis->pIOBuf = pThis->asyncBuf[0].pBuf; + pThis->bStopWriter = 0; + if(pthread_create(&pThis->writerThreadID, NULL, asyncWriterThread, pThis) != 0) + DBGPRINTF("ERROR: stream %p cold not create writer thread\n", pThis); + } else { + /* we work synchronously, so we need to alloc a fixed pIOBuf */ + CHKmalloc(pThis->pIOBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize)); } finalize_it: @@ -611,12 +645,33 @@ finalize_it: } +/* stop the writer thread (we MUST be runnnig asynchronously when this method + * is called!). Note that the mutex must be locked! -- rgerhards, 2009-07-06 + */ +static inline void +stopWriter(strm_t *pThis) +{ + BEGINfunc + pThis->bStopWriter = 1; + pthread_cond_signal(&pThis->notEmpty); + d_pthread_mutex_unlock(&pThis->mut); + pthread_join(pThis->writerThreadID, NULL); + ENDfunc +} + + /* destructor for the strm object */ BEGINobjDestruct(strm) /* be sure to specify the object type also in END and CODESTART macros! */ + int i; CODESTARTobjDestruct(strm) + if(pThis->bAsyncWrite) + /* Note: mutex will be unlocked in stopWriter! */ + d_pthread_mutex_lock(&pThis->mut); + if(pThis->tOperationsMode != STREAMMODE_READ) strmFlush(pThis); +dbgprintf("XXX: destruct stream %p\n", pThis); /* ... then free resources */ if(pThis->fd != -1) strmCloseFile(pThis); @@ -625,16 +680,23 @@ CODESTARTobjDestruct(strm) objRelease(zlibw, LM_ZLIBW_FILENAME); } - if(pThis->iFlushInterval != 0) { - // TODO: check if there is an apc and remove it! - pthread_mutex_destroy(&pThis->mut); - } - free(pThis->pszDir); - free(pThis->pIOBuf); free(pThis->pZipBuf); free(pThis->pszCurrFName); free(pThis->pszFName); + + if(pThis->bAsyncWrite) { + stopWriter(pThis); + pthread_mutex_destroy(&pThis->mut); + pthread_cond_destroy(&pThis->notFull); + pthread_cond_destroy(&pThis->notEmpty); + pthread_cond_destroy(&pThis->isEmpty); + for(i = 0 ; i < STREAM_ASYNC_NUMBUFS ; ++i) { + free(pThis->asyncBuf[i].pBuf); + } + } else { + free(pThis->pIOBuf); + } ENDobjDestruct(strm) @@ -650,6 +712,9 @@ static rsRetVal strmCheckNextOutputFile(strm_t *pThis) if(pThis->fd == -1) FINALIZE; + /* wait for output to be empty, so that our counts are correct */ + strmWaitAsyncWriterDone(pThis); + if(pThis->iCurrOffs >= pThis->iMaxFileSize) { dbgoprint((obj_t*) pThis, "max file size %ld reached for %d, now %ld - starting new file\n", (long) pThis->iMaxFileSize, pThis->fd, (long) pThis->iCurrOffs); @@ -730,12 +795,165 @@ doWriteCall(strm_t *pThis, uchar *pBuf, size_t *pLenBuf) pWriteBuf += iWritten; } while(lenBuf > 0); /* Warning: do..while()! */ + dbgoprint((obj_t*) pThis, "file %d write wrote %d bytes\n", pThis->fd, (int) iWritten); + finalize_it: *pLenBuf = iTotalWritten; RETiRet; } + +/* write memory buffer to a stream object. + */ +static inline rsRetVal +doWriteInternal(strm_t *pThis, uchar *pBuf, size_t lenBuf) +{ + DEFiRet; + + ASSERT(pThis != NULL); + + if(pThis->iZipLevel) { + CHKiRet(doZipWrite(pThis, pBuf, lenBuf)); + } else { + /* write without zipping */ + CHKiRet(strmPhysWrite(pThis, pBuf, lenBuf)); + } + +finalize_it: + RETiRet; +} + + +/* This function is called to "do" an async write call, what primarily means that + * the data is handed over to the writer thread (which will then do the actual write + * in parallel). Note that the stream mutex has already been locked by the + * strmWrite...() calls. Also note that we always have only a single producer, + * so we can simply serially assign the next free buffer to it and be sure that + * the very some producer comes back in sequence to submit the then-filled buffers. + * This also enables us to timout on partially written buffers. -- rgerhards, 2009-07-06 + */ +static inline rsRetVal +doAsyncWriteInternal(strm_t *pThis, size_t lenBuf) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, strm); + +dbgprintf("XXX: doAsyncWriteInternal: strm %p, len %ld\n", pThis, (long) lenBuf); + while(pThis->iCnt >= STREAM_ASYNC_NUMBUFS) + d_pthread_cond_wait(&pThis->notFull, &pThis->mut); + + pThis->asyncBuf[pThis->iEnq % STREAM_ASYNC_NUMBUFS].lenBuf = lenBuf; + pThis->pIOBuf = pThis->asyncBuf[++pThis->iEnq % STREAM_ASYNC_NUMBUFS].pBuf; + + pThis->bDoTimedWait = 0; /* everything written, no need to timeout partial buffer writes */ + if(++pThis->iCnt == 1) + pthread_cond_signal(&pThis->notEmpty); + +finalize_it: + RETiRet; +} + + +/* schedule writing to the stream. Depending on our concurrency settings, + * this either directly writes to the stream or schedules writing via + * the background thread. -- rgerhards, 2009-07-07 + */ +static rsRetVal +strmSchedWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) +{ + DEFiRet; + + ASSERT(pThis != NULL); + + if(pThis->bAsyncWrite) { + CHKiRet(doAsyncWriteInternal(pThis, lenBuf)); + } else { + CHKiRet(doWriteInternal(pThis, pBuf, lenBuf)); + } + + pThis->iBufPtr = 0; /* we are at the begin of a new buffer */ + +finalize_it: + RETiRet; +} + + + +/* This is the writer thread for asynchronous mode. + * -- rgerhards, 2009-07-06 + */ +static void* +asyncWriterThread(void *pPtr) +{ + int iDeq; + struct timespec t; + bool bTimedOut = 0; + strm_t *pThis = (strm_t*) pPtr; + ISOBJ_TYPE_assert(pThis, strm); + + BEGINfunc +# if HAVE_PRCTL && defined PR_SET_NAME + if(prctl(PR_SET_NAME, "rs:asyn strmwr", 0, 0, 0) != 0) { + DBGPRINTF("prctl failed, not setting thread name for '%s'\n", "stream writer"); + } +#endif + + while(1) { /* loop broken inside */ + d_pthread_mutex_lock(&pThis->mut); + while(pThis->iCnt == 0) { + if(pThis->bStopWriter) { + pthread_cond_broadcast(&pThis->isEmpty); + d_pthread_mutex_unlock(&pThis->mut); + goto finalize_it; /* break main loop */ + } + if(bTimedOut && pThis->iBufPtr > 0) { + /* if we timed out, we need to flush pending data */ + strmFlush(pThis); + bTimedOut = 0; + continue; /* now we should have data */ + } + bTimedOut = 0; + timeoutComp(&t, pThis->iFlushInterval * 2000); /* *1000 millisconds */ + if(pThis->bDoTimedWait) { + if(pthread_cond_timedwait(&pThis->notEmpty, &pThis->mut, &t) != 0) { + int err = errno; + if(err == ETIMEDOUT) { + bTimedOut = 1; + } else { + bTimedOut = 1; + char errStr[1024]; + rs_strerror_r(err, errStr, sizeof(errStr)); + DBGPRINTF("stream async writer timeout with error (%d): %s - ignoring\n", + err, errStr); + } + } + } else { + d_pthread_cond_wait(&pThis->notEmpty, &pThis->mut); + } + } + + bTimedOut = 0; /* we may have timed out, but there *is* work to do... */ + + iDeq = pThis->iDeq++ % STREAM_ASYNC_NUMBUFS; + doWriteInternal(pThis, pThis->asyncBuf[iDeq].pBuf, pThis->asyncBuf[iDeq].lenBuf); + // TODO: error check????? 2009-07-06 + + --pThis->iCnt; + if(pThis->iCnt < STREAM_ASYNC_NUMBUFS) { + pthread_cond_signal(&pThis->notFull); + if(pThis->iCnt == 0) + pthread_cond_broadcast(&pThis->isEmpty); + } + d_pthread_mutex_unlock(&pThis->mut); + } + +finalize_it: + ENDfunc + return NULL; /* to keep pthreads happy */ +} + + /* sync the file to disk, so that any unwritten data is persisted. This * also syncs the directory and thus makes sure that the file survives * fatal failure. Note that we do NOT return an error status if the @@ -789,9 +1007,7 @@ strmPhysWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) iWritten = lenBuf; CHKiRet(doWriteCall(pThis, pBuf, &iWritten)); - dbgoprint((obj_t*) pThis, "file %d write wrote %d bytes\n", pThis->fd, (int) iWritten); - pThis->iBufPtr = 0; pThis->iCurrOffs += iWritten; /* update user counter, if provided */ if(pThis->pUsrWCntr != NULL) @@ -839,7 +1055,7 @@ doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) /* see note in file header for the params we use with deflateInit2() */ zRet = zlibw.DeflateInit2(&zstrm, pThis->iZipLevel, Z_DEFLATED, 31, 9, Z_DEFAULT_STRATEGY); if(zRet != Z_OK) { - dbgprintf("error %d returned from zlib/deflateInit2()\n", zRet); + DBGPRINTF("error %d returned from zlib/deflateInit2()\n", zRet); ABORT_FINALIZE(RS_RET_ZLIB_ERR); } @@ -849,11 +1065,11 @@ doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) /* run deflate() on input until output buffer not full, finish compression if all of source has been read in */ do { - dbgprintf("in deflate() loop, avail_in %d, total_in %ld\n", zstrm.avail_in, zstrm.total_in); + DBGPRINTF("in deflate() loop, avail_in %d, total_in %ld\n", zstrm.avail_in, zstrm.total_in); zstrm.avail_out = pThis->sIOBufSize; zstrm.next_out = pThis->pZipBuf; zRet = zlibw.Deflate(&zstrm, Z_FINISH); /* no bad return value */ - dbgprintf("after deflate, ret %d, avail_out %d\n", zRet, zstrm.avail_out); + DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, zstrm.avail_out); assert(zRet != Z_STREAM_ERROR); /* state not clobbered */ CHKiRet(strmPhysWrite(pThis, (uchar*)pThis->pZipBuf, pThis->sIOBufSize - zstrm.avail_out)); } while (zstrm.avail_out == 0); @@ -862,7 +1078,7 @@ doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) zRet = zlibw.DeflateEnd(&zstrm); if(zRet != Z_OK) { - dbgprintf("error %d returned from zlib/deflateEnd()\n", zRet); + DBGPRINTF("error %d returned from zlib/deflateEnd()\n", zRet); ABORT_FINALIZE(RS_RET_ZLIB_ERR); } @@ -871,33 +1087,6 @@ finalize_it: } -/* write memory buffer to a stream object. - * To support direct writes of large objects, this method may be called - * with a buffer pointing to some region other than the stream buffer itself. - * However, in that case the stream buffer must be empty (strmFlush() has to - * be called before), because we would otherwise mess up with the sequence - * inside the stream. -- rgerhards, 2008-01-10 - */ -static rsRetVal -strmWriteInternal(strm_t *pThis, uchar *pBuf, size_t lenBuf) -{ - DEFiRet; - - ASSERT(pThis != NULL); - ASSERT(pBuf == pThis->pIOBuf || pThis->iBufPtr == 0); - - if(pThis->iZipLevel) { - CHKiRet(doZipWrite(pThis, pBuf, lenBuf)); - } else { - /* write without zipping */ - CHKiRet(strmPhysWrite(pThis, pBuf, lenBuf)); - } - -finalize_it: - RETiRet; -} - - /* flush stream output buffer to persistent storage. This can be called at any time * and is automatically called when the output buffer is full. * rgerhards, 2008-01-10 @@ -911,7 +1100,7 @@ strmFlush(strm_t *pThis) dbgoprint((obj_t*) pThis, "file %d flush, buflen %ld\n", pThis->fd, (long) pThis->iBufPtr); if(pThis->tOperationsMode != STREAMMODE_READ && pThis->iBufPtr > 0) { - iRet = strmWriteInternal(pThis, pThis->pIOBuf, pThis->iBufPtr); + iRet = strmSchedWrite(pThis, pThis->pIOBuf, pThis->iBufPtr); } RETiRet; @@ -964,6 +1153,12 @@ static rsRetVal strmWriteChar(strm_t *pThis, uchar c) ASSERT(pThis != NULL); + if(pThis->bAsyncWrite) + d_pthread_mutex_lock(&pThis->mut); + + if(pThis->bDisabled) + ABORT_FINALIZE(RS_RET_STREAM_DISABLED); + /* if the buffer is full, we need to flush before we can write */ if(pThis->iBufPtr == pThis->sIOBufSize) { CHKiRet(strmFlush(pThis)); @@ -973,11 +1168,18 @@ static rsRetVal strmWriteChar(strm_t *pThis, uchar c) pThis->iBufPtr++; finalize_it: + if(pThis->bAsyncWrite) + d_pthread_mutex_unlock(&pThis->mut); + RETiRet; } -/* write an integer value (actually a long) to a stream object */ +/* write an integer value (actually a long) to a stream object + * Note that we do not need to lock the mutex here, because we call + * strmWrite(), which does the lock (aka: we must not lock it, else we + * would run into a recursive lock, resulting in a deadlock!) + */ static rsRetVal strmWriteLong(strm_t *pThis, long i) { DEFiRet; @@ -993,89 +1195,65 @@ finalize_it: } -/* schedule an Apc flush request. - * rgerhards, 2009-06-15 - */ -static inline rsRetVal -scheduleFlushRequest(strm_t *pThis) -{ - apc_t *pApc; - DEFiRet; - - if(!pThis->apcRequested) { - /* we do an request only if none is yet pending */ - pThis->apcRequested = 1; - // TODO: find similar thing later CHKiRet(apc.CancelApc(pThis->apcID)); -dbgprintf("XXX: requesting to add apc!\n"); - CHKiRet(apc.Construct(&pApc)); - CHKiRet(apc.SetProcedure(pApc, (void (*)(void*, void*))flushApc)); - CHKiRet(apc.SetParam1(pApc, pThis)); - CHKiRet(apc.ConstructFinalize(pApc, &pThis->apcID)); - } - -finalize_it: - RETiRet; -} - - -/* write memory buffer to a stream object +/* write memory buffer to a stream object. + * process the data in chunks and copy it over to our buffer. The caller-provided data + * may theoritically be larger than our buffer. In that case, we do multiple copies. One + * may argue if it were more efficient to write out the caller-provided buffer in that case + * and earlier versions of rsyslog did this. However, this introduces a lot of complexity + * inside the buffered writer and potential performance bottlenecks when trying to solve + * it. Now keep in mind that we actually do (almost?) never have a case where the + * caller-provided buffer is larger than our one. So instead of optimizing a case + * which normally does not exist, we expect some degradation in its case but make us + * perform better in the regular cases. -- rgerhards, 2009-07-07 */ static rsRetVal strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) { - DEFVARS_mutexProtection_uncond; DEFiRet; - size_t iPartial; + size_t iWrite; + size_t iOffset; ASSERT(pThis != NULL); ASSERT(pBuf != NULL); -dbgprintf("strmWrite(%p, '%65.65s', %ld);, disabled %d, sizelim %ld, size %lld\n", pThis, pBuf,lenBuf, pThis->bDisabled, pThis->iSizeLimit, pThis->iCurrOffs); +//DBGPRINTF("strmWrite(%p, '%65.65s', %ld);, disabled %d, sizelim %ld, size %lld\n", pThis, pBuf,lenBuf, pThis->bDisabled, pThis->iSizeLimit, pThis->iCurrOffs); + if(pThis->bAsyncWrite) + d_pthread_mutex_lock(&pThis->mut); + if(pThis->bDisabled) ABORT_FINALIZE(RS_RET_STREAM_DISABLED); -RUNLOG_VAR("%d", pThis->iFlushInterval); - if(pThis->iFlushInterval != 0) { - BEGIN_MTX_PROTECTED_OPERATIONS_UNCOND(&pThis->mut); - } - - /* check if the to-be-written data is larger than our buffer size */ - if(lenBuf >= pThis->sIOBufSize) { - /* it is - so we do a direct write, that is most efficient. - * TODO: is it really? think about disk block sizes! - */ - CHKiRet(strmFlush(pThis)); /* we need to flush first!!! */ - CHKiRet(strmWriteInternal(pThis, pBuf, lenBuf)); - } else { - /* data fits into a buffer - we just need to see if it - * fits into the current buffer... - */ - if(pThis->iBufPtr + lenBuf > pThis->sIOBufSize) { - /* nope, so we must split it */ - iPartial = pThis->sIOBufSize - pThis->iBufPtr; /* this fits in current buf */ - if(iPartial > 0) { /* the buffer was exactly full, can not write anything! */ - memcpy(pThis->pIOBuf + pThis->iBufPtr, pBuf, iPartial); - pThis->iBufPtr += iPartial; - } + iOffset = 0; + do { + if(pThis->iBufPtr == pThis->sIOBufSize) { CHKiRet(strmFlush(pThis)); /* get a new buffer for rest of data */ - memcpy(pThis->pIOBuf, pBuf + iPartial, lenBuf - iPartial); - pThis->iBufPtr = lenBuf - iPartial; - } else { - /* we have space, so we simply copy over the string */ - memcpy(pThis->pIOBuf + pThis->iBufPtr, pBuf, lenBuf); - pThis->iBufPtr += lenBuf; } - } - - /* we ignore the outcome of scheduleFlushRequest(), as we will write the data always at - * termination. For Zip mode, it could be fatal if we write after each record. + iWrite = pThis->sIOBufSize - pThis->iBufPtr; /* this fits in current buf */ + if(iWrite > lenBuf) + iWrite = lenBuf; + memcpy(pThis->pIOBuf + pThis->iBufPtr, pBuf + iOffset, iWrite); + pThis->iBufPtr += iWrite; + iOffset += iWrite; + lenBuf -= iWrite; + } while(lenBuf > 0); + + /* now check if the buffer right at the end of the write is full and, if so, + * write it. This seems more natural than waiting (hours?) for the next message... */ - if(pThis->iFlushInterval != 0) - scheduleFlushRequest(pThis); + if(pThis->iBufPtr == pThis->sIOBufSize) { + CHKiRet(strmFlush(pThis)); /* get a new buffer for rest of data */ + } finalize_it: - if(pThis->iFlushInterval != 0) { - END_MTX_PROTECTED_OPERATIONS_UNCOND(&pThis->mut); + if(pThis->bAsyncWrite) { + if(pThis->bDoTimedWait == 0) { + /* we potentially have a partial buffer, so re-activate the + * writer thread that it can set and pick up timeouts. + */ + pThis->bDoTimedWait = 1; + pthread_cond_signal(&pThis->notEmpty); + } + d_pthread_mutex_unlock(&pThis->mut); } RETiRet; @@ -1390,7 +1568,6 @@ ENDobjQueryInterface(strm) */ BEGINObjClassInit(strm, 1, OBJ_IS_CORE_MODULE) /* request objects we use */ - CHKiRet(objUse(apc, CORE_COMPONENT)); OBJSetMethodHandler(objMethod_SERIALIZE, strmSerialize); OBJSetMethodHandler(objMethod_SETPROPERTY, strmSetProperty); diff --git a/runtime/stream.h b/runtime/stream.h index ac003c7b..64ffb6e1 100644 --- a/runtime/stream.h +++ b/runtime/stream.h @@ -87,6 +87,7 @@ typedef enum { /* when extending, do NOT change existing modes! */ STREAMMODE_WRITE_APPEND = 4 } strmMode_t; +#define STREAM_ASYNC_NUMBUFS 2 /* must be a power of 2 -- TODO: make configurable */ /* The strm_t data structure */ typedef struct strm_s { BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */ @@ -112,17 +113,32 @@ typedef struct strm_s { int fd; /* the file descriptor, -1 if closed */ int fdDir; /* the directory's descriptor, in case bSync is requested (-1 if closed) */ uchar *pszCurrFName; /* name of current file (if open) */ - uchar *pIOBuf; /* io Buffer */ + uchar *pIOBuf; /* the iobuffer currently in use to gather data */ size_t iBufPtrMax; /* current max Ptr in Buffer (if partial read!) */ size_t iBufPtr; /* pointer into current buffer */ int iUngetC; /* char set via UngetChar() call or -1 if none set */ bool bInRecord; /* if 1, indicates that we are currently writing a not-yet complete record */ + bool bInClose; /* used to break "deadly close loops", tells us we are already inside a close */ int iZipLevel; /* zip level (0..9). If 0, zip is completely disabled */ Bytef *pZipBuf; /* support for async flush procesing */ + bool bAsyncWrite; /* do asynchronous writes (always if a flush interval is given) */ + bool bStopWriter; /* shall writer thread terminate? */ + bool bDoTimedWait; /* instruct writer thread to do a times wait to support flush timeouts */ int iFlushInterval; /* flush in which interval - 0, no flushing */ apc_id_t apcID; /* id of current Apc request (used for cancelling) */ pthread_mutex_t mut;/* mutex for flush in async mode */ + pthread_cond_t notFull; + pthread_cond_t notEmpty; + pthread_cond_t isEmpty; + unsigned short iEnq; /* this MUST be unsigned as we use module arithmetic (else invalid indexing happens!) */ + unsigned short iDeq; /* this MUST be unsigned as we use module arithmetic (else invalid indexing happens!) */ + short iCnt; /* current nbr of elements in buffer */ + struct { + uchar *pBuf; + size_t lenBuf; + } asyncBuf[STREAM_ASYNC_NUMBUFS]; + pthread_t writerThreadID; int apcRequested; /* is an apc Requested? */ /* support for omfile size-limiting commands, special counters, NOT persisted! */ off_t iSizeLimit; /* file size limit, 0 = no limit */ @@ -130,6 +146,7 @@ typedef struct strm_s { bool bIsTTY; /* is this a tty file? */ } strm_t; + /* interfaces */ BEGINinterface(strm) /* name must also be changed in ENDinterface macro! */ rsRetVal (*Construct)(strm_t **ppThis); diff --git a/runtime/wti.c b/runtime/wti.c index 156d8116..abdf4add 100644 --- a/runtime/wti.c +++ b/runtime/wti.c @@ -41,7 +41,6 @@ #ifdef OS_SOLARIS # include <sched.h> -# define pthread_yield() sched_yield() #endif #include "rsyslog.h" diff --git a/runtime/wtp.c b/runtime/wtp.c index 87c3b324..596ff866 100644 --- a/runtime/wtp.c +++ b/runtime/wtp.c @@ -54,6 +54,7 @@ #include "wtp.h" #include "wti.h" #include "obj.h" +#include "unicode-helper.h" #include "glbl.h" /* static data */ @@ -422,6 +423,8 @@ wtpWrkrExecCancelCleanup(void *arg) static void * wtpWorker(void *arg) /* the arg is actually a wti object, even though we are in wtp! */ { + uchar *pszDbgHdr; + uchar thrdName[32] = "rs:"; DEFiRet; DEFVARS_mutexProtection; wti_t *pWti = (wti_t*) arg; @@ -437,7 +440,9 @@ wtpWorker(void *arg) /* the arg is actually a wti object, even though we are in # if HAVE_PRCTL && defined PR_SET_NAME /* set thread name - we ignore if the call fails, has no harsh consequences... */ - if(prctl(PR_SET_NAME, wtpGetDbgHdr(pThis), 0, 0, 0) != 0) { + pszDbgHdr = wtpGetDbgHdr(pThis); + ustrncpy(thrdName+3, pszDbgHdr, 20); + if(prctl(PR_SET_NAME, thrdName, 0, 0, 0) != 0) { DBGPRINTF("prctl failed, not setting thread name for '%s'\n", wtpGetDbgHdr(pThis)); } # endif |