diff options
Diffstat (limited to 'runtime')
55 files changed, 3544 insertions, 759 deletions
diff --git a/runtime/Makefile.am b/runtime/Makefile.am index 2e6c4041..eeb656d6 100644 --- a/runtime/Makefile.am +++ b/runtime/Makefile.am @@ -7,6 +7,7 @@ pkglib_LTLIBRARIES = librsyslog_la_SOURCES = \ rsyslog.c \ rsyslog.h \ + unicode-helper.h \ atomic.h \ syslogd-types.h \ module-template.h \ @@ -16,6 +17,8 @@ librsyslog_la_SOURCES = \ glbl.c \ conf.c \ conf.h \ + parser.h \ + parser.c \ msg.c \ msg.h \ linkedlist.c \ @@ -118,6 +121,13 @@ lmnetstrms_la_CPPFLAGS = $(PTHREADS_CFLAGS) $(RSRT_CFLAGS) lmnetstrms_la_LDFLAGS = -module -avoid-version lmnetstrms_la_LIBADD = +# generic stream server framework +pkglib_LTLIBRARIES += lmstrmsrv.la +lmstrmsrv_la_SOURCES = strmsrv.c strmsrv.h strms_sess.c strms_sess.h +lmstrmsrv_la_CPPFLAGS = $(PTHREADS_CFLAGS) $(RSRT_CFLAGS) +lmstrmsrv_la_LDFLAGS = -module -avoid-version +lmstrmsrv_la_LIBADD = + # netstream drivers # plain tcp driver - main driver diff --git a/runtime/atomic.h b/runtime/atomic.h index 10dbac90..fdf64214 100644 --- a/runtime/atomic.h +++ b/runtime/atomic.h @@ -42,6 +42,7 @@ */ #ifdef HAVE_ATOMIC_BUILTINS # define ATOMIC_INC(data) ((void) __sync_fetch_and_add(&(data), 1)) +# define ATOMIC_DEC(data) ((void) __sync_sub_and_fetch(&(data), 1)) # define ATOMIC_DEC_AND_FETCH(data) __sync_sub_and_fetch(&(data), 1) # define ATOMIC_FETCH_32BIT(data) ((unsigned) __sync_fetch_and_and(&(data), 0xffffffff)) # define ATOMIC_STORE_1_TO_32BIT(data) __sync_lock_test_and_set(&(data), 1) @@ -55,6 +56,7 @@ */ # warning "atomic builtins not available, using nul operations - rsyslogd will probably be racy!" # define ATOMIC_INC(data) (++(data)) +# define ATOMIC_DEC(data) (--(data)) # define ATOMIC_DEC_AND_FETCH(data) (--(data)) # define ATOMIC_FETCH_32BIT(data) (data) # define ATOMIC_STORE_1_TO_32BIT(data) (data) = 1 diff --git a/runtime/cfsysline.c b/runtime/cfsysline.c index c4490b48..0fb4247d 100644 --- a/runtime/cfsysline.c +++ b/runtime/cfsysline.c @@ -364,7 +364,7 @@ static rsRetVal doGetGID(uchar **pp, rsRetVal (*pSetHdlr)(void*, uid_t), void *p /* we set value via a set function */ CHKiRet(pSetHdlr(pVal, pgBuf->gr_gid)); } - dbgprintf("gid %d obtained for group '%s'\n", pgBuf->gr_gid, szName); + dbgprintf("gid %d obtained for group '%s'\n", (int) pgBuf->gr_gid, szName); } skipWhiteSpace(pp); /* skip over any whitespace */ @@ -406,7 +406,7 @@ static rsRetVal doGetUID(uchar **pp, rsRetVal (*pSetHdlr)(void*, uid_t), void *p /* we set value via a set function */ CHKiRet(pSetHdlr(pVal, ppwBuf->pw_uid)); } - dbgprintf("uid %d obtained for user '%s'\n", ppwBuf->pw_uid, szName); + dbgprintf("uid %d obtained for user '%s'\n", (int) ppwBuf->pw_uid, szName); } skipWhiteSpace(pp); /* skip over any whitespace */ @@ -462,7 +462,7 @@ getWord(uchar **pp, cstr_t **ppStrB) ASSERT(*pp != NULL); ASSERT(ppStrB != NULL); - CHKiRet(rsCStrConstruct(ppStrB)); + CHKiRet(cstrConstruct(ppStrB)); skipWhiteSpace(pp); /* skip over any whitespace */ @@ -470,9 +470,9 @@ getWord(uchar **pp, cstr_t **ppStrB) p = *pp; while(*p && !isspace((int) *p)) { - CHKiRet(rsCStrAppendChar(*ppStrB, *p++)); + CHKiRet(cstrAppendChar(*ppStrB, *p++)); } - CHKiRet(rsCStrFinish(*ppStrB)); + CHKiRet(cstrFinalize(*ppStrB)); *pp = p; @@ -506,7 +506,7 @@ static rsRetVal doGetWord(uchar **pp, rsRetVal (*pSetHdlr)(void*, uchar*), void ASSERT(*pp != NULL); CHKiRet(getWord(pp, &pStrB)); - CHKiRet(rsCStrConvSzStrAndDestruct(pStrB, &pNewVal, 0)); + CHKiRet(cstrConvSzStrAndDestruct(pStrB, &pNewVal, 0)); pStrB = NULL; /* we got the word, now set it */ @@ -525,7 +525,7 @@ static rsRetVal doGetWord(uchar **pp, rsRetVal (*pSetHdlr)(void*, uchar*), void finalize_it: if(iRet != RS_RET_OK) { if(pStrB != NULL) - rsCStrDestruct(&pStrB); + cstrDestruct(&pStrB); } RETiRet; @@ -548,7 +548,7 @@ doSyslogName(uchar **pp, rsRetVal (*pSetHdlr)(void*, int), void *pVal, syslogNam ASSERT(*pp != NULL); CHKiRet(getWord(pp, &pStrB)); /* get word */ - iNewVal = decodeSyslogName(rsCStrGetSzStr(pStrB), pNameTable); + iNewVal = decodeSyslogName(cstrGetSzStr(pStrB), pNameTable); if(pSetHdlr == NULL) { /* we should set value directly to var */ diff --git a/runtime/conf.c b/runtime/conf.c index ad2f4c15..a1d74b93 100644 --- a/runtime/conf.c +++ b/runtime/conf.c @@ -46,7 +46,9 @@ #include <glob.h> #include <sys/types.h> #ifdef HAVE_LIBGEN_H -# include <libgen.h> +# ifndef OS_SOLARIS +# include <libgen.h> +# endif #endif #include "rsyslog.h" @@ -68,6 +70,9 @@ #include "ctok.h" #include "ctok_token.h" +#ifdef OS_SOLARIS +# define NAME_MAX MAXNAMELEN +#endif /* forward definitions */ static rsRetVal cfline(uchar *line, selector_t **pfCurr); @@ -521,17 +526,15 @@ rsRetVal cflineParseTemplateName(uchar** pp, omodStringRequest_t *pOMSR, int iEn tplName = (uchar*) strdup((char*)dfltTplName); } else { /* template specified, pick it up */ - if(rsCStrConstruct(&pStrB) != RS_RET_OK) { - ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); - } + CHKiRet(cstrConstruct(&pStrB)); /* now copy the string */ while(*p && *p != '#' && !isspace((int) *p)) { - CHKiRet(rsCStrAppendChar(pStrB, *p)); + CHKiRet(cstrAppendChar(pStrB, *p)); ++p; } - CHKiRet(rsCStrFinish(pStrB)); - CHKiRet(rsCStrConvSzStrAndDestruct(pStrB, &tplName, 0)); + CHKiRet(cstrFinalize(pStrB)); + CHKiRet(cstrConvSzStrAndDestruct(pStrB, &tplName, 0)); } iRet = OMSRsetEntry(pOMSR, iEntry, tplName, iTplOpts); @@ -795,6 +798,9 @@ dbgprintf("calling expression parser, pp %p ('%s')\n", *pline, *pline); CHKiRet(ctok.Getpp(tok, pline)); CHKiRet(ctok.Destruct(&tok)); + /* debug support - print vmprg after construction (uncomment to use) */ + /* vmprgDebugPrint(f->f_filterData.f_expr->pVmprg); */ + /* we now need to skip whitespace to the action part, else we confuse * the legacy rsyslog conf parser. -- rgerhards, 2008-02-25 */ @@ -879,6 +885,8 @@ static rsRetVal cflineProcessPropFilter(uchar **pline, register selector_t *f) f->f_filterData.prop.operation = FIOP_STARTSWITH; } else if(!rsCStrOffsetSzStrCmp(pCSCompOp, iOffset, (unsigned char*) "regex", 5)) { f->f_filterData.prop.operation = FIOP_REGEX; + } else if(!rsCStrOffsetSzStrCmp(pCSCompOp, iOffset, (unsigned char*) "ereregex", 8)) { + f->f_filterData.prop.operation = FIOP_EREREGEX; } else { errmsg.LogError(0, NO_ERRCODE, "error: invalid compare operation '%s' - ignoring selector", (char*) rsCStrGetSzStrNoNULL(pCSCompOp)); @@ -915,7 +923,7 @@ static rsRetVal cflineProcessPropFilter(uchar **pline, register selector_t *f) */ static rsRetVal cflineProcessHostSelector(uchar **pline) { - rsRetVal iRet; + DEFiRet; ASSERT(pline != NULL); ASSERT(*pline != NULL); @@ -941,21 +949,20 @@ static rsRetVal cflineProcessHostSelector(uchar **pline) dbgprintf("resetting BSD-like hostname filter\n"); eDfltHostnameCmpMode = HN_NO_COMP; if(pDfltHostnameCmp != NULL) { - if((iRet = rsCStrSetSzStr(pDfltHostnameCmp, NULL)) != RS_RET_OK) - return(iRet); + CHKiRet(rsCStrSetSzStr(pDfltHostnameCmp, NULL)); } } else { dbgprintf("setting BSD-like hostname filter to '%s'\n", *pline); if(pDfltHostnameCmp == NULL) { /* create string for parser */ - if((iRet = rsCStrConstructFromszStr(&pDfltHostnameCmp, *pline)) != RS_RET_OK) - return(iRet); + CHKiRet(rsCStrConstructFromszStr(&pDfltHostnameCmp, *pline)); } else { /* string objects exists, just update... */ - if((iRet = rsCStrSetSzStr(pDfltHostnameCmp, *pline)) != RS_RET_OK) - return(iRet); + CHKiRet(rsCStrSetSzStr(pDfltHostnameCmp, *pline)); } } - return RS_RET_OK; + +finalize_it: + RETiRet; } @@ -966,7 +973,7 @@ static rsRetVal cflineProcessHostSelector(uchar **pline) */ static rsRetVal cflineProcessTagSelector(uchar **pline) { - rsRetVal iRet; + DEFiRet; ASSERT(pline != NULL); ASSERT(*pline != NULL); @@ -985,21 +992,20 @@ static rsRetVal cflineProcessTagSelector(uchar **pline) if(**pline != '\0' && **pline == '*' && *(*pline+1) == '\0') { dbgprintf("resetting programname filter\n"); if(pDfltProgNameCmp != NULL) { - if((iRet = rsCStrSetSzStr(pDfltProgNameCmp, NULL)) != RS_RET_OK) - return(iRet); + CHKiRet(rsCStrSetSzStr(pDfltProgNameCmp, NULL)); } } else { dbgprintf("setting programname filter to '%s'\n", *pline); if(pDfltProgNameCmp == NULL) { /* create string for parser */ - if((iRet = rsCStrConstructFromszStr(&pDfltProgNameCmp, *pline)) != RS_RET_OK) - return(iRet); + CHKiRet(rsCStrConstructFromszStr(&pDfltProgNameCmp, *pline)); } else { /* string objects exists, just update... */ - if((iRet = rsCStrSetSzStr(pDfltProgNameCmp, *pline)) != RS_RET_OK) - return(iRet); + CHKiRet(rsCStrSetSzStr(pDfltProgNameCmp, *pline)); } } - return RS_RET_OK; + +finalize_it: + RETiRet; } diff --git a/runtime/ctok.c b/runtime/ctok.c index ceab15bd..263e656c 100644 --- a/runtime/ctok.c +++ b/runtime/ctok.c @@ -259,12 +259,17 @@ ctokGetVar(ctok_t *pThis, ctok_token_t *pToken) } CHKiRet(rsCStrConstruct(&pstrVal)); - /* this loop is quite simple, a variable name is terminated by whitespace. */ - while(!isspace(c)) { + /* this loop is quite simple, a variable name is terminated when a non-supported + * character is detected. Note that we currently permit a numerical digit as the + * first char, which is not permitted by ABNF. -- rgerhards, 2009-03-10 + */ + while(isalpha(c) || isdigit(c) || (c == '_') || (c == '-')) { CHKiRet(rsCStrAppendChar(pstrVal, tolower(c))); CHKiRet(ctokGetCharFromStream(pThis, &c)); } - CHKiRet(rsCStrFinish(pStrB)); + CHKiRet(ctokUngetCharFromStream(pThis, c)); /* put not processed char back */ + + CHKiRet(cstrFinalize(pstrVal)); CHKiRet(var.SetString(pToken->pVar, pstrVal)); pstrVal = NULL; @@ -314,7 +319,7 @@ ctokGetSimpStr(ctok_t *pThis, ctok_token_t *pToken) } CHKiRet(ctokGetCharFromStream(pThis, &c)); } - CHKiRet(rsCStrFinish(pStrB)); + CHKiRet(cstrFinalize(pstrVal)); CHKiRet(var.SetString(pToken->pVar, pstrVal)); pstrVal = NULL; @@ -389,6 +394,7 @@ ctokGetToken(ctok_t *pThis, ctok_token_t **ppToken) uchar c; uchar szWord[128]; int bRetry = 0; /* retry parse? Only needed for inline comments... */ + cstr_t *pstrVal; ISOBJ_TYPE_assert(pThis, ctok); ASSERT(ppToken != NULL); @@ -512,7 +518,10 @@ ctokGetToken(ctok_t *pThis, ctok_token_t **ppToken) /* push c back, higher level parser needs it */ CHKiRet(ctokUngetCharFromStream(pThis, c)); pToken->tok = ctok_FUNCTION; - /* TODO: fill function name */ + /* fill function name */ + CHKiRet(rsCStrConstruct(&pstrVal)); + CHKiRet(rsCStrSetSzStr(pstrVal, szWord)); + CHKiRet(var.SetString(pToken->pVar, pstrVal)); } else { /* give up... */ dbgprintf("parser has an invalid word (token) '%s'\n", szWord); pToken->tok = ctok_INVALID; diff --git a/runtime/datetime.c b/runtime/datetime.c index 20ca6191..40ab4e9c 100644 --- a/runtime/datetime.c +++ b/runtime/datetime.c @@ -62,9 +62,14 @@ DEFobjCurrIf(errmsg) * most portable and removes the need for additional structures * (but I have to admit it is somewhat "bulky";)). * - * Obviously, all caller-provided pointers must not be NULL... + * Obviously, *t must not be NULL... + * + * rgerhards, 2008-10-07: added ttSeconds to provide a way to + * obtain the second-resolution UNIX timestamp. This is needed + * in some situations to minimize time() calls (namely when doing + * output processing). This can be left NULL if not needed. */ -static void getCurrTime(struct syslogTime *t) +static void getCurrTime(struct syslogTime *t, time_t *ttSeconds) { struct timeval tp; struct tm *tm; @@ -83,6 +88,9 @@ static void getCurrTime(struct syslogTime *t) # else gettimeofday(&tp, NULL); # endif + if(ttSeconds != NULL) + *ttSeconds = tp.tv_sec; + tm = localtime_r((time_t*) &(tp.tv_sec), &tmBuf); t->year = tm->tm_year + 1900; @@ -142,13 +150,12 @@ static void getCurrTime(struct syslogTime *t) * \retval The number parsed. */ -static int srSLMGParseInt32(char** ppsz) +static int srSLMGParseInt32(uchar** ppsz) { - int i; + register int i; i = 0; - while(isdigit((int) **ppsz)) - { + while(isdigit((int) **ppsz)) { i = i * 10 + **ppsz - '0'; ++(*ppsz); } @@ -164,9 +171,9 @@ static int srSLMGParseInt32(char** ppsz) * could be obtained (restriction added 2008-09-16 by rgerhards). */ static rsRetVal -ParseTIMESTAMP3339(struct syslogTime *pTime, char** ppszTS) +ParseTIMESTAMP3339(struct syslogTime *pTime, uchar** ppszTS) { - char *pszTS = *ppszTS; + uchar *pszTS = *ppszTS; /* variables to temporarily hold time information while we parse */ int year; int month; @@ -226,7 +233,7 @@ ParseTIMESTAMP3339(struct syslogTime *pTime, char** ppszTS) /* Now let's see if we have secfrac */ if(*pszTS == '.') { - char *pszStart = ++pszTS; + uchar *pszStart = ++pszTS; secfrac = srSLMGParseInt32(&pszTS); secfracPrecision = (int) (pszTS - pszStart); } else { @@ -299,16 +306,17 @@ finalize_it: * time() call reduction ;). */ static rsRetVal -ParseTIMESTAMP3164(struct syslogTime *pTime, char** ppszTS) +ParseTIMESTAMP3164(struct syslogTime *pTime, uchar** ppszTS) { /* variables to temporarily hold time information while we parse */ int month; int day; + int year = 0; /* 0 means no year provided */ int hour; /* 24 hour clock */ int minute; int second; /* end variables to temporarily hold time information while we parse */ - char *pszTS; + uchar *pszTS; DEFiRet; assert(ppszTS != NULL); @@ -464,7 +472,23 @@ ParseTIMESTAMP3164(struct syslogTime *pTime, char** ppszTS) if(*pszTS++ != ' ') ABORT_FINALIZE(RS_RET_INVLD_TIME); + + /* time part */ hour = srSLMGParseInt32(&pszTS); + if(hour > 1970 && hour < 2100) { + /* if so, we assume this actually is a year. This is a format found + * e.g. in Cisco devices. + * (if you read this 2100+ trying to fix a bug, congratulate me + * to how long the code survived - me no longer ;)) -- rgerhards, 2008-11-18 + */ + year = hour; + + /* re-query the hour, this time it must be valid */ + if(*pszTS++ != ' ') + ABORT_FINALIZE(RS_RET_INVLD_TIME); + hour = srSLMGParseInt32(&pszTS); + } + if(hour < 0 || hour > 23) ABORT_FINALIZE(RS_RET_INVLD_TIME); @@ -494,6 +518,8 @@ ParseTIMESTAMP3164(struct syslogTime *pTime, char** ppszTS) *ppszTS = pszTS; /* provide updated parse position back to caller */ pTime->timeType = 1; pTime->month = month; + if(year > 0) + pTime->year = year; /* persist year if detected */ pTime->day = day; pTime->hour = hour; pTime->minute = minute; diff --git a/runtime/datetime.h b/runtime/datetime.h index 2210af02..efb0a0af 100644 --- a/runtime/datetime.h +++ b/runtime/datetime.h @@ -35,9 +35,9 @@ typedef struct datetime_s { /* interfaces */ BEGINinterface(datetime) /* name must also be changed in ENDinterface macro! */ - void (*getCurrTime)(struct syslogTime *t); - rsRetVal (*ParseTIMESTAMP3339)(struct syslogTime *pTime, char** ppszTS); - rsRetVal (*ParseTIMESTAMP3164)(struct syslogTime *pTime, char** pszTS); + void (*getCurrTime)(struct syslogTime *t, time_t *ttSeconds); + rsRetVal (*ParseTIMESTAMP3339)(struct syslogTime *pTime, uchar** ppszTS); + rsRetVal (*ParseTIMESTAMP3164)(struct syslogTime *pTime, uchar** pszTS); int (*formatTimestampToMySQL)(struct syslogTime *ts, char* pDst, size_t iLenDst); int (*formatTimestampToPgSQL)(struct syslogTime *ts, char *pDst, size_t iLenDst); int (*formatTimestamp3339)(struct syslogTime *ts, char* pBuf, size_t iLenBuf); diff --git a/runtime/debug.c b/runtime/debug.c index 62e2a687..4ee90226 100644 --- a/runtime/debug.c +++ b/runtime/debug.c @@ -1,3 +1,4 @@ +#include <sys/syscall.h> /* debug.c * * This file proides debug and run time error analysis support. Some of the @@ -886,7 +887,9 @@ dbgprint(obj_t *pObj, char *pszMsg, size_t lenMsg) if(stddbg != -1) write(stddbg, pszWriteBuf, lenWriteBuf); if(altdbg != -1) write(altdbg, pszWriteBuf, lenWriteBuf); } + lenWriteBuf = snprintf(pszWriteBuf, sizeof(pszWriteBuf), "%s: ", pszThrdName); + // use for testing: lenWriteBuf = snprintf(pszWriteBuf, sizeof(pszWriteBuf), "{%ld}%s: ", (long) syscall(SYS_gettid), pszThrdName); if(stddbg != -1) write(stddbg, pszWriteBuf, lenWriteBuf); if(altdbg != -1) write(altdbg, pszWriteBuf, lenWriteBuf); /* print object name header if we have an object */ @@ -1340,7 +1343,7 @@ rsRetVal dbgClassInit(void) if(pszAltDbgFileName != NULL) { /* we have a secondary file, so let's open it) */ - if((altdbg = open(pszAltDbgFileName, O_WRONLY|O_CREAT|O_TRUNC|O_NOCTTY, S_IRUSR|S_IWUSR)) == -1) { + if((altdbg = open(pszAltDbgFileName, O_WRONLY|O_CREAT|O_TRUNC|O_NOCTTY|O_CLOEXEC, S_IRUSR|S_IWUSR)) == -1) { fprintf(stderr, "alternate debug file could not be opened, ignoring. Error: %s\n", strerror(errno)); } } diff --git a/runtime/debug.h b/runtime/debug.h index 43836024..1375493d 100644 --- a/runtime/debug.h +++ b/runtime/debug.h @@ -101,6 +101,8 @@ void dbgSetThrdName(uchar *pszName); void dbgPrintAllDebugInfo(void); /* macros */ +#define DBGPRINTF(...) if(Debug) { dbgprintf(__VA_ARGS__); } +#define DBGOPRINT(...) if(Debug) { dbgoprint(__VA_ARGS__); } #ifdef RTINST # define BEGINfunc static dbgFuncDB_t *pdbgFuncDB; int dbgCALLStaCK_POP_POINT = dbgEntrFunc(&pdbgFuncDB, __FILE__, __func__, __LINE__); # define ENDfunc dbgExitFunc(pdbgFuncDB, dbgCALLStaCK_POP_POINT, RS_RET_NO_IRET); diff --git a/runtime/expr.c b/runtime/expr.c index ee5b9e2c..e449d1c7 100644 --- a/runtime/expr.c +++ b/runtime/expr.c @@ -55,11 +55,63 @@ DEFobjCurrIf(ctok) * rgerhards, 2008-02-19 */ -/* forward definiton - thanks to recursive ABNF, we can not avoid at least one ;) */ +/* forward definition - thanks to recursive ABNF, we can not avoid at least one ;) */ static rsRetVal expr(expr_t *pThis, ctok_t *tok); static rsRetVal +function(expr_t *pThis, ctok_t *tok) +{ + DEFiRet; + ctok_token_t *pToken = NULL; + int iNumArgs = 0; + var_t *pVar; + + ISOBJ_TYPE_assert(pThis, expr); + ISOBJ_TYPE_assert(tok, ctok); + + CHKiRet(ctok.GetToken(tok, &pToken)); + /* note: pToken is destructed in finalize_it */ + + if(pToken->tok == ctok_LPAREN) { + CHKiRet(ctok_token.Destruct(&pToken)); /* token processed, "eat" it */ + CHKiRet(ctok.GetToken(tok, &pToken)); /* get next one */ + } else + ABORT_FINALIZE(RS_RET_FUNC_NO_LPAREN); + + /* we first push all the params on the stack. Then we call the function */ + while(pToken->tok != ctok_RPAREN) { + ++iNumArgs; + CHKiRet(ctok.UngetToken(tok, pToken)); /* not for us, so let others process it */ + CHKiRet(expr(pThis, tok)); + CHKiRet(ctok.GetToken(tok, &pToken)); /* get next one, needed for while() check */ + if(pToken->tok == ctok_COMMA) { + CHKiRet(ctok_token.Destruct(&pToken)); /* token processed, "eat" it */ + CHKiRet(ctok.GetToken(tok, &pToken)); /* get next one */ + if(pToken->tok == ctok_RPAREN) { + ABORT_FINALIZE(RS_RET_FUNC_MISSING_EXPR); + } + } + } + + + /* now push number of arguments - this must be on top of the stack */ + CHKiRet(var.Construct(&pVar)); + CHKiRet(var.ConstructFinalize(pVar)); + CHKiRet(var.SetNumber(pVar, iNumArgs)); + CHKiRet(vmprg.AddVarOperation(pThis->pVmprg, opcode_PUSHCONSTANT, pVar)); /* add to program */ + + +finalize_it: + if(pToken != NULL) { + ctok_token.Destruct(&pToken); /* "eat" processed token */ + } + + RETiRet; +} + + +static rsRetVal terminal(expr_t *pThis, ctok_t *tok) { DEFiRet; @@ -85,8 +137,14 @@ terminal(expr_t *pThis, ctok_t *tok) break; case ctok_FUNCTION: dbgoprint((obj_t*) pThis, "function\n"); - /* TODO: vm: call - well, need to implement that first */ - ABORT_FINALIZE(RS_RET_NOT_IMPLEMENTED); + CHKiRet(function(pThis, tok)); /* this creates the stack call frame */ + /* ... but we place the call instruction onto the stack ourselfs (because + * we have all relevant information) + */ + CHKiRet(ctok_token.UnlinkVar(pToken, &pVar)); + CHKiRet(var.ConvToString(pVar)); /* make sure we have a string */ + CHKiRet(vmprg.AddCallOperation(pThis->pVmprg, pVar->val.pStr)); /* add to program */ + CHKiRet(var.Destruct(&pVar)); break; case ctok_MSGVAR: dbgoprint((obj_t*) pThis, "MSGVAR\n"); @@ -406,6 +464,7 @@ ENDobjQueryInterface(expr) */ BEGINObjClassInit(expr, 1, OBJ_IS_CORE_MODULE) /* class, version */ /* request objects we use */ + CHKiRet(objUse(var, CORE_COMPONENT)); CHKiRet(objUse(vmprg, CORE_COMPONENT)); CHKiRet(objUse(var, CORE_COMPONENT)); CHKiRet(objUse(ctok_token, CORE_COMPONENT)); diff --git a/runtime/glbl.c b/runtime/glbl.c index 1114fcd3..28f14320 100644 --- a/runtime/glbl.c +++ b/runtime/glbl.c @@ -51,12 +51,16 @@ DEFobjStaticHelpers * class... */ static uchar *pszWorkDir = NULL; +static int bOptimizeUniProc = 1; /* enable uniprocessor optimizations */ +static int bHUPisRestart = 1; /* should SIGHUP cause a full system restart? */ +static int bPreserveFQDN = 0; /* should FQDNs always be preserved? */ static int iMaxLine = 2048; /* maximum length of a syslog message */ static int iDefPFFamily = PF_UNSPEC; /* protocol family (IPv4, IPv6 or both) */ static int bDropMalPTRMsgs = 0;/* Drop messages which have malicious PTR records during DNS lookup */ static int option_DisallowWarning = 1; /* complain if message from disallowed sender is received */ static int bDisableDNS = 0; /* don't look up IP addresses of remote messages */ static uchar *LocalHostName = NULL;/* our hostname - read-only after startup */ +static uchar *LocalFQDNName = NULL;/* our hostname as FQDN - read-only after startup */ static uchar *LocalDomain; /* our local domain name - read-only after startup */ static char **StripDomains = NULL;/* these domains may be stripped before writing logs - r/o after s.u., never touched by init */ static char **LocalHosts = NULL;/* these hosts are logged with their hostname - read-only after startup, never touched by init */ @@ -85,6 +89,9 @@ static dataType Get##nameFunc(void) \ return(nameVar); \ } +SIMP_PROP(OptimizeUniProc, bOptimizeUniProc, int) +SIMP_PROP(PreserveFQDN, bPreserveFQDN, int) +SIMP_PROP(HUPisRestart, bHUPisRestart, int) SIMP_PROP(MaxLine, iMaxLine, int) SIMP_PROP(DefPFFamily, iDefPFFamily, int) /* note that in the future we may check the family argument */ SIMP_PROP(DropMalPTRMsgs, bDropMalPTRMsgs, int) @@ -94,6 +101,7 @@ SIMP_PROP(LocalDomain, LocalDomain, uchar*) SIMP_PROP(StripDomains, StripDomains, char**) SIMP_PROP(LocalHosts, LocalHosts, char**) +SIMP_PROP_SET(LocalFQDNName, LocalFQDNName, uchar*) SIMP_PROP_SET(LocalHostName, LocalHostName, uchar*) SIMP_PROP_SET(DfltNetstrmDrvr, pszDfltNetstrmDrvr, uchar*) /* TODO: use custom function which frees existing value */ SIMP_PROP_SET(DfltNetstrmDrvrCAF, pszDfltNetstrmDrvrCAF, uchar*) /* TODO: use custom function which frees existing value */ @@ -110,7 +118,27 @@ SIMP_PROP_SET(DfltNetstrmDrvrCertFile, pszDfltNetstrmDrvrCertFile, uchar*) /* TO static uchar* GetLocalHostName(void) { - return(LocalHostName == NULL ? (uchar*) "[localhost]" : LocalHostName); + uchar *pszRet; + + if(LocalHostName == NULL) + pszRet = (uchar*) "[localhost]"; + else { + if(GetPreserveFQDN() == 1) + pszRet = LocalFQDNName; + else + pszRet = LocalHostName; + } + return(pszRet); +} + + +/* return the current localhost name as FQDN (requires FQDN to be set) + * TODO: we should set the FQDN ourselfs in here! + */ +static uchar* +GetLocalFQDNName(void) +{ + return(LocalFQDNName == NULL ? (uchar*) "[localhost]" : LocalFQDNName); } @@ -173,10 +201,14 @@ CODESTARTobjQueryInterface(glbl) pIf->Get##name = Get##name; \ pIf->Set##name = Set##name; SIMP_PROP(MaxLine); + SIMP_PROP(OptimizeUniProc); + SIMP_PROP(PreserveFQDN); + SIMP_PROP(HUPisRestart); SIMP_PROP(DefPFFamily); SIMP_PROP(DropMalPTRMsgs); SIMP_PROP(Option_DisallowWarning); SIMP_PROP(DisableDNS); + SIMP_PROP(LocalFQDNName) SIMP_PROP(LocalHostName) SIMP_PROP(LocalDomain) SIMP_PROP(StripDomains) @@ -216,6 +248,9 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a pszWorkDir = NULL; } bDropMalPTRMsgs = 0; + bOptimizeUniProc = 1; + bHUPisRestart = 1; + bPreserveFQDN = 0; return RS_RET_OK; } @@ -235,6 +270,9 @@ BEGINAbstractObjClassInit(glbl, 1, OBJ_IS_CORE_MODULE) /* class, version */ CHKiRet(regCfSysLineHdlr((uchar *)"defaultnetstreamdrivercafile", 0, eCmdHdlrGetWord, NULL, &pszDfltNetstrmDrvrCAF, NULL)); CHKiRet(regCfSysLineHdlr((uchar *)"defaultnetstreamdriverkeyfile", 0, eCmdHdlrGetWord, NULL, &pszDfltNetstrmDrvrKeyFile, NULL)); CHKiRet(regCfSysLineHdlr((uchar *)"defaultnetstreamdrivercertfile", 0, eCmdHdlrGetWord, NULL, &pszDfltNetstrmDrvrCertFile, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"optimizeforuniprocessor", 0, eCmdHdlrBinary, NULL, &bOptimizeUniProc, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"hupisrestart", 0, eCmdHdlrBinary, NULL, &bHUPisRestart, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"preservefqdn", 0, eCmdHdlrBinary, NULL, &bPreserveFQDN, NULL)); CHKiRet(regCfSysLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler, resetConfigVariables, NULL, NULL)); ENDObjClassInit(glbl) @@ -255,6 +293,8 @@ BEGINObjClassExit(glbl, OBJ_IS_CORE_MODULE) /* class, version */ free(pszWorkDir); if(LocalHostName != NULL) free(LocalHostName); + if(LocalFQDNName != NULL) + free(LocalFQDNName); ENDObjClassExit(glbl) /* vi:set ai: diff --git a/runtime/glbl.h b/runtime/glbl.h index 0c83bdd5..5bdf4f57 100644 --- a/runtime/glbl.h +++ b/runtime/glbl.h @@ -41,10 +41,14 @@ BEGINinterface(glbl) /* name must also be changed in ENDinterface macro! */ dataType (*Get##name)(void); \ rsRetVal (*Set##name)(dataType); SIMP_PROP(MaxLine, int) + SIMP_PROP(OptimizeUniProc, int) + SIMP_PROP(HUPisRestart, int) + SIMP_PROP(PreserveFQDN, int) SIMP_PROP(DefPFFamily, int) SIMP_PROP(DropMalPTRMsgs, int) SIMP_PROP(Option_DisallowWarning, int) SIMP_PROP(DisableDNS, int) + SIMP_PROP(LocalFQDNName, uchar*) SIMP_PROP(LocalHostName, uchar*) SIMP_PROP(LocalDomain, uchar*) SIMP_PROP(StripDomains, char**) @@ -55,7 +59,8 @@ BEGINinterface(glbl) /* name must also be changed in ENDinterface macro! */ SIMP_PROP(DfltNetstrmDrvrCertFile, uchar*) #undef SIMP_PROP ENDinterface(glbl) -#define glblCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */ +#define glblCURR_IF_VERSION 2 /* increment whenever you change the interface structure! */ +/* version 2 had PreserveFQDN added - rgerhards, 2008-12-08 */ /* the remaining prototypes */ PROTOTYPEObj(glbl); diff --git a/runtime/module-template.h b/runtime/module-template.h index eb39b587..3e963199 100644 --- a/runtime/module-template.h +++ b/runtime/module-template.h @@ -39,7 +39,8 @@ #define DEF_OMOD_STATIC_DATA \ DEF_MOD_STATIC_DATA \ - DEFobjCurrIf(obj) + DEFobjCurrIf(obj) \ + static __attribute__((unused)) int bCoreSupportsBatching; #define DEF_IMOD_STATIC_DATA \ DEF_MOD_STATIC_DATA \ DEFobjCurrIf(obj) @@ -160,6 +161,37 @@ static rsRetVal isCompatibleWithFeature(syslogFeature __attribute__((unused)) eF RETiRet;\ } + +/* beginTransaction() + * introduced in v4.3.3 -- rgerhards, 2009-04-27 + */ +#define BEGINbeginTransaction \ +static rsRetVal beginTransaction(instanceData __attribute__((unused)) *pData)\ +{\ + DEFiRet; + +#define CODESTARTbeginTransaction /* currently empty, but may be extended */ + +#define ENDbeginTransaction \ + RETiRet;\ +} + + +/* endTransaction() + * introduced in v4.3.3 -- rgerhards, 2009-04-27 + */ +#define BEGINendTransaction \ +static rsRetVal endTransaction(instanceData __attribute__((unused)) *pData)\ +{\ + DEFiRet; + +#define CODESTARTendTransaction /* currently empty, but may be extended */ + +#define ENDendTransaction \ + RETiRet;\ +} + + /* doAction() */ #define BEGINdoAction \ @@ -324,6 +356,18 @@ static rsRetVal queryEtryPt(uchar *name, rsRetVal (**pEtryPoint)())\ *pEtryPoint = tryResume;\ } + +/* the following definition is queryEtryPt block that must be added + * if an output module supports the transactional interface. + * rgerhards, 2009-04-27 + */ +#define CODEqueryEtryPt_TXIF_OMOD_QUERIES \ + else if(!strcmp((char*) name, "beginTransaction")) {\ + *pEtryPoint = beginTransaction;\ + } else if(!strcmp((char*) name, "endTransaction")) {\ + *pEtryPoint = endTransaction;\ + } + /* the following definition is the standard block for queryEtryPt for INPUT * modules. This can be used if no specific handling (e.g. to cover version * differences) is needed. @@ -393,6 +437,32 @@ finalize_it:\ } +/* now come some check functions, which enable a standard way of obtaining feature + * information from the core. feat is the to-be-tested feature and featVar is a + * variable that receives the result (0-not support, 1-supported). + * This must be a macro, so that it is put into the output's code. Otherwise, we + * would need to rely on a library entry point, which is what we intend to avoid ;) + * rgerhards, 2009-04-27 + */ +#define INITChkCoreFeature(featVar, feat) \ +{ \ + rsRetVal MACRO_Ret; \ + rsRetVal (*pQueryCoreFeatureSupport)(int*, unsigned); \ + int bSupportsIt; \ + featVar = 0; \ + MACRO_Ret = pHostQueryEtryPt((uchar*)"queryCoreFeatureSupport", &pQueryCoreFeatureSupport); \ + if(MACRO_Ret == RS_RET_OK) { \ + /* found entry point, so let's see if core supports it */ \ + CHKiRet((*pQueryCoreFeatureSupport)(&bSupportsIt, feat)); \ + if(bSupportsIt) \ + featVar = 1; \ + } else if(MACRO_Ret != RS_RET_ENTRY_POINT_NOT_FOUND) { \ + ABORT_FINALIZE(MACRO_Ret); /* Something else went wrong, what is not acceptable */ \ + } \ +} + + + /* definitions for host API queries */ #define CODEmodInit_QueryRegCFSLineHdlr \ CHKiRet(pHostQueryEtryPt((uchar*)"regCfSysLineHdlr", &omsdRegCFSLineHdlr)); @@ -481,6 +551,33 @@ static rsRetVal afterRun(void)\ } -/* - * vi:set ai: +/* doHUP() + * This function is optional. Currently, it is available to output plugins + * only, but may be made available to other types of plugins in the future. + * A plugin does not need to define this entry point. If if does, it gets + * called when a non-restart type of HUP is done. A plugin should register + * this function so that it can close files, connection or other ressources + * on HUP - if it can be assume the user wanted to do this as a part of HUP + * processing. Note that the name "HUP" has historical reasons, it stems back + * to the infamous SIGHUP which was sent to restart a syslogd. We still retain + * that legacy, but may move this to a different signal. + * rgerhards, 2008-10-22 + */ +#define CODEqueryEtryPt_doHUP \ + else if(!strcmp((char*) name, "doHUP")) {\ + *pEtryPoint = doHUP;\ + } +#define BEGINdoHUP \ +static rsRetVal doHUP(instanceData __attribute__((unused)) *pData)\ +{\ + DEFiRet; + +#define CODESTARTdoHUP + +#define ENDdoHUP \ + RETiRet;\ +} + + +/* vim:set ai: */ diff --git a/runtime/modules.c b/runtime/modules.c index d5730ede..32ae659f 100644 --- a/runtime/modules.c +++ b/runtime/modules.c @@ -40,6 +40,7 @@ #include <time.h> #include <assert.h> #include <errno.h> +#include <pthread.h> #ifdef OS_BSD # include "libgen.h" #endif @@ -49,6 +50,10 @@ #include <unistd.h> #include <sys/file.h> +#ifdef OS_SOLARIS +# define PATH_MAX MAXPATHLEN +#endif + #include "cfsysline.h" #include "modules.h" #include "errmsg.h" @@ -57,6 +62,14 @@ DEFobjStaticHelpers DEFobjCurrIf(errmsg) +/* we must ensure that only one thread at one time tries to load or unload + * modules, otherwise we may see race conditions. This first came up with + * imdiag/imtcp, which both use the same stream drivers. Below is the mutex + * for that handling. + * rgerhards, 2009-05-25 + */ +static pthread_mutex_t mutLoadUnload; + static modInfo_t *pLoadedModules = NULL; /* list of currently-loaded modules */ static modInfo_t *pLoadedModulesLast = NULL; /* tail-pointer */ @@ -221,6 +234,8 @@ static rsRetVal queryHostEtryPt(uchar *name, rsRetVal (**pEtryPoint)()) *pEtryPoint = regCfSysLineHdlr; } else if(!strcmp((char*) name, "objGetObjInterface")) { *pEtryPoint = objGetObjInterface; + } else if(!strcmp((char*) name, "OMSRgetSupportedTplOpts")) { + *pEtryPoint = OMSRgetSupportedTplOpts; } else { *pEtryPoint = NULL; /* to be on the safe side */ ABORT_FINALIZE(RS_RET_ENTRY_POINT_NOT_FOUND); @@ -347,6 +362,7 @@ static rsRetVal doModInit(rsRetVal (*modInit)(int, int*, rsRetVal(**)(), rsRetVal(*)(), modInfo_t*), uchar *name, void *pModHdlr) { DEFiRet; + rsRetVal localRet; modInfo_t *pNew = NULL; rsRetVal (*modGetType)(eModType_t *pType); @@ -391,6 +407,10 @@ doModInit(rsRetVal (*modInit)(int, int*, rsRetVal(**)(), rsRetVal(*)(), modInfo_ CHKiRet((*pNew->modQueryEtryPt)((uchar*)"parseSelectorAct", &pNew->mod.om.parseSelectorAct)); CHKiRet((*pNew->modQueryEtryPt)((uchar*)"isCompatibleWithFeature", &pNew->isCompatibleWithFeature)); CHKiRet((*pNew->modQueryEtryPt)((uchar*)"tryResume", &pNew->tryResume)); + /* try load optional interfaces */ + localRet = (*pNew->modQueryEtryPt)((uchar*)"doHUP", &pNew->doHUP); + if(localRet != RS_RET_OK && localRet != RS_RET_MODULE_ENTRY_POINT_NOT_FOUND) + ABORT_FINALIZE(localRet); break; case eMOD_LIB: break; @@ -468,6 +488,8 @@ modUnlinkAndDestroy(modInfo_t **ppThis) pThis = *ppThis; assert(pThis != NULL); + pthread_mutex_lock(&mutLoadUnload); + /* first check if we are permitted to unload */ if(pThis->eType == eMOD_LIB) { if(pThis->uRefCnt > 0) { @@ -502,6 +524,7 @@ modUnlinkAndDestroy(modInfo_t **ppThis) moduleDestruct(pThis); finalize_it: + pthread_mutex_unlock(&mutLoadUnload); RETiRet; } @@ -576,6 +599,8 @@ Load(uchar *pModName) assert(pModName != NULL); dbgprintf("Requested to load module '%s'\n", pModName); + pthread_mutex_lock(&mutLoadUnload); + iModNameLen = strlen((char *) pModName); if(iModNameLen > 3 && !strcmp((char *) pModName + iModNameLen - 3, ".so")) { iModNameLen -= 3; @@ -599,7 +624,7 @@ Load(uchar *pModName) iLoadCnt = 0; do { /* now build our load module name */ - if(*pModName == '/') { + if(*pModName == '/' || *pModName == '.') { *szPath = '\0'; /* we do not need to append the path - its already in the module name */ iPathLen = 0; } else { @@ -685,6 +710,7 @@ Load(uchar *pModName) } finalize_it: + pthread_mutex_unlock(&mutLoadUnload); RETiRet; } @@ -780,6 +806,15 @@ BEGINObjClassExit(module, OBJ_IS_LOADABLE_MODULE) /* CHANGE class also in END MA CODESTARTObjClassExit(module) /* release objects we no longer need */ objRelease(errmsg, CORE_COMPONENT); + /* We have a problem in our reference counting, which leads to this function + * being called too early. This usually is no problem, but if we destroy + * the mutex object, we get into trouble. So rather than finding the root cause, + * we do not release the mutex right now and have a very, very slight leak. + * We know that otherwise no bad effects happen, so this acceptable for the + * time being. -- rgerhards, 2009-05-25 + * + * TODO: add again: pthread_mutex_destroy(&mutLoadUnload); + */ # ifdef DEBUG modUsrPrintAll(); /* debug aid - TODO: integrate with debug.c, at least the settings! */ @@ -822,6 +857,7 @@ ENDobjQueryInterface(module) */ BEGINAbstractObjClassInit(module, 1, OBJ_IS_CORE_MODULE) /* class, version - CHANGE class also in END MACRO! */ uchar *pModPath; + pthread_mutexattr_t mutAttr; /* use any module load path specified in the environment */ if((pModPath = (uchar*) getenv("RSYSLOG_MODDIR")) != NULL) { @@ -839,6 +875,10 @@ BEGINAbstractObjClassInit(module, 1, OBJ_IS_CORE_MODULE) /* class, version - CHA SetModDir(glblModPath); } + pthread_mutexattr_init(&mutAttr); + pthread_mutexattr_settype(&mutAttr, PTHREAD_MUTEX_RECURSIVE); + pthread_mutex_init(&mutLoadUnload, &mutAttr); + /* request objects we use */ CHKiRet(objUse(errmsg, CORE_COMPONENT)); ENDObjClassInit(module) diff --git a/runtime/modules.h b/runtime/modules.h index 7d34bcf7..372529ee 100644 --- a/runtime/modules.h +++ b/runtime/modules.h @@ -44,8 +44,11 @@ * rgerhards, 2008-03-04 * version 3 adds modInfo_t ptr to call of modInit -- rgerhards, 2008-03-10 * version 4 removes needUDPSocket OM callback -- rgerhards, 2008-03-22 + * version 5 changes the way parsing works for input modules. This is + * an important change, parseAndSubmitMessage() goes away. Other + * module types are not affected. -- rgerhards, 2008-10-09 */ -#define CURR_MOD_IF_VERSION 4 +#define CURR_MOD_IF_VERSION 5 typedef enum eModType_ { eMOD_IN, /* input module */ @@ -88,6 +91,7 @@ typedef struct modInfo_s { rsRetVal (*tryResume)(void*);/* called to see if module actin can be resumed now */ rsRetVal (*modExit)(void); /* called before termination or module unload */ rsRetVal (*modGetID)(void **); /* get its unique ID from module */ + rsRetVal (*doHUP)(void *); /* non-restart type HUP handler */ /* below: parse a configuration line - return if processed * or not. If not, must be parsed to next module. */ diff --git a/runtime/msg.c b/runtime/msg.c index 375b9861..e26fe5ee 100644 --- a/runtime/msg.c +++ b/runtime/msg.c @@ -45,6 +45,7 @@ #include "glbl.h" #include "regexp.h" #include "atomic.h" +#include "unicode-helper.h" /* static data */ DEFobjStaticHelpers @@ -103,6 +104,7 @@ static syslogCODE rs_facilitynames[] = /* some forward declarations */ static int getAPPNAMELen(msg_t *pM); +static int getProtocolVersion(msg_t *pM); /* The following functions will support advanced output module * multithreading, once this is implemented. Currently, we @@ -140,8 +142,8 @@ void (*funcMsgPrepareEnqueue)(msg_t *pMsg); #define MsgLock(pMsg) funcLock(pMsg) #define MsgUnlock(pMsg) funcUnlock(pMsg) #else -#define MsgLock(pMsg) {dbgprintf("line %d\n - ", __LINE__); funcLock(pMsg);; } -#define MsgUnlock(pMsg) {dbgprintf("line %d - ", __LINE__); funcUnlock(pMsg); } +#define MsgLock(pMsg) {dbgprintf("MsgLock line %d\n - ", __LINE__); funcLock(pMsg);; } +#define MsgUnlock(pMsg) {dbgprintf("MsgUnlock line %d - ", __LINE__); funcUnlock(pMsg); } #endif /* the next function is a dummy to be used by the looking functions @@ -242,12 +244,21 @@ rsRetVal MsgEnableThreadSafety(void) /* end locking functions */ -/* "Constructor" for a msg "object". Returns a pointer to +/* This is common code for all Constructors. It is defined in an + * inline'able function so that we can save a function call in the + * actual constructors (otherwise, the msgConstruct would need + * to call msgConstructWithTime(), which would require a + * function call). Now, both can use this inline function. This + * enables us to be optimal, but still have the code just once. * the new object or NULL if no such object could be allocated. * An object constructed via this function should only be destroyed - * via "msgDestruct()". + * via "msgDestruct()". This constructor does not query system time + * itself but rather uses a user-supplied value. This enables the caller + * to do some tricks to save processing time (done, for example, in the + * udp input). + * rgerhards, 2008-10-06 */ -rsRetVal msgConstruct(msg_t **ppThis) +static inline rsRetVal msgBaseConstruct(msg_t **ppThis) { DEFiRet; msg_t *pM; @@ -260,21 +271,59 @@ rsRetVal msgConstruct(msg_t **ppThis) pM->iRefCount = 1; pM->iSeverity = -1; pM->iFacility = -1; + objConstructSetObjInfo(pM); + + /* DEV debugging only! dbgprintf("msgConstruct\t0x%x, ref 1\n", (int)pM);*/ + + *ppThis = pM; +finalize_it: + RETiRet; +} + + +/* "Constructor" for a msg "object". Returns a pointer to + * the new object or NULL if no such object could be allocated. + * An object constructed via this function should only be destroyed + * via "msgDestruct()". This constructor does not query system time + * itself but rather uses a user-supplied value. This enables the caller + * to do some tricks to save processing time (done, for example, in the + * udp input). + * rgerhards, 2008-10-06 + */ +rsRetVal msgConstructWithTime(msg_t **ppThis, struct syslogTime *stTime, time_t ttGenTime) +{ + DEFiRet; + + CHKiRet(msgBaseConstruct(ppThis)); + (*ppThis)->ttGenTime = ttGenTime; + memcpy(&(*ppThis)->tRcvdAt, stTime, sizeof(struct syslogTime)); + memcpy(&(*ppThis)->tTIMESTAMP, stTime, sizeof(struct syslogTime)); + +finalize_it: + RETiRet; +} + + +/* "Constructor" for a msg "object". Returns a pointer to + * the new object or NULL if no such object could be allocated. + * An object constructed via this function should only be destroyed + * via "msgDestruct()". This constructor, for historical reasons, + * also sets the two timestamps to the current time. + */ +rsRetVal msgConstruct(msg_t **ppThis) +{ + DEFiRet; + + CHKiRet(msgBaseConstruct(ppThis)); /* we initialize both timestamps to contain the current time, so that they * are consistent. Also, this saves us from doing any further time calls just * to obtain a timestamp. The memcpy() should not really make a difference, * especially as I think there is no codepath currently where it would not be * required (after I have cleaned up the pathes ;)). -- rgerhards, 2008-10-02 */ - datetime.getCurrTime(&(pM->tRcvdAt)); - memcpy(&pM->tTIMESTAMP, &pM->tRcvdAt, sizeof(struct syslogTime)); - - objConstructSetObjInfo(pM); - - /* DEV debugging only! dbgprintf("msgConstruct\t0x%x, ref 1\n", (int)pM);*/ - - *ppThis = pM; + datetime.getCurrTime(&((*ppThis)->tRcvdAt), &((*ppThis)->ttGenTime)); + memcpy(&(*ppThis)->tTIMESTAMP, &(*ppThis)->tRcvdAt, sizeof(struct syslogTime)); finalize_it: RETiRet; @@ -294,52 +343,27 @@ CODESTARTobjDestruct(msg) if(currRefCount == 0) { /* DEV Debugging Only! dbgprintf("msgDestruct\t0x%lx, RefCount now 0, doing DESTROY\n", (unsigned long)pThis); */ - if(pThis->pszUxTradMsg != NULL) - free(pThis->pszUxTradMsg); - if(pThis->pszRawMsg != NULL) - free(pThis->pszRawMsg); - if(pThis->pszTAG != NULL) - free(pThis->pszTAG); - if(pThis->pszHOSTNAME != NULL) - free(pThis->pszHOSTNAME); - if(pThis->pszInputName != NULL) - free(pThis->pszInputName); - if(pThis->pszRcvFrom != NULL) - free(pThis->pszRcvFrom); - if(pThis->pszRcvFromIP != NULL) - free(pThis->pszRcvFromIP); - if(pThis->pszMSG != NULL) - free(pThis->pszMSG); - if(pThis->pszFacility != NULL) - free(pThis->pszFacility); - if(pThis->pszFacilityStr != NULL) - free(pThis->pszFacilityStr); - if(pThis->pszSeverity != NULL) - free(pThis->pszSeverity); - if(pThis->pszSeverityStr != NULL) - free(pThis->pszSeverityStr); - if(pThis->pszRcvdAt3164 != NULL) - free(pThis->pszRcvdAt3164); - if(pThis->pszRcvdAt3339 != NULL) - free(pThis->pszRcvdAt3339); - if(pThis->pszRcvdAt_SecFrac != NULL) - free(pThis->pszRcvdAt_SecFrac); - if(pThis->pszRcvdAt_MySQL != NULL) - free(pThis->pszRcvdAt_MySQL); - if(pThis->pszRcvdAt_PgSQL != NULL) - free(pThis->pszRcvdAt_PgSQL); - if(pThis->pszTIMESTAMP3164 != NULL) - free(pThis->pszTIMESTAMP3164); - if(pThis->pszTIMESTAMP3339 != NULL) - free(pThis->pszTIMESTAMP3339); - if(pThis->pszTIMESTAMP_SecFrac != NULL) - free(pThis->pszTIMESTAMP_SecFrac); - if(pThis->pszTIMESTAMP_MySQL != NULL) - free(pThis->pszTIMESTAMP_MySQL); - if(pThis->pszTIMESTAMP_PgSQL != NULL) - free(pThis->pszTIMESTAMP_PgSQL); - if(pThis->pszPRI != NULL) - free(pThis->pszPRI); + free(pThis->pszRawMsg); + free(pThis->pszTAG); + free(pThis->pszHOSTNAME); + free(pThis->pszInputName); + free(pThis->pszRcvFrom); + free(pThis->pszRcvFromIP); + free(pThis->pszMSG); + free(pThis->pszFacility); + free(pThis->pszFacilityStr); + free(pThis->pszSeverity); + free(pThis->pszSeverityStr); + free(pThis->pszRcvdAt3164); + free(pThis->pszRcvdAt3339); + free(pThis->pszRcvdAt_SecFrac); + free(pThis->pszRcvdAt_MySQL); + free(pThis->pszRcvdAt_PgSQL); + free(pThis->pszTIMESTAMP3164); + free(pThis->pszTIMESTAMP3339); + free(pThis->pszTIMESTAMP_SecFrac); + free(pThis->pszTIMESTAMP_MySQL); + free(pThis->pszTIMESTAMP_PgSQL); if(pThis->pCSProgName != NULL) rsCStrDestruct(&pThis->pCSProgName); if(pThis->pCSStrucData != NULL) @@ -402,7 +426,7 @@ msg_t* MsgDup(msg_t* pOld) assert(pOld != NULL); BEGINfunc - if(msgConstruct(&pNew) != RS_RET_OK) { + if(msgConstructWithTime(&pNew, &pOld->tTIMESTAMP, pOld->ttGenTime) != RS_RET_OK) { return NULL; } @@ -413,16 +437,19 @@ msg_t* MsgDup(msg_t* pOld) pNew->bParseHOSTNAME = pOld->bParseHOSTNAME; pNew->msgFlags = pOld->msgFlags; pNew->iProtocolVersion = pOld->iProtocolVersion; - memcpy(&pNew->tRcvdAt, &pOld->tRcvdAt, sizeof(struct syslogTime)); - memcpy(&pNew->tTIMESTAMP, &pOld->tTIMESTAMP, sizeof(struct syslogTime)); + pNew->ttGenTime = pOld->ttGenTime; + /* enable this, if someone actually uses UxTradMsg, delete after some time has + * passed and nobody complained -- rgerhards, 2009-06-16 + pNew->offAfterPRI = pOld->offAfterPRI; + */ + memcpy(pNew->bufPRI, pOld->bufPRI, pOld->iLenPRI); + pNew->iLenPRI = pOld->iLenPRI; tmpCOPYSZ(Severity); tmpCOPYSZ(SeverityStr); tmpCOPYSZ(Facility); tmpCOPYSZ(FacilityStr); - tmpCOPYSZ(PRI); tmpCOPYSZ(RawMsg); tmpCOPYSZ(MSG); - tmpCOPYSZ(UxTradMsg); tmpCOPYSZ(TAG); tmpCOPYSZ(HOSTNAME); tmpCOPYSZ(RcvFrom); @@ -468,12 +495,16 @@ static rsRetVal MsgSerialize(msg_t *pThis, strm_t *pStrm) objSerializeSCALAR(pStrm, iSeverity, SHORT); objSerializeSCALAR(pStrm, iFacility, SHORT); objSerializeSCALAR(pStrm, msgFlags, INT); + objSerializeSCALAR(pStrm, ttGenTime, INT); objSerializeSCALAR(pStrm, tRcvdAt, SYSLOGTIME); objSerializeSCALAR(pStrm, tTIMESTAMP, SYSLOGTIME); + /* enable this, if someone actually uses UxTradMsg, delete after some time has + * passed and nobody complained -- rgerhards, 2009-06-16 + objSerializeSCALAR(pStrm, offsAfterPRI, SHORT); + */ objSerializePTR(pStrm, pszRawMsg, PSZ); objSerializePTR(pStrm, pszMSG, PSZ); - objSerializePTR(pStrm, pszUxTradMsg, PSZ); objSerializePTR(pStrm, pszTAG, PSZ); objSerializePTR(pStrm, pszHOSTNAME, PSZ); objSerializePTR(pStrm, pszInputName, PSZ); @@ -564,7 +595,7 @@ static rsRetVal aquirePROCIDFromTAG(msg_t *pM) } /* OK, finaally we could obtain a PROCID. So let's use it ;) */ - CHKiRet(rsCStrFinish(pM->pCSPROCID)); + CHKiRet(cstrFinalize(pM->pCSPROCID)); finalize_it: RETiRet; @@ -606,7 +637,7 @@ static rsRetVal aquireProgramName(msg_t *pM) ; ++i) { CHKiRet(rsCStrAppendChar(pM->pCSProgName, pM->pszTAG[i])); } - CHKiRet(rsCStrFinish(pM->pCSProgName)); + CHKiRet(cstrFinalize(pM->pCSProgName)); } finalize_it: RETiRet; @@ -647,7 +678,7 @@ void setProtocolVersion(msg_t *pM, int iNewVersion) pM->iProtocolVersion = iNewVersion; } -int getProtocolVersion(msg_t *pM) +static int getProtocolVersion(msg_t *pM) { assert(pM != NULL); return(pM->iProtocolVersion); @@ -666,7 +697,7 @@ int getMSGLen(msg_t *pM) } -char *getRawMsg(msg_t *pM) +static char *getRawMsg(msg_t *pM) { if(pM == NULL) return ""; @@ -677,16 +708,17 @@ char *getRawMsg(msg_t *pM) return (char*)pM->pszRawMsg; } + +/* enable this, if someone actually uses UxTradMsg, delete after some time has + * passed and nobody complained -- rgerhards, 2009-06-16 char *getUxTradMsg(msg_t *pM) { if(pM == NULL) return ""; else - if(pM->pszUxTradMsg == NULL) - return ""; - else - return (char*)pM->pszUxTradMsg; + return (char*)pM->pszRawMsg + pM->offAfterPRI; } +*/ char *getMSG(msg_t *pM) { @@ -700,43 +732,36 @@ char *getMSG(msg_t *pM) } -/* Get PRI value in text form */ -char *getPRI(msg_t *pM) +/* Get PRI value as integer */ +static int getPRIi(msg_t *pM) { - int pri; + assert(pM != NULL); + return (pM->iFacility << 3) + (pM->iSeverity); +} + +/* Get PRI value in text form */ +static inline char *getPRI(msg_t *pM) +{ if(pM == NULL) return ""; + /* there are some cases where bufPRI may not contain a valid string, + * and then we need to build it. + */ MsgLock(pM); - if(pM->pszPRI == NULL) { - /* OK, we need to construct it... we use a 5 byte buffer - as of - * RFC 3164, it can't be longer. Should it still be, snprintf will truncate... - * Note that we do not use the LOG_MAKEPRI macro. This macro - * is a simple add of the two values under FreeBSD 7. So we implement - * the logic in our own code. This is a change from a bug - * report. -- rgerhards, 2008-07-14 - */ - pri = pM->iFacility * 8 + pM->iSeverity; - if((pM->pszPRI = malloc(5)) == NULL) return ""; - pM->iLenPRI = snprintf((char*)pM->pszPRI, 5, "%d", pri); + if(pM->bufPRI[0] == '\0') { + snprintf((char*)pM->bufPRI, sizeof(pM->bufPRI), "%d", getPRIi(pM)); } MsgUnlock(pM); - return (char*)pM->pszPRI; -} - - -/* Get PRI value as integer */ -int getPRIi(msg_t *pM) -{ - assert(pM != NULL); - return (pM->iFacility << 3) + (pM->iSeverity); + return (char*)pM->bufPRI; } -char *getTimeReported(msg_t *pM, enum tplFormatTypes eFmt) +static inline char *getTimeReported(msg_t *pM, enum tplFormatTypes eFmt) { + BEGINfunc if(pM == NULL) return ""; @@ -808,11 +833,13 @@ char *getTimeReported(msg_t *pM, enum tplFormatTypes eFmt) MsgUnlock(pM); return(pM->pszTIMESTAMP_SecFrac); } + ENDfunc return "INVALID eFmt OPTION!"; } -char *getTimeGenerated(msg_t *pM, enum tplFormatTypes eFmt) +static inline char *getTimeGenerated(msg_t *pM, enum tplFormatTypes eFmt) { + BEGINfunc if(pM == NULL) return ""; @@ -884,11 +911,12 @@ char *getTimeGenerated(msg_t *pM, enum tplFormatTypes eFmt) MsgUnlock(pM); return(pM->pszRcvdAt_SecFrac); } + ENDfunc return "INVALID eFmt OPTION!"; } -char *getSeverity(msg_t *pM) +static inline char *getSeverity(msg_t *pM) { if(pM == NULL) return ""; @@ -905,7 +933,7 @@ char *getSeverity(msg_t *pM) } -char *getSeverityStr(msg_t *pM) +static inline char *getSeverityStr(msg_t *pM) { syslogCODE *c; int val; @@ -935,7 +963,7 @@ char *getSeverityStr(msg_t *pM) return((char*)pM->pszSeverityStr); } -char *getFacility(msg_t *pM) +static inline char *getFacility(msg_t *pM) { if(pM == NULL) return ""; @@ -954,7 +982,7 @@ char *getFacility(msg_t *pM) return((char*)pM->pszFacility); } -char *getFacilityStr(msg_t *pM) +static inline char *getFacilityStr(msg_t *pM) { syslogCODE *c; int val; @@ -1006,6 +1034,17 @@ MsgSetFlowControlType(msg_t *pMsg, flowControl_t eFlowCtl) RETiRet; } +/* set offset after which PRI in raw msg starts + * rgerhards, 2009-06-16 + */ +rsRetVal +MsgSetAfterPRIOffs(msg_t *pMsg, short offs) +{ + assert(pMsg != NULL); + pMsg->offAfterPRI = offs; + return RS_RET_OK; +} + /* rgerhards 2004-11-24: set APP-NAME in msg object * TODO: revisit msg locking code! @@ -1061,7 +1100,7 @@ finalize_it: /* rgerhards, 2005-11-24 */ -int getPROCIDLen(msg_t *pM) +static inline int getPROCIDLen(msg_t *pM) { assert(pM != NULL); MsgLock(pM); @@ -1106,19 +1145,10 @@ finalize_it: RETiRet; } -/* rgerhards, 2005-11-24 - */ -#if 0 /* This method is currently not called, be we like to preserve it */ -static int getMSGIDLen(msg_t *pM) -{ - return (pM->pCSMSGID == NULL) ? 1 : rsCStrLen(pM->pCSMSGID); -} -#endif - /* rgerhards, 2005-11-24 */ -char *getMSGID(msg_t *pM) +static inline char *getMSGID(msg_t *pM) { return (pM->pCSMSGID == NULL) ? "-" : (char*) rsCStrGetSzStrNoNULL(pM->pCSMSGID); } @@ -1142,8 +1172,7 @@ void MsgAssignTAG(msg_t *pMsg, uchar *pBuf) void MsgSetTAG(msg_t *pMsg, char* pszTAG) { assert(pMsg != NULL); - if(pMsg->pszTAG != NULL) - free(pMsg->pszTAG); + free(pMsg->pszTAG); pMsg->iLenTAG = strlen(pszTAG); if((pMsg->pszTAG = malloc(pMsg->iLenTAG + 1)) != NULL) memcpy(pMsg->pszTAG, pszTAG, pMsg->iLenTAG + 1); @@ -1200,7 +1229,7 @@ static int getTAGLen(msg_t *pM) #endif -char *getTAG(msg_t *pM) +static inline char *getTAG(msg_t *pM) { char *ret; @@ -1243,7 +1272,7 @@ char *getHOSTNAME(msg_t *pM) } -uchar *getInputName(msg_t *pM) +static uchar *getInputName(msg_t *pM) { if(pM == NULL) return (uchar*) ""; @@ -1255,15 +1284,15 @@ uchar *getInputName(msg_t *pM) } -char *getRcvFrom(msg_t *pM) +uchar *getRcvFrom(msg_t *pM) { if(pM == NULL) - return ""; + return UCHAR_CONSTANT(""); else if(pM->pszRcvFrom == NULL) - return ""; + return UCHAR_CONSTANT(""); else - return (char*) pM->pszRcvFrom; + return pM->pszRcvFrom; } @@ -1310,7 +1339,7 @@ static int getStructuredDataLen(msg_t *pM) /* get the "STRUCTURED-DATA" as sz string * rgerhards, 2005-11-24 */ -char *getStructuredData(msg_t *pM) +static inline char *getStructuredData(msg_t *pM) { return (pM->pCSStrucData == NULL) ? "-" : (char*) rsCStrGetSzStrNoNULL(pM->pCSStrucData); } @@ -1436,14 +1465,13 @@ static int getAPPNAMELen(msg_t *pM) } /* rgerhards 2008-09-10: set pszInputName in msg object + * rgerhards, 2009-06-16 */ -void MsgSetInputName(msg_t *pMsg, char* pszInputName) +void MsgSetInputName(msg_t *pMsg, uchar* pszInputName, size_t lenInputName) { assert(pMsg != NULL); - if(pMsg->pszInputName != NULL) - free(pMsg->pszInputName); - - pMsg->iLenInputName = strlen(pszInputName); + free(pMsg->pszInputName); + pMsg->iLenInputName = lenInputName; if((pMsg->pszInputName = malloc(pMsg->iLenInputName + 1)) != NULL) { memcpy(pMsg->pszInputName, pszInputName, pMsg->iLenInputName + 1); } @@ -1451,13 +1479,12 @@ void MsgSetInputName(msg_t *pMsg, char* pszInputName) /* rgerhards 2004-11-16: set pszRcvFrom in msg object */ -void MsgSetRcvFrom(msg_t *pMsg, char* pszRcvFrom) +void MsgSetRcvFrom(msg_t *pMsg, uchar* pszRcvFrom) { assert(pMsg != NULL); - if(pMsg->pszRcvFrom != NULL) - free(pMsg->pszRcvFrom); + free(pMsg->pszRcvFrom); - pMsg->iLenRcvFrom = strlen(pszRcvFrom); + pMsg->iLenRcvFrom = ustrlen(pszRcvFrom); if((pMsg->pszRcvFrom = malloc(pMsg->iLenRcvFrom + 1)) != NULL) { memcpy(pMsg->pszRcvFrom, pszRcvFrom, pMsg->iLenRcvFrom + 1); } @@ -1508,51 +1535,16 @@ void MsgAssignHOSTNAME(msg_t *pMsg, char *pBuf) * we need it. The rest of the code already knows how to handle an * unset HOSTNAME. */ -void MsgSetHOSTNAME(msg_t *pMsg, char* pszHOSTNAME) +void MsgSetHOSTNAME(msg_t *pMsg, uchar* pszHOSTNAME) { assert(pMsg != NULL); - if(pMsg->pszHOSTNAME != NULL) - free(pMsg->pszHOSTNAME); + free(pMsg->pszHOSTNAME); - pMsg->iLenHOSTNAME = strlen(pszHOSTNAME); + pMsg->iLenHOSTNAME = ustrlen(pszHOSTNAME); if((pMsg->pszHOSTNAME = malloc(pMsg->iLenHOSTNAME + 1)) != NULL) memcpy(pMsg->pszHOSTNAME, pszHOSTNAME, pMsg->iLenHOSTNAME + 1); else - dbgprintf("Could not allocate memory in MsgSetHOSTNAME()\n"); -} - - -/* Set the UxTradMsg to a caller-provided string. This is thought - * to be a heap buffer that the caller will no longer use. This - * function is a performance optimization over MsgSetUxTradMsg(). - * rgerhards 2004-11-19 - */ -#if 0 /* This method is currently not called, be we like to preserve it */ -static void MsgAssignUxTradMsg(msg_t *pMsg, char *pBuf) -{ - assert(pMsg != NULL); - assert(pBuf != NULL); - pMsg->iLenUxTradMsg = strlen(pBuf); - pMsg->pszUxTradMsg = pBuf; -} -#endif - - -/* rgerhards 2004-11-17: set the traditional Unix message in msg object - */ -int MsgSetUxTradMsg(msg_t *pMsg, char* pszUxTradMsg) -{ - assert(pMsg != NULL); - assert(pszUxTradMsg != NULL); - pMsg->iLenUxTradMsg = strlen(pszUxTradMsg); - if(pMsg->pszUxTradMsg != NULL) - free(pMsg->pszUxTradMsg); - if((pMsg->pszUxTradMsg = malloc(pMsg->iLenUxTradMsg + 1)) != NULL) - memcpy(pMsg->pszUxTradMsg, pszUxTradMsg, pMsg->iLenUxTradMsg + 1); - else - dbgprintf("Could not allocate memory for pszUxTradMsg buffer."); - - return(0); + DBGPRINTF("Could not allocate memory in MsgSetHOSTNAME()\n"); } @@ -1631,7 +1623,7 @@ static uchar *getNOW(eNOWType eNow) return NULL; } - datetime.getCurrTime(&t); + datetime.getCurrTime(&t, NULL); switch(eNow) { case NOW_NOW: snprintf((char*) pBuf, tmpBUFSIZE, "%4.4d-%2.2d-%2.2d", t.year, t.month, t.day); @@ -1734,12 +1726,15 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, pRes = getMSG(pMsg); } else if(!strcmp((char*) pName, "rawmsg")) { pRes = getRawMsg(pMsg); + /* enable this, if someone actually uses UxTradMsg, delete after some time has + * passed and nobody complained -- rgerhards, 2009-06-16 } else if(!strcmp((char*) pName, "uxtradmsg")) { pRes = getUxTradMsg(pMsg); + */ } else if(!strcmp((char*) pName, "inputname")) { pRes = (char*) getInputName(pMsg); } else if(!strcmp((char*) pName, "fromhost")) { - pRes = getRcvFrom(pMsg); + pRes = (char*) getRcvFrom(pMsg); } else if(!strcmp((char*) pName, "fromhost-ip")) { pRes = (char*) getRcvFromIP(pMsg); } else if(!strcmp((char*) pName, "source") || !strcmp((char*) pName, "hostname")) { @@ -2356,6 +2351,40 @@ char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, } } + /* finally, we need to check if the property should be formatted in CSV + * format (we use RFC 4180, and always use double quotes). As of this writing, + * this should be the last action carried out on the property, but in the + * future there may be reasons to change that. -- rgerhards, 2009-04-02 + */ + 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 = strlen(pRes); + char *pBStart; + char *pDst; + char *pSrc; + /* the malloc may be optimized, we currently use the worst case... */ + pBStart = pDst = malloc((2 * iBufLen + 3) * sizeof(char)); + if(pDst == NULL) { + if(*pbMustBeFreed == 1) + free(pRes); + *pbMustBeFreed = 0; + return "**OUT OF MEMORY**"; + } + pSrc = pRes; + *pDst++ = '"'; /* starting quote */ + while(*pSrc) { + if(*pSrc == '"') + *pDst++ = '"'; /* need to add double double quote (see RFC4180) */ + *pDst++ = *pSrc++; + } + *pDst++ = '"'; /* ending quote */ + *pDst = '\0'; + if(*pbMustBeFreed == 1) + free(pRes); + pRes = pBStart; + *pbMustBeFreed = 1; + } + /*dbgprintf("MsgGetProp(\"%s\"): \"%s\"\n", pName, pRes); only for verbose debug logging */ return(pRes); } @@ -2425,20 +2454,25 @@ rsRetVal MsgSetProperty(msg_t *pThis, var_t *pProp) pThis->msgFlags = pProp->val.num; } else if(isProp("pszRawMsg")) { MsgSetRawMsg(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); + /* enable this, if someone actually uses UxTradMsg, delete after some time has + * passed and nobody complained -- rgerhards, 2009-06-16 + } else if(isProp("offAfterPRI")) { + pThis->offAfterPRI = pProp->val.num; + */ } else if(isProp("pszMSG")) { MsgSetMSG(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); } else if(isProp("pszUxTradMsg")) { - MsgSetUxTradMsg(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); + /*IGNORE*/; /* this *was* a property, but does no longer exist */ } else if(isProp("pszTAG")) { MsgSetTAG(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); } else if(isProp("pszInputName")) { - MsgSetInputName(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); + MsgSetInputName(pThis, rsCStrGetSzStrNoNULL(pProp->val.pStr), rsCStrLen(pProp->val.pStr)); } else if(isProp("pszRcvFromIP")) { MsgSetRcvFromIP(pThis, rsCStrGetSzStrNoNULL(pProp->val.pStr)); } else if(isProp("pszRcvFrom")) { - MsgSetRcvFrom(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); + MsgSetRcvFrom(pThis, rsCStrGetSzStrNoNULL(pProp->val.pStr)); } else if(isProp("pszHOSTNAME")) { - MsgSetHOSTNAME(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); + MsgSetHOSTNAME(pThis, rsCStrGetSzStrNoNULL(pProp->val.pStr)); } else if(isProp("pCSStrucData")) { MsgSetStructuredData(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); } else if(isProp("pCSAPPNAME")) { @@ -2447,6 +2481,8 @@ rsRetVal MsgSetProperty(msg_t *pThis, var_t *pProp) MsgSetPROCID(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); } else if(isProp("pCSMSGID")) { MsgSetMSGID(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); + } else if(isProp("ttGenTime")) { + pThis->ttGenTime = pProp->val.num; } else if(isProp("tRcvdAt")) { memcpy(&pThis->tRcvdAt, &pProp->val.vSyslogTime, sizeof(struct syslogTime)); } else if(isProp("tTIMESTAMP")) { @@ -2508,7 +2544,5 @@ BEGINObjClassInit(msg, 1, OBJ_IS_CORE_MODULE) funcDeleteMutex = MsgLockingDummy; funcMsgPrepareEnqueue = MsgLockingDummy; ENDObjClassInit(msg) - -/* - * vi:set ai: +/* vim:set ai: */ diff --git a/runtime/msg.h b/runtime/msg.h index 1bad9c66..fe9f87fa 100644 --- a/runtime/msg.h +++ b/runtime/msg.h @@ -51,9 +51,11 @@ struct msg { BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */ pthread_mutexattr_t mutAttr; -short bDoLock; /* use the mutex? */ + bool bDoLock; /* use the mutex? */ pthread_mutex_t mut; - int iRefCount; /* reference counter (0 = unused) */ + flowControl_t flowCtlType; /**< type of flow control we can apply, for enqueueing, needs not to be persisted because + once data has entered the queue, this property is no longer needed. */ + short iRefCount; /* reference counter (0 = unused) */ short bParseHOSTNAME; /* should the hostname be parsed from the message? */ /* background: the hostname is not present on "regular" messages * received via UNIX domain sockets from the same machine. However, @@ -61,8 +63,6 @@ short bDoLock; /* use the mutex? */ * sockets. All in all, the parser would need parse templates, that would * resolve all these issues... rgerhards, 2005-10-06 */ - flowControl_t flowCtlType; /**< type of flow control we can apply, for enqueueing, needs not to be persisted because - once data has entered the queue, this property is no longer needed. */ short iSeverity; /* the severity 0..7 */ uchar *pszSeverity; /* severity as string... */ int iLenSeverity; /* ... and its length. */ @@ -73,12 +73,13 @@ short bDoLock; /* use the mutex? */ int iLenFacility; /* ... and its length. */ uchar *pszFacilityStr; /* facility name... */ int iLenFacilityStr; /* ... and its length. */ - uchar *pszPRI; /* the PRI as a string */ + uchar bufPRI[5]; /* PRI as string */ int iLenPRI; /* and its length */ uchar *pszRawMsg; /* message as it was received on the * wire. This is important in case we * need to preserve cryptographic verifiers. */ + short offAfterPRI; /* offset, at which raw message WITHOUT PRI part starts in pszRawMsg */ int iLenRawMsg; /* length of raw message */ uchar *pszMSG; /* the MSG part itself */ int iLenMSG; /* Length of the MSG part */ @@ -100,6 +101,13 @@ short bDoLock; /* use the mutex? */ cstr_t *pCSAPPNAME; /* APP-NAME */ cstr_t *pCSPROCID; /* PROCID */ cstr_t *pCSMSGID; /* MSGID */ + time_t ttGenTime; /* time msg object was generated, same as tRcvdAt, but a Unix timestamp. + While this field looks redundant, it is required because a Unix timestamp + is used at later processing stages (namely in the output arena). Thanks to + the subleties of how time is defined, there is no reliable way to reconstruct + the Unix timestamp from the syslogTime fields (in practice, we may be close + enough to reliable, but I prefer to leave the subtle things to the OS, where + it obviously is solved in way or another...). */ struct syslogTime tRcvdAt;/* time the message entered this program */ char *pszRcvdAt3164; /* time as RFC3164 formatted string (always 15 charcters) */ char *pszRcvdAt3339; /* time as RFC3164 formatted string (32 charcters at most) */ @@ -113,64 +121,71 @@ short bDoLock; /* use the mutex? */ char *pszTIMESTAMP_PgSQL;/* TIMESTAMP as PgSQL formatted string (always 21 characters) */ char *pszTIMESTAMP_SecFrac;/* TIMESTAMP fractional seconds (always 6 characters) */ int msgFlags; /* flags associated with this message */ + /* now follow fixed-size buffers to safe some time otherwise used for allocs */ + }; + +/* message flags (msgFlags), not an enum for historical reasons + */ +#define NOFLAG 0x000 /* no flag is set (to be used when a flag must be specified and none is required) */ +#define INTERNAL_MSG 0x001 /* msg generated by logmsgInternal() --> special handling */ +/* 0x002 not used because it was previously a known value - rgerhards, 2008-10-09 */ +#define IGNDATE 0x004 /* ignore, if given, date in message and use date of reception as msg date */ +#define MARK 0x008 /* this message is a mark */ +#define NEEDS_PARSING 0x010 /* raw message, must be parsed before processing can be done */ +#define PARSE_HOSTNAME 0x020 /* parse the hostname during message parsing */ + + /* function prototypes */ PROTOTYPEObjClassInit(msg); -char* getProgramName(msg_t*); rsRetVal msgConstruct(msg_t **ppThis); +rsRetVal msgConstructWithTime(msg_t **ppThis, struct syslogTime *stTime, time_t ttGenTime); rsRetVal msgDestruct(msg_t **ppM); msg_t* MsgDup(msg_t* pOld); msg_t *MsgAddRef(msg_t *pM); void setProtocolVersion(msg_t *pM, int iNewVersion); -int getProtocolVersion(msg_t *pM); -char *getProtocolVersionString(msg_t *pM); -int getMSGLen(msg_t *pM); -char *getRawMsg(msg_t *pM); -char *getUxTradMsg(msg_t *pM); -char *getMSG(msg_t *pM); -char *getPRI(msg_t *pM); -int getPRIi(msg_t *pM); -char *getTimeReported(msg_t *pM, enum tplFormatTypes eFmt); -char *getTimeGenerated(msg_t *pM, enum tplFormatTypes eFmt); -char *getSeverity(msg_t *pM); -char *getSeverityStr(msg_t *pM); -char *getFacility(msg_t *pM); -char *getFacilityStr(msg_t *pM); -void MsgSetInputName(msg_t *pMsg, char*); +void MsgSetInputName(msg_t *pMsg, uchar*, size_t); rsRetVal MsgSetAPPNAME(msg_t *pMsg, char* pszAPPNAME); -char *getAPPNAME(msg_t *pM); rsRetVal MsgSetPROCID(msg_t *pMsg, char* pszPROCID); -int getPROCIDLen(msg_t *pM); -char *getPROCID(msg_t *pM); rsRetVal MsgSetMSGID(msg_t *pMsg, char* pszMSGID); void MsgAssignTAG(msg_t *pMsg, uchar *pBuf); void MsgSetTAG(msg_t *pMsg, char* pszTAG); rsRetVal MsgSetFlowControlType(msg_t *pMsg, flowControl_t eFlowCtl); -char *getTAG(msg_t *pM); -int getHOSTNAMELen(msg_t *pM); -char *getHOSTNAME(msg_t *pM); -char *getRcvFrom(msg_t *pM); rsRetVal MsgSetStructuredData(msg_t *pMsg, char* pszStrucData); -char *getStructuredData(msg_t *pM); -int getProgramNameLen(msg_t *pM); -char *getProgramName(msg_t *pM); -void MsgSetRcvFrom(msg_t *pMsg, char* pszRcvFrom); +void MsgSetRcvFrom(msg_t *pMsg, uchar* pszRcvFrom); rsRetVal MsgSetRcvFromIP(msg_t *pMsg, uchar* pszRcvFromIP); void MsgAssignHOSTNAME(msg_t *pMsg, char *pBuf); -void MsgSetHOSTNAME(msg_t *pMsg, char* pszHOSTNAME); -int MsgSetUxTradMsg(msg_t *pMsg, char* pszUxTradMsg); +void MsgSetHOSTNAME(msg_t *pMsg, uchar* pszHOSTNAME); +rsRetVal MsgSetAfterPRIOffs(msg_t *pMsg, short offs); void MsgSetMSG(msg_t *pMsg, char* pszMSG); void MsgSetRawMsg(msg_t *pMsg, char* pszRawMsg); void moveHOSTNAMEtoTAG(msg_t *pM); -char *getMSGID(msg_t *pM); char *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, cstr_t *pCSPropName, unsigned short *pbMustBeFreed); char *textpri(char *pRes, size_t pResLen, int pri); rsRetVal msgGetMsgVar(msg_t *pThis, cstr_t *pstrPropName, var_t **ppVar); rsRetVal MsgEnableThreadSafety(void); +/* TODO: remove these five (so far used in action.c) */ +char *getMSG(msg_t *pM); +char *getHOSTNAME(msg_t *pM); +char *getPROCID(msg_t *pM); +char *getAPPNAME(msg_t *pM); +int getMSGLen(msg_t *pM); + +char *getHOSTNAME(msg_t *pM); +int getHOSTNAMELen(msg_t *pM); +char *getProgramName(msg_t *pM); +int getProgramNameLen(msg_t *pM); +uchar *getRcvFrom(msg_t *pM); + +#if 0 +char *getUxTradMsg(msg_t *pM); +int MsgSetUxTradMsg(msg_t *pMsg, char* pszUxTradMsg); +#endif + /* The MsgPrepareEnqueue() function is a macro for performance reasons. * It needs one global variable to work. This is acceptable, as it gains * us quite some performance and is fully abstracted using this header file. @@ -181,6 +196,5 @@ extern void (*funcMsgPrepareEnqueue)(msg_t *pMsg); #define MsgPrepareEnqueue(pMsg) funcMsgPrepareEnqueue(pMsg) #endif /* #ifndef MSG_H_INCLUDED */ -/* - * vi:set ai: +/* vim:set ai: */ diff --git a/runtime/net.c b/runtime/net.c index f289ecc5..5cafe522 100644 --- a/runtime/net.c +++ b/runtime/net.c @@ -63,6 +63,11 @@ #include "errmsg.h" #include "net.h" +#ifdef OS_SOLARIS +# define s6_addr32 _S6_un._S6_u32 + typedef unsigned int u_int32_t; +#endif + MODULE_TYPE_LIB /* static data */ @@ -1233,7 +1238,9 @@ rsRetVal cvthname(struct sockaddr_storage *f, uchar *pszHost, uchar *pszHostFQDN * make this in option in the long term. (rgerhards, 2007-09-11) */ strcpy((char*)pszHost, (char*)pszHostFQDN); - if ((p = (uchar*) strchr((char*)pszHost, '.'))) { /* find start of domain name "machine.example.com" */ + if( (glbl.GetPreserveFQDN() == 0) + && (p = (uchar*) strchr((char*)pszHost, '.'))) { /* find start of domain name "machine.example.com" */ + strcmp((char*)(p + 1), (char*)glbl.GetLocalDomain()); if(strcmp((char*)(p + 1), (char*)glbl.GetLocalDomain()) == 0) { *p = '\0'; /* simply terminate the string */ } else { @@ -1489,6 +1496,49 @@ int *create_udp_socket(uchar *hostname, uchar *pszPort, int bIsServer) } +/* check if two provided socket addresses point to the same host. Note that the + * length of the sockets must be provided as third parameter. This is necessary to + * compare non IPv4/v6 hosts, in which case we do a simple memory compare of the + * address structure (in that case, the same host may not reliably be detected). + * Note that we need to do the comparison not on the full structure, because it contains things + * like the port, which we do not need to look at when thinking about hostnames. So we look + * at the relevant fields, what means a somewhat more complicated processing. + * Also note that we use a non-standard calling interface, as this is much more natural and + * it looks extremely unlikely that we get an exception of any kind here. What we + * return is mimiced after memcmp(), and as such useful for building binary trees + * (the order relation may be a bit arbritrary, but at least it is consistent). + * rgerhards, 2009-09-03 + */ +static int CmpHost(struct sockaddr_storage *s1, struct sockaddr_storage* s2, size_t socklen) +{ + int ret; + + if(((struct sockaddr*) s1)->sa_family != ((struct sockaddr*) s2)->sa_family) { + ret = memcmp(s1, s2, socklen); + goto finalize_it; + } + + if(((struct sockaddr*) s1)->sa_family == AF_INET) { + if(((struct sockaddr_in *) s1)->sin_addr.s_addr == ((struct sockaddr_in*)s2)->sin_addr.s_addr) { + ret = 0; + } else if(((struct sockaddr_in *) s1)->sin_addr.s_addr < ((struct sockaddr_in*)s2)->sin_addr.s_addr) { + ret = -1; + } else { + ret = 1; + } + } else if(((struct sockaddr*) s1)->sa_family == AF_INET6) { + /* IPv6 addresses are always 16 octets long */ + ret = memcmp(((struct sockaddr_in6 *)s1)->sin6_addr.s6_addr, ((struct sockaddr_in6*)s2)->sin6_addr.s6_addr, 16); + } else { + ret = memcmp(s1, s2, socklen); + } + +dbgprintf("CmpHost returns %d\n", ret); +finalize_it: + return ret; +} + + /* queryInterface function * rgerhards, 2008-03-05 */ @@ -1517,6 +1567,7 @@ CODESTARTobjQueryInterface(net) pIf->AddPermittedPeer = AddPermittedPeer; pIf->DestructPermittedPeers = DestructPermittedPeers; pIf->PermittedPeerWildcardMatch = PermittedPeerWildcardMatch; + pIf->CmpHost = CmpHost; finalize_it: ENDobjQueryInterface(net) diff --git a/runtime/net.h b/runtime/net.h index 092c3116..ec364b1c 100644 --- a/runtime/net.h +++ b/runtime/net.h @@ -146,11 +146,13 @@ BEGINinterface(net) /* name must also be changed in ENDinterface macro! */ rsRetVal (*AddPermittedPeer)(permittedPeers_t **ppRootPeer, uchar *pszID); rsRetVal (*DestructPermittedPeers)(permittedPeers_t **ppRootPeer); rsRetVal (*PermittedPeerWildcardMatch)(permittedPeers_t *pPeer, uchar *pszNameToMatch, int *pbIsMatching); + /* v5 interface additions */ + int (*CmpHost)(struct sockaddr_storage *, struct sockaddr_storage*, size_t); /* data members - these should go away over time... TODO */ int *pACLAddHostnameOnFail; /* add hostname to acl when DNS resolving has failed */ int *pACLDontResolve; /* add hostname to acl instead of resolving it to IP(s) */ ENDinterface(net) -#define netCURR_IF_VERSION 4 /* increment whenever you change the interface structure! */ +#define netCURR_IF_VERSION 5 /* increment whenever you change the interface structure! */ /* prototypes */ PROTOTYPEObj(net); diff --git a/runtime/netstrm.c b/runtime/netstrm.c index ffa1c578..3658006f 100644 --- a/runtime/netstrm.c +++ b/runtime/netstrm.c @@ -17,7 +17,7 @@ * Rainer Gerhards and Adiscon GmbH have agreed to permit using the code * under the terms of the GNU Lesser General Public License. * - * Copyright 2007, 2008 Rainer Gerhards and Adiscon GmbH. + * Copyright 2007-2009 Rainer Gerhards and Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -233,6 +233,19 @@ Send(netstrm_t *pThis, uchar *pBuf, ssize_t *pLenBuf) RETiRet; } +/* Enable Keep-Alive handling for those drivers that support it. + * rgerhards, 2009-06-02 + */ +static rsRetVal +EnableKeepAlive(netstrm_t *pThis) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, netstrm); + iRet = pThis->Drvr.EnableKeepAlive(pThis->pDrvrData); + RETiRet; +} + + /* check connection - slim wrapper for NSD driver function */ static void @@ -337,6 +350,7 @@ CODESTARTobjQueryInterface(netstrm) pIf->SetDrvrPermPeers = SetDrvrPermPeers; pIf->CheckConnection = CheckConnection; pIf->GetSock = GetSock; + pIf->EnableKeepAlive = EnableKeepAlive; finalize_it: ENDobjQueryInterface(netstrm) diff --git a/runtime/netstrm.h b/runtime/netstrm.h index 3ab790e8..f6931104 100644 --- a/runtime/netstrm.h +++ b/runtime/netstrm.h @@ -31,6 +31,7 @@ struct netstrm_s { BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */ nsd_t *pDrvrData; /**< the driver's data elements (at most other places, this is called pNsd) */ nsd_if_t Drvr; /**< our stream driver */ + void *pUsr; /**< pointer to user-provided data structure */ netstrms_t *pNS; /**< pointer to our netstream subsystem object */ }; @@ -68,9 +69,13 @@ BEGINinterface(netstrm) /* name must also be changed in ENDinterface macro! */ * sockets - not really desirable, but not the end of the world... TODO: should be * reconsidered when a new ACL system is build. -- rgerhards, 2008-12-01 */ + /* v4 */ + rsRetVal (*EnableKeepAlive)(netstrm_t *pThis); ENDinterface(netstrm) -#define netstrmCURR_IF_VERSION 3 /* increment whenever you change the interface structure! */ -/* interface version 3 added GetRemAddr() */ +#define netstrmCURR_IF_VERSION 4 /* increment whenever you change the interface structure! */ +/* interface version 3 added GetRemAddr() + * interface version 4 added EnableKeepAlive() -- rgerhards, 2009-06-02 + * */ /* prototypes */ PROTOTYPEObj(netstrm); diff --git a/runtime/nsd.h b/runtime/nsd.h index f0c9b9b6..8668c934 100644 --- a/runtime/nsd.h +++ b/runtime/nsd.h @@ -69,9 +69,13 @@ BEGINinterface(nsd) /* name must also be changed in ENDinterface macro! */ * sockets - not really desirable, but not the end of the world... TODO: should be * reconsidered when a new ACL system is build. -- rgerhards, 2008-12-01 */ + /* v5 */ + rsRetVal (*EnableKeepAlive)(nsd_t *pThis); ENDinterface(nsd) -#define nsdCURR_IF_VERSION 4 /* increment whenever you change the interface structure! */ -/* interface version 4 added GetRemAddr() */ +#define nsdCURR_IF_VERSION 5 /* increment whenever you change the interface structure! */ +/* interface version 4 added GetRemAddr() + * interface version 5 added EnableKeepAlive() -- rgerhards, 2009-06-02 + */ /* interface for the select call */ BEGINinterface(nsdsel) /* name must also be changed in ENDinterface macro! */ diff --git a/runtime/nsd_gtls.c b/runtime/nsd_gtls.c index 3a79a015..19dc8678 100644 --- a/runtime/nsd_gtls.c +++ b/runtime/nsd_gtls.c @@ -82,6 +82,7 @@ static gnutls_certificate_credentials xcred; static gnutls_dh_params dh_params; #ifdef DEBUG +#if 0 /* uncomment, if needed some time again -- DEV Debug only */ /* This defines a log function to be provided to GnuTLS. It hopefully * helps us track down hard to find problems. * rgerhards, 2008-06-20 @@ -90,6 +91,7 @@ static void logFunction(int level, const char *msg) { dbgprintf("GnuTLS log msg, level %d: %s\n", level, msg); } +#endif #endif /* #ifdef DEBUG */ @@ -333,7 +335,7 @@ gtlsGetCertInfo(nsd_gtls_t *pThis, cstr_t **ppStr) gnutls_x509_crt_deinit(cert); } - CHKiRet(rsCStrFinish(pStr)); + CHKiRet(cstrFinalize(pStr)); *ppStr = pStr; finalize_it: @@ -453,7 +455,7 @@ GenFingerprintStr(uchar *pFingerprint, size_t sizeFingerprint, cstr_t **ppStr) snprintf((char*)buf, sizeof(buf), ":%2.2X", pFingerprint[i]); CHKiRet(rsCStrAppendStrWithLen(pStr, buf, 3)); } - CHKiRet(rsCStrFinish(pStr)); + CHKiRet(cstrFinalize(pStr)); *ppStr = pStr; @@ -721,7 +723,7 @@ gtlsGetCN(nsd_gtls_t *pThis, gnutls_x509_crt *pCert, cstr_t **ppstrCN) } ++i; /* char processed */ } - CHKiRet(rsCStrFinish(pstrCN)); + CHKiRet(cstrFinalize(pstrCN)); /* we got it - we ignore the rest of the DN string (if any). So we may * not detect if it contains more than one CN @@ -882,7 +884,7 @@ gtlsChkPeerName(nsd_gtls_t *pThis, gnutls_x509_crt *pCert) if(!bFoundPositiveMatch) { dbgprintf("invalid peer name, not permitted to talk to it\n"); if(pThis->bReportAuthErr == 1) { - CHKiRet(rsCStrFinish(pStr)); + CHKiRet(cstrFinalize(pStr)); errno = 0; errmsg.LogError(0, RS_RET_INVALID_FINGERPRINT, "error: peer name not authorized - " "not permitted to talk to it. Names: %s", @@ -1558,6 +1560,16 @@ finalize_it: RETiRet; } +/* Enable KEEPALIVE handling on the socket. + * rgerhards, 2009-06-02 + */ +static rsRetVal +EnableKeepAlive(nsd_t *pNsd) +{ + return nsd_ptcp.EnableKeepAlive(pNsd); +} + + /* open a connection to a remote host (server). With GnuTLS, we always * open a plain tcp socket and then, if in TLS mode, do a handshake on it. @@ -1667,6 +1679,7 @@ CODESTARTobjQueryInterface(nsd_gtls) pIf->GetRemoteHName = GetRemoteHName; pIf->GetRemoteIP = GetRemoteIP; pIf->GetRemAddr = GetRemAddr; + pIf->EnableKeepAlive = EnableKeepAlive; finalize_it: ENDobjQueryInterface(nsd_gtls) diff --git a/runtime/nsd_ptcp.c b/runtime/nsd_ptcp.c index cc531ca0..54ee0666 100644 --- a/runtime/nsd_ptcp.c +++ b/runtime/nsd_ptcp.c @@ -614,6 +614,34 @@ finalize_it: } +/* Enable KEEPALIVE handling on the socket. + * rgerhards, 2009-06-02 + */ +static rsRetVal +EnableKeepAlive(nsd_t *pNsd) +{ + nsd_ptcp_t *pThis = (nsd_ptcp_t*) pNsd; + int ret; + int optval; + socklen_t optlen; + DEFiRet; + ISOBJ_TYPE_assert(pThis, nsd_ptcp); + + optval = 1; + optlen = sizeof(optval); + ret = setsockopt(pThis->sock, SOL_SOCKET, SO_KEEPALIVE, &optval, optlen); + if(ret < 0) { + dbgprintf("EnableKeepAlive socket call returns error %d\n", ret); + ABORT_FINALIZE(RS_RET_ERR); + } + + dbgprintf("KEEPALIVE enabled for nsd %p\n", pThis); + +finalize_it: + RETiRet; +} + + /* open a connection to a remote host (server). * rgerhards, 2008-03-19 */ @@ -754,6 +782,7 @@ CODESTARTobjQueryInterface(nsd_ptcp) pIf->GetRemoteHName = GetRemoteHName; pIf->GetRemoteIP = GetRemoteIP; pIf->CheckConnection = CheckConnection; + pIf->EnableKeepAlive = EnableKeepAlive; finalize_it: ENDobjQueryInterface(nsd_ptcp) diff --git a/runtime/obj-types.h b/runtime/obj-types.h index 914c2f2c..78829f94 100644 --- a/runtime/obj-types.h +++ b/runtime/obj-types.h @@ -280,7 +280,7 @@ rsRetVal objName##ClassExit(void) \ * rgerhards, 2008-01-30 */ #define BEGINobjDestruct(OBJ) \ - rsRetVal OBJ##Destruct(OBJ##_t **ppThis) \ + rsRetVal OBJ##Destruct(OBJ##_t __attribute__((unused)) **ppThis) \ { \ DEFiRet; \ int iCancelStateSave; \ @@ -314,7 +314,7 @@ rsRetVal objName##ClassExit(void) \ #define PROTOTYPEObjDebugPrint(obj) rsRetVal obj##DebugPrint(obj##_t *pThis) #define INTERFACEObjDebugPrint(obj) rsRetVal (*DebugPrint)(obj##_t *pThis) #define BEGINobjDebugPrint(obj) \ - rsRetVal obj##DebugPrint(obj##_t *pThis) \ + rsRetVal obj##DebugPrint(obj##_t __attribute__((unused)) *pThis) \ { \ DEFiRet; \ diff --git a/runtime/obj.c b/runtime/obj.c index 2a9df9ed..355c0f97 100644 --- a/runtime/obj.c +++ b/runtime/obj.c @@ -447,7 +447,7 @@ objDeserializeEmbedStr(cstr_t **ppStr, strm_t *pStrm) CHKiRet(rsCStrAppendChar(pStr, c)); NEXTC; } - CHKiRet(rsCStrFinish(pStr)); + CHKiRet(cstrFinalize(pStr)); *ppStr = pStr; @@ -515,7 +515,7 @@ static rsRetVal objDeserializeStr(cstr_t **ppCStr, int iLen, strm_t *pStrm) CHKiRet(rsCStrAppendChar(pCStr, c)); NEXTC; } - CHKiRet(rsCStrFinish(pCStr)); + CHKiRet(cstrFinalize(pCStr)); /* check terminator */ if(c != ':') ABORT_FINALIZE(RS_RET_INVALID_DELIMITER); @@ -629,7 +629,7 @@ static rsRetVal objDeserializeProperty(var_t *pProp, strm_t *pStrm) CHKiRet(rsCStrAppendChar(pProp->pcsName, c)); NEXTC; } - CHKiRet(rsCStrFinish(pProp->pcsName)); + CHKiRet(cstrFinalize(pProp->pcsName)); /* property type */ CHKiRet(objDeserializeNumber(&i, pStrm)); diff --git a/runtime/objomsr.c b/runtime/objomsr.c index 21d284f3..8dddc4b8 100644 --- a/runtime/objomsr.c +++ b/runtime/objomsr.c @@ -141,5 +141,23 @@ int OMSRgetEntry(omodStringRequest_t *pThis, int iEntry, uchar **ppTplName, int return RS_RET_OK; } + + +/* return the full set of template options that are supported by this version of + * OMSR. They are returned in an unsigned long value. The caller can mask that + * value to check on the option he is interested in. + * Note that this interface was added in 4.1.6, so a plugin must obtain a pointer + * to this interface via queryHostEtryPt(). + * rgerhards, 2009-04-03 + */ +rsRetVal +OMSRgetSupportedTplOpts(unsigned long *pOpts) +{ + DEFiRet; + assert(pOpts != NULL); + *pOpts = OMSR_RQD_TPL_OPT_SQL | OMSR_TPL_AS_ARRAY; + RETiRet; +} + /* vim:set ai: */ diff --git a/runtime/objomsr.h b/runtime/objomsr.h index 2255e4f3..75ad0fb8 100644 --- a/runtime/objomsr.h +++ b/runtime/objomsr.h @@ -1,6 +1,6 @@ /* Definition of the omsr (omodStringRequest) object. * - * Copyright 2007 Rainer Gerhards and Adiscon GmbH. + * Copyright 2007, 2009 Rainer Gerhards and Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -27,7 +27,8 @@ /* define flags for required template options */ #define OMSR_NO_RQD_TPL_OPTS 0 #define OMSR_RQD_TPL_OPT_SQL 1 -/* next option is 2, 4, 8, ... */ +#define OMSR_TPL_AS_ARRAY 2 /* introduced in 4.1.6, 2009-04-03 */ +/* next option is 4, 8, 16, ... */ struct omodStringRequest_s { /* strings requested by output module for doAction() */ int iNumEntries; /* number of array entries for data elements below */ @@ -40,6 +41,7 @@ typedef struct omodStringRequest_s omodStringRequest_t; rsRetVal OMSRdestruct(omodStringRequest_t *pThis); rsRetVal OMSRconstruct(omodStringRequest_t **ppThis, int iNumEntries); rsRetVal OMSRsetEntry(omodStringRequest_t *pThis, int iEntry, uchar *pTplName, int iTplOpts); +rsRetVal OMSRgetSupportedTplOpts(unsigned long *pOpts); int OMSRgetEntryCount(omodStringRequest_t *pThis); int OMSRgetEntry(omodStringRequest_t *pThis, int iEntry, uchar **ppTplName, int *piTplOpts); diff --git a/runtime/parser.c b/runtime/parser.c new file mode 100644 index 00000000..0b45bfd5 --- /dev/null +++ b/runtime/parser.c @@ -0,0 +1,349 @@ +/* parser.c + * This module contains functions for message parsers. It still needs to be + * converted into an object (and much extended). + * + * Module begun 2008-10-09 by Rainer Gerhards (based on previous code from syslogd.c) + * + * Copyright 2008 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ +#include "config.h" +#include <stdlib.h> +#include <ctype.h> +#include <string.h> +#include <assert.h> +#ifdef USE_NETZIP +#include <zlib.h> +#endif + +#include "rsyslog.h" +#include "dirty.h" +#include "msg.h" +#include "obj.h" +#include "errmsg.h" + +/* some defines */ +#define DEFUPRI (LOG_USER|LOG_NOTICE) + +/* definitions for objects we access */ +DEFobjStaticHelpers +DEFobjCurrIf(glbl) +DEFobjCurrIf(errmsg) + +/* static data */ + + +/* this is a dummy class init + */ +rsRetVal parserClassInit(void) +{ + DEFiRet; + + /* request objects we use */ + CHKiRet(objGetObjInterface(&obj)); /* this provides the root pointer for all other queries */ + CHKiRet(objUse(glbl, CORE_COMPONENT)); + CHKiRet(objUse(errmsg, CORE_COMPONENT)); +// TODO: free components! see action.c +finalize_it: + RETiRet; +} + + +/* uncompress a received message if it is compressed. + * pMsg->pszRawMsg buffer is updated. + * rgerhards, 2008-10-09 + */ +static inline rsRetVal uncompressMessage(msg_t *pMsg) +{ + DEFiRet; +# ifdef USE_NETZIP + uchar *deflateBuf = NULL; + uLongf iLenDefBuf; + uchar *pszMsg; + size_t lenMsg; + + assert(pMsg != NULL); + pszMsg = pMsg->pszRawMsg; + lenMsg = pMsg->iLenRawMsg; + + /* we first need to check if we have a compressed record. If so, + * we must decompress it. + */ + if(lenMsg > 0 && *pszMsg == 'z') { /* compressed data present? (do NOT change order if conditions!) */ + /* we have compressed data, so let's deflate it. We support a maximum + * message size of iMaxLine. If it is larger, an error message is logged + * and the message is dropped. We do NOT try to decompress larger messages + * as such might be used for denial of service. It might happen to later + * builds that such functionality be added as an optional, operator-configurable + * feature. + */ + int ret; + iLenDefBuf = glbl.GetMaxLine(); + CHKmalloc(deflateBuf = malloc(sizeof(uchar) * (iLenDefBuf + 1))); + ret = uncompress((uchar *) deflateBuf, &iLenDefBuf, (uchar *) pszMsg+1, lenMsg-1); + DBGPRINTF("Compressed message uncompressed with status %d, length: new %ld, old %d.\n", + ret, (long) iLenDefBuf, (int) (lenMsg-1)); + /* Now check if the uncompression worked. If not, there is not much we can do. In + * that case, we log an error message but ignore the message itself. Storing the + * compressed text is dangerous, as it contains control characters. So we do + * not do this. If someone would like to have a copy, this code here could be + * modified to do a hex-dump of the buffer in question. We do not include + * this functionality right now. + * rgerhards, 2006-12-07 + */ + if(ret != Z_OK) { + errmsg.LogError(0, NO_ERRCODE, "Uncompression of a message failed with return code %d " + "- enable debug logging if you need further information. " + "Message ignored.", ret); + FINALIZE; /* unconditional exit, nothing left to do... */ + } + free(pMsg->pszRawMsg); + pMsg->pszRawMsg = deflateBuf; + pMsg->iLenRawMsg = iLenDefBuf; + deflateBuf = NULL; /* logically "freed" - caller is now responsible */ + } +finalize_it: + if(deflateBuf != NULL) + free(deflateBuf); + +# else /* ifdef USE_NETZIP */ + + /* in this case, we still need to check if the message is compressed. If so, we must + * tell the user we can not accept it. + */ + if(pMsg->iLenRawMsg > 0 && *pMsg->pszRawMsg == 'z') { + errmsg.LogError(0, NO_ERRCODE, "Received a compressed message, but rsyslogd does not have compression " + "support enabled. The message will be ignored."); + ABORT_FINALIZE(RS_RET_NO_ZIP); + } + +finalize_it: +# endif /* ifdef USE_NETZIP */ + + RETiRet; +} + + +/* sanitize a received message + * if a message gets to large during sanitization, it is truncated. This is + * as specified in the upcoming syslog RFC series. + * rgerhards, 2008-10-09 + * We check if we have a NUL character at the very end of the + * message. This seems to be a frequent problem with a number of senders. + * So I have now decided to drop these NULs. However, if they are intentional, + * that may cause us some problems, e.g. with syslog-sign. On the other hand, + * current code always has problems with intentional NULs (as it needs to escape + * them to prevent problems with the C string libraries), so that does not + * really matter. Just to be on the save side, we'll log destruction of such + * NULs in the debug log. + * rgerhards, 2007-09-14 + */ +static inline rsRetVal +sanitizeMessage(msg_t *pMsg) +{ + DEFiRet; + uchar *pszMsg; + uchar *pDst; /* destination for copy job */ + size_t lenMsg; + size_t iSrc; + size_t iDst; + size_t iMaxLine; + + assert(pMsg != NULL); + +# ifdef USE_NETZIP + CHKiRet(uncompressMessage(pMsg)); +# endif + + pszMsg = pMsg->pszRawMsg; + lenMsg = pMsg->iLenRawMsg; + + /* remove NUL character at end of message (see comment in function header) */ + if(pszMsg[lenMsg-1] == '\0') { + DBGPRINTF("dropped NUL at very end of message\n"); + lenMsg--; + } + + /* then we check if we need to drop trailing LFs, which often make + * their way into syslog messages unintentionally. In order to remain + * compatible to recent IETF developments, we allow the user to + * turn on/off this handling. rgerhards, 2007-07-23 + */ + if(bDropTrailingLF && pszMsg[lenMsg-1] == '\n') { + DBGPRINTF("dropped LF at very end of message (DropTrailingLF is set)\n"); + lenMsg--; + } + + /* it is much quicker to sweep over the message and see if it actually + * needs sanitation than to do the sanitation in any case. So we first do + * this and terminate when it is not needed - which is expectedly the case + * for the vast majority of messages. -- rgerhards, 2009-06-15 + */ + int bNeedSanitize = 0; + for(iSrc = 0 ; iSrc < lenMsg ; iSrc++) { + if(pszMsg[iSrc] < 32) { + if(pszMsg[iSrc] == '\0' || bEscapeCCOnRcv) { + bNeedSanitize = 1; + break; + } + } + } + if(bNeedSanitize == 0) { + /* what a shame - we do not have a \0 byte... + * TODO: think about adding it or otherwise be able to use it... + */ + uchar *pRaw; + CHKmalloc(pRaw = realloc(pMsg->pszRawMsg, pMsg->iLenRawMsg + 1)); + pRaw[pMsg->iLenRawMsg] = '\0'; + pMsg->pszRawMsg = pRaw; + FINALIZE; + } + + /* now copy over the message and sanitize it */ + /* TODO: can we get cheaper memory alloc? {alloca()?}*/ + iMaxLine = glbl.GetMaxLine(); + CHKmalloc(pDst = malloc(sizeof(uchar) * (iMaxLine + 1))); + iSrc = iDst = 0; + while(iSrc < lenMsg && iDst < iMaxLine) { + if(pszMsg[iSrc] == '\0') { /* guard against \0 characters... */ + /* changed to the sequence (somewhat) proposed in + * draft-ietf-syslog-protocol-19. rgerhards, 2006-11-30 + */ + if(iDst + 3 < iMaxLine) { /* do we have space? */ + pDst[iDst++] = cCCEscapeChar; + pDst[iDst++] = '0'; + pDst[iDst++] = '0'; + pDst[iDst++] = '0'; + } /* if we do not have space, we simply ignore the '\0'... */ + /* log an error? Very questionable... rgerhards, 2006-11-30 */ + /* decided: we do not log an error, it won't help... rger, 2007-06-21 */ + } else if(bEscapeCCOnRcv && iscntrl((int) pszMsg[iSrc])) { + /* we are configured to escape control characters. Please note + * that this most probably break non-western character sets like + * Japanese, Korean or Chinese. rgerhards, 2007-07-17 + * Note: sysklogd logs octal values only for DEL and CCs above 127. + * For others, it logs ^n where n is the control char converted to an + * alphabet character. We like consistency and thus escape it to octal + * in all cases. If someone complains, we may change the mode. At least + * we known now what's going on. + * rgerhards, 2007-07-17 + */ + if(iDst + 3 < iMaxLine) { /* do we have space? */ + pDst[iDst++] = cCCEscapeChar; + pDst[iDst++] = '0' + ((pszMsg[iSrc] & 0300) >> 6); + pDst[iDst++] = '0' + ((pszMsg[iSrc] & 0070) >> 3); + pDst[iDst++] = '0' + ((pszMsg[iSrc] & 0007)); + } /* again, if we do not have space, we ignore the char - see comment at '\0' */ + } else { + pDst[iDst++] = pszMsg[iSrc]; + } + ++iSrc; + } + pDst[iDst] = '\0'; /* space *is* reserved for this! */ + + /* we have a sanitized string. Let's save it now */ + free(pMsg->pszRawMsg); + if((pMsg->pszRawMsg = malloc((iDst+1) * sizeof(uchar))) == NULL) { + /* when we get no new buffer, we use what we already have ;) */ + pMsg->pszRawMsg = pDst; + } else { + /* trim buffer */ + memcpy(pMsg->pszRawMsg, pDst, iDst+1); + free(pDst); /* too big! */ + pMsg->iLenRawMsg = iDst; + } + +finalize_it: + RETiRet; +} + +/* Parse a received message. The object's rawmsg property is taken and + * parsed according to the relevant standards. This can later be + * extended to support configured parsers. + * rgerhards, 2008-10-09 + */ +rsRetVal parseMsg(msg_t *pMsg) +{ + DEFiRet; + uchar *msg; + int pri; + int iPriText; + + CHKiRet(sanitizeMessage(pMsg)); + + /* we needed to sanitize first, because we otherwise do not have a C-string we can print... */ + DBGPRINTF("msg parser: flags %x, from '%s', msg '%s'\n", pMsg->msgFlags, pMsg->pszRcvFrom, pMsg->pszRawMsg); + + /* pull PRI */ + pri = DEFUPRI; + msg = pMsg->pszRawMsg; + iPriText = 0; + if(*msg == '<') { + /* while we process the PRI, we also fill the PRI textual representation + * inside the msg object. This may not be ideal from an OOP point of view, + * but it offers us performance... + */ + pri = 0; + while(isdigit((int) *++msg)) { + pMsg->bufPRI[iPriText++ % 4] = *msg; /* mod 4 to guard against malformed messages! */ + pri = 10 * pri + (*msg - '0'); + } + pMsg->bufPRI[iPriText % 4] = '\0'; + pMsg->iLenPRI = iPriText % 4; + if(*msg == '>') + ++msg; + if(pri & ~(LOG_FACMASK|LOG_PRIMASK)) + pri = DEFUPRI; + } + pMsg->iFacility = LOG_FAC(pri); + pMsg->iSeverity = LOG_PRI(pri); + MsgSetAfterPRIOffs(pMsg, msg - pMsg->pszRawMsg); + + if(pMsg->bParseHOSTNAME == 0) + MsgSetHOSTNAME(pMsg, pMsg->pszRcvFrom); + + /* rger 2005-11-24 (happy thanksgiving!): we now need to check if we have + * a traditional syslog message or one formatted according to syslog-protocol. + * We need to apply different parsers depending on that. We use the + * -protocol VERSION field for the detection. + */ + if(msg[0] == '1' && msg[1] == ' ') { + dbgprintf("Message has syslog-protocol format.\n"); + setProtocolVersion(pMsg, 1); + if(parseRFCSyslogMsg(pMsg, pMsg->msgFlags) == 1) { + msgDestruct(&pMsg); + ABORT_FINALIZE(RS_RET_ERR); // TODO: we need to handle these cases! + } + } else { /* we have legacy syslog */ + dbgprintf("Message has legacy syslog format.\n"); + setProtocolVersion(pMsg, 0); + if(parseLegacySyslogMsg(pMsg, pMsg->msgFlags) == 1) { + msgDestruct(&pMsg); + ABORT_FINALIZE(RS_RET_ERR); // TODO: we need to handle these cases! + } + } + + /* finalize message object */ + pMsg->msgFlags &= ~NEEDS_PARSING; /* this message is now parsed */ + MsgPrepareEnqueue(pMsg); /* "historical" name - preparese for multi-threading */ + +finalize_it: + RETiRet; +} diff --git a/runtime/parser.h b/runtime/parser.h new file mode 100644 index 00000000..cec9c083 --- /dev/null +++ b/runtime/parser.h @@ -0,0 +1,30 @@ +/* header for parser.c + * This is not yet an object, but contains all those code necessary to + * parse syslog messages. + * + * Copyright 2008 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ +#ifndef INCLUDED_PARSE_H +#define INCLUDED_PARSE_H + +extern rsRetVal parserClassInit(void); +extern rsRetVal parseMsg(msg_t*); + +#endif /* #ifndef INCLUDED_PARSE_H */ diff --git a/runtime/queue.c b/runtime/queue.c index e58b1686..4e017e84 100644 --- a/runtime/queue.c +++ b/runtime/queue.c @@ -49,20 +49,26 @@ #include "obj.h" #include "wtp.h" #include "wti.h" +#include "atomic.h" + +#ifdef OS_SOLARIS +# include <sched.h> +# define pthread_yield() sched_yield() +#endif /* static data */ DEFobjStaticHelpers DEFobjCurrIf(glbl) /* forward-definitions */ -rsRetVal queueChkPersist(queue_t *pThis); -static rsRetVal queueSetEnqOnly(queue_t *pThis, int bEnqOnly, int bLockMutex); -static rsRetVal queueRateLimiter(queue_t *pThis); -static int queueChkStopWrkrDA(queue_t *pThis); -static int queueIsIdleDA(queue_t *pThis); -static rsRetVal queueConsumerDA(queue_t *pThis, wti_t *pWti, int iCancelStateSave); -static rsRetVal queueConsumerCancelCleanup(void *arg1, void *arg2); -static rsRetVal queueUngetObj(queue_t *pThis, obj_t *pUsr, int bLockMutex); +rsRetVal qqueueChkPersist(qqueue_t *pThis); +static rsRetVal qqueueSetEnqOnly(qqueue_t *pThis, int bEnqOnly, int bLockMutex); +static rsRetVal qqueueRateLimiter(qqueue_t *pThis); +static int qqueueChkStopWrkrDA(qqueue_t *pThis); +static int qqueueIsIdleDA(qqueue_t *pThis); +static rsRetVal qqueueConsumerDA(qqueue_t *pThis, wti_t *pWti, int iCancelStateSave); +static rsRetVal qqueueConsumerCancelCleanup(void *arg1, void *arg2); +static rsRetVal qqueueUngetObj(qqueue_t *pThis, obj_t *pUsr, int bLockMutex); /* some constants for queuePersist () */ #define QUEUE_CHECKPOINT 1 @@ -76,7 +82,7 @@ static rsRetVal queueUngetObj(queue_t *pThis, obj_t *pUsr, int bLockMutex); * rgerhards, 2008-01-29 */ static inline int -queueGetOverallQueueSize(queue_t *pThis) +qqueueGetOverallQueueSize(qqueue_t *pThis) { #if 0 /* leave a bit in for debugging -- rgerhards, 2008-01-30 */ BEGINfunc @@ -95,7 +101,7 @@ ENDfunc * This function returns void, as it makes no sense to communicate an error back, even if * it happens. */ -static inline void queueDrain(queue_t *pThis) +static inline void queueDrain(qqueue_t *pThis) { void *pUsr; @@ -118,26 +124,26 @@ static inline void queueDrain(queue_t *pThis) * this point in time. The mutex must be locked when * ths function is called. -- rgerhards, 2008-01-25 */ -static inline rsRetVal queueAdviseMaxWorkers(queue_t *pThis) +static inline rsRetVal qqueueAdviseMaxWorkers(qqueue_t *pThis) { DEFiRet; int iMaxWorkers; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); if(!pThis->bEnqOnly) { if(pThis->bRunsDA) { /* if we have not yet reached the high water mark, there is no need to start a * worker. -- rgerhards, 2008-01-26 */ - if(queueGetOverallQueueSize(pThis) >= pThis->iHighWtrMrk || pThis->bQueueStarted == 0) { + if(qqueueGetOverallQueueSize(pThis) >= pThis->iHighWtrMrk || pThis->bQueueStarted == 0) { wtpAdviseMaxWorkers(pThis->pWtpDA, 1); /* disk queues have always one worker */ } } else { if(pThis->qType == QUEUETYPE_DISK || pThis->iMinMsgsPerWrkr == 0) { iMaxWorkers = 1; } else { - iMaxWorkers = queueGetOverallQueueSize(pThis) / pThis->iMinMsgsPerWrkr + 1; + iMaxWorkers = qqueueGetOverallQueueSize(pThis) / pThis->iMinMsgsPerWrkr + 1; } wtpAdviseMaxWorkers(pThis->pWtpReg, iMaxWorkers); /* disk queues have always one worker */ } @@ -152,11 +158,11 @@ static inline rsRetVal queueAdviseMaxWorkers(queue_t *pThis) * rgerhards, 2008-02-27 */ static rsRetVal -queueWaitDAModeInitialized(queue_t *pThis) +qqueueWaitDAModeInitialized(qqueue_t *pThis) { DEFiRet; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); ASSERT(pThis->bRunsDA); while(pThis->bRunsDA != 2) { @@ -178,17 +184,17 @@ queueWaitDAModeInitialized(queue_t *pThis) * rgerhards, 2008-01-15 */ static rsRetVal -queueTurnOffDAMode(queue_t *pThis) +qqueueTurnOffDAMode(qqueue_t *pThis) { DEFiRet; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); ASSERT(pThis->bRunsDA); /* at this point, we need a fully initialized DA queue. So if it isn't, we finally need * to wait for its startup... -- rgerhards, 2008-01-25 */ - queueWaitDAModeInitialized(pThis); + qqueueWaitDAModeInitialized(pThis); /* if we need to pull any data that we still need from the (child) disk queue, * now would be the time to do so. At present, we do not need this, but I'd like to @@ -207,15 +213,15 @@ queueTurnOffDAMode(queue_t *pThis) /* we destruct the queue object, which will also shutdown the queue worker. As the queue is empty, * this will be quick. */ - queueDestruct(&pThis->pqDA); /* and now we are ready to destruct the DA queue */ + qqueueDestruct(&pThis->pqDA); /* and now we are ready to destruct the DA queue */ dbgoprint((obj_t*) pThis, "disk-assistance has been turned off, disk queue was empty (iRet %d)\n", iRet); /* now we need to check if the regular queue has some messages. This may be the case * when it is waiting that the high water mark is reached again. If so, we need to start up * a regular worker. -- rgerhards, 2008-01-26 */ - if(queueGetOverallQueueSize(pThis) > 0) { - queueAdviseMaxWorkers(pThis); + if(qqueueGetOverallQueueSize(pThis) > 0) { + qqueueAdviseMaxWorkers(pThis); } } @@ -231,11 +237,11 @@ queueTurnOffDAMode(queue_t *pThis) * rgerhards, 2008-01-14 */ static rsRetVal -queueChkIsDA(queue_t *pThis) +qqueueChkIsDA(qqueue_t *pThis) { DEFiRet; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); if(pThis->pszFilePrefix != NULL) { pThis->bIsDA = 1; dbgoprint((obj_t*) pThis, "is disk-assisted, disk will be used on demand\n"); @@ -259,18 +265,18 @@ queueChkIsDA(queue_t *pThis) * rgerhards, 2008-01-15 */ static rsRetVal -queueStartDA(queue_t *pThis) +qqueueStartDA(qqueue_t *pThis) { DEFiRet; uchar pszDAQName[128]; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); if(pThis->bRunsDA == 2) /* check if already in (fully initialized) DA mode... */ FINALIZE; /* ... then we are already done! */ /* create message queue */ - CHKiRet(queueConstruct(&pThis->pqDA, QUEUETYPE_DISK , 1, 0, pThis->pConsumer)); + CHKiRet(qqueueConstruct(&pThis->pqDA, QUEUETYPE_DISK , 1, 0, pThis->pConsumer)); /* give it a name */ snprintf((char*) pszDAQName, sizeof(pszDAQName)/sizeof(uchar), "%s[DA]", obj.GetName((obj_t*) pThis)); @@ -281,30 +287,30 @@ queueStartDA(queue_t *pThis) */ pThis->pqDA->pqParent = pThis; - CHKiRet(queueSetpUsr(pThis->pqDA, pThis->pUsr)); - CHKiRet(queueSetsizeOnDiskMax(pThis->pqDA, pThis->sizeOnDiskMax)); - CHKiRet(queueSetiDeqSlowdown(pThis->pqDA, pThis->iDeqSlowdown)); - CHKiRet(queueSetMaxFileSize(pThis->pqDA, pThis->iMaxFileSize)); - CHKiRet(queueSetFilePrefix(pThis->pqDA, pThis->pszFilePrefix, pThis->lenFilePrefix)); - CHKiRet(queueSetiPersistUpdCnt(pThis->pqDA, pThis->iPersistUpdCnt)); - CHKiRet(queueSettoActShutdown(pThis->pqDA, pThis->toActShutdown)); - CHKiRet(queueSettoEnq(pThis->pqDA, pThis->toEnq)); - CHKiRet(queueSetEnqOnly(pThis->pqDA, pThis->bDAEnqOnly, MUTEX_ALREADY_LOCKED)); - CHKiRet(queueSetiDeqtWinFromHr(pThis->pqDA, pThis->iDeqtWinFromHr)); - CHKiRet(queueSetiDeqtWinToHr(pThis->pqDA, pThis->iDeqtWinToHr)); - CHKiRet(queueSetiHighWtrMrk(pThis->pqDA, 0)); - CHKiRet(queueSetiDiscardMrk(pThis->pqDA, 0)); + CHKiRet(qqueueSetpUsr(pThis->pqDA, pThis->pUsr)); + CHKiRet(qqueueSetsizeOnDiskMax(pThis->pqDA, pThis->sizeOnDiskMax)); + CHKiRet(qqueueSetiDeqSlowdown(pThis->pqDA, pThis->iDeqSlowdown)); + CHKiRet(qqueueSetMaxFileSize(pThis->pqDA, pThis->iMaxFileSize)); + CHKiRet(qqueueSetFilePrefix(pThis->pqDA, pThis->pszFilePrefix, pThis->lenFilePrefix)); + CHKiRet(qqueueSetiPersistUpdCnt(pThis->pqDA, pThis->iPersistUpdCnt)); + CHKiRet(qqueueSettoActShutdown(pThis->pqDA, pThis->toActShutdown)); + CHKiRet(qqueueSettoEnq(pThis->pqDA, pThis->toEnq)); + CHKiRet(qqueueSetEnqOnly(pThis->pqDA, pThis->bDAEnqOnly, MUTEX_ALREADY_LOCKED)); + CHKiRet(qqueueSetiDeqtWinFromHr(pThis->pqDA, pThis->iDeqtWinFromHr)); + CHKiRet(qqueueSetiDeqtWinToHr(pThis->pqDA, pThis->iDeqtWinToHr)); + CHKiRet(qqueueSetiHighWtrMrk(pThis->pqDA, 0)); + CHKiRet(qqueueSetiDiscardMrk(pThis->pqDA, 0)); if(pThis->toQShutdown == 0) { - CHKiRet(queueSettoQShutdown(pThis->pqDA, 0)); /* if the user really wants... */ + CHKiRet(qqueueSettoQShutdown(pThis->pqDA, 0)); /* if the user really wants... */ } else { /* we use the shortest possible shutdown (0 is endless!) because when we run on disk AND * have an obviously large backlog, we can't finish it in any case. So there is no point * in holding shutdown longer than necessary. -- rgerhards, 2008-01-15 */ - CHKiRet(queueSettoQShutdown(pThis->pqDA, 1)); + CHKiRet(qqueueSettoQShutdown(pThis->pqDA, 1)); } - iRet = queueStart(pThis->pqDA); + iRet = qqueueStart(pThis->pqDA); /* file not found is expected, that means it is no previous QIF available */ if(iRet != RS_RET_OK && iRet != RS_RET_FILE_NOT_FOUND) FINALIZE; /* something is wrong */ @@ -322,12 +328,12 @@ queueStartDA(queue_t *pThis) pthread_cond_broadcast(&pThis->condDAReady); /* signal we are now initialized and ready to go ;) */ dbgoprint((obj_t*) pThis, "is now running in disk assisted mode, disk queue 0x%lx\n", - queueGetID(pThis->pqDA)); + qqueueGetID(pThis->pqDA)); finalize_it: if(iRet != RS_RET_OK) { if(pThis->pqDA != NULL) { - queueDestruct(&pThis->pqDA); + qqueueDestruct(&pThis->pqDA); } dbgoprint((obj_t*) pThis, "error %d creating disk queue - giving up.\n", iRet); pThis->bIsDA = 0; @@ -344,7 +350,7 @@ finalize_it: * rgerhards, 2008-01-16 */ static inline rsRetVal -queueInitDA(queue_t *pThis, int bEnqOnly, int bLockMutex) +qqueueInitDA(qqueue_t *pThis, int bEnqOnly, int bLockMutex) { DEFiRet; DEFVARS_mutexProtection; @@ -362,12 +368,12 @@ queueInitDA(queue_t *pThis, int bEnqOnly, int bLockMutex) lenBuf = snprintf((char*)pszBuf, sizeof(pszBuf), "%s:DA", obj.GetName((obj_t*) pThis)); CHKiRet(wtpConstruct (&pThis->pWtpDA)); CHKiRet(wtpSetDbgHdr (pThis->pWtpDA, pszBuf, lenBuf)); - CHKiRet(wtpSetpfChkStopWrkr (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, int)) queueChkStopWrkrDA)); - CHKiRet(wtpSetpfIsIdle (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, int)) queueIsIdleDA)); - CHKiRet(wtpSetpfDoWork (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, void *pWti, int)) queueConsumerDA)); - CHKiRet(wtpSetpfOnWorkerCancel (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, void*pWti)) queueConsumerCancelCleanup)); - CHKiRet(wtpSetpfOnWorkerStartup (pThis->pWtpDA, (rsRetVal (*)(void *pUsr)) queueStartDA)); - CHKiRet(wtpSetpfOnWorkerShutdown(pThis->pWtpDA, (rsRetVal (*)(void *pUsr)) queueTurnOffDAMode)); + CHKiRet(wtpSetpfChkStopWrkr (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, int)) qqueueChkStopWrkrDA)); + CHKiRet(wtpSetpfIsIdle (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, int)) qqueueIsIdleDA)); + CHKiRet(wtpSetpfDoWork (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, void *pWti, int)) qqueueConsumerDA)); + CHKiRet(wtpSetpfOnWorkerCancel (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, void*pWti)) qqueueConsumerCancelCleanup)); + CHKiRet(wtpSetpfOnWorkerStartup (pThis->pWtpDA, (rsRetVal (*)(void *pUsr)) qqueueStartDA)); + CHKiRet(wtpSetpfOnWorkerShutdown(pThis->pWtpDA, (rsRetVal (*)(void *pUsr)) qqueueTurnOffDAMode)); CHKiRet(wtpSetpmutUsr (pThis->pWtpDA, pThis->mut)); CHKiRet(wtpSetpcondBusy (pThis->pWtpDA, &pThis->notEmpty)); CHKiRet(wtpSetiNumWorkerThreads (pThis->pWtpDA, 1)); @@ -400,14 +406,14 @@ finalize_it: * rgerhards, 2008-01-14 */ static inline rsRetVal -queueChkStrtDA(queue_t *pThis) +qqueueChkStrtDA(qqueue_t *pThis) { DEFiRet; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); /* if we do not hit the high water mark, we have nothing to do */ - if(queueGetOverallQueueSize(pThis) != pThis->iHighWtrMrk) + if(qqueueGetOverallQueueSize(pThis) != pThis->iHighWtrMrk) ABORT_FINALIZE(RS_RET_OK); if(pThis->bRunsDA) { @@ -421,15 +427,15 @@ queueChkStrtDA(queue_t *pThis) * we need at least one). */ dbgoprint((obj_t*) pThis, "%d entries - passed high water mark in DA mode, send notify\n", - queueGetOverallQueueSize(pThis)); - queueAdviseMaxWorkers(pThis); + qqueueGetOverallQueueSize(pThis)); + qqueueAdviseMaxWorkers(pThis); } else { /* this is the case when we are currently not running in DA mode. So it is time * to turn it back on. */ dbgoprint((obj_t*) pThis, "%d entries - passed high water mark for disk-assisted mode, initiating...\n", - queueGetOverallQueueSize(pThis)); - queueInitDA(pThis, QUEUE_MODE_ENQDEQ, MUTEX_ALREADY_LOCKED); /* initiate DA mode */ + qqueueGetOverallQueueSize(pThis)); + qqueueInitDA(pThis, QUEUE_MODE_ENQDEQ, MUTEX_ALREADY_LOCKED); /* initiate DA mode */ } finalize_it: @@ -447,7 +453,7 @@ finalize_it: */ /* -------------------- fixed array -------------------- */ -static rsRetVal qConstructFixedArray(queue_t *pThis) +static rsRetVal qConstructFixedArray(qqueue_t *pThis) { DEFiRet; @@ -463,14 +469,14 @@ static rsRetVal qConstructFixedArray(queue_t *pThis) pThis->tVars.farray.head = 0; pThis->tVars.farray.tail = 0; - queueChkIsDA(pThis); + qqueueChkIsDA(pThis); finalize_it: RETiRet; } -static rsRetVal qDestructFixedArray(queue_t *pThis) +static rsRetVal qDestructFixedArray(qqueue_t *pThis) { DEFiRet; @@ -485,7 +491,7 @@ static rsRetVal qDestructFixedArray(queue_t *pThis) } -static rsRetVal qAddFixedArray(queue_t *pThis, void* in) +static rsRetVal qAddFixedArray(qqueue_t *pThis, void* in) { DEFiRet; @@ -498,7 +504,7 @@ static rsRetVal qAddFixedArray(queue_t *pThis, void* in) RETiRet; } -static rsRetVal qDelFixedArray(queue_t *pThis, void **out) +static rsRetVal qDelFixedArray(qqueue_t *pThis, void **out) { DEFiRet; @@ -517,7 +523,7 @@ static rsRetVal qDelFixedArray(queue_t *pThis, void **out) /* first some generic functions which are also used for the unget linked list */ -static inline rsRetVal queueAddLinkedList(qLinkedList_t **ppRoot, qLinkedList_t **ppLast, void* pUsr) +static inline rsRetVal qqueueAddLinkedList(qLinkedList_t **ppRoot, qLinkedList_t **ppLast, void* pUsr) { DEFiRet; qLinkedList_t *pEntry; @@ -543,7 +549,7 @@ finalize_it: RETiRet; } -static inline rsRetVal queueDelLinkedList(qLinkedList_t **ppRoot, qLinkedList_t **ppLast, obj_t **ppUsr) +static inline rsRetVal qqueueDelLinkedList(qLinkedList_t **ppRoot, qLinkedList_t **ppLast, obj_t **ppUsr) { DEFiRet; qLinkedList_t *pEntry; @@ -570,7 +576,7 @@ static inline rsRetVal queueDelLinkedList(qLinkedList_t **ppRoot, qLinkedList_t /* end generic functions which are also used for the unget linked list */ -static rsRetVal qConstructLinkedList(queue_t *pThis) +static rsRetVal qConstructLinkedList(qqueue_t *pThis) { DEFiRet; @@ -579,13 +585,13 @@ static rsRetVal qConstructLinkedList(queue_t *pThis) pThis->tVars.linklist.pRoot = 0; pThis->tVars.linklist.pLast = 0; - queueChkIsDA(pThis); + qqueueChkIsDA(pThis); RETiRet; } -static rsRetVal qDestructLinkedList(queue_t __attribute__((unused)) *pThis) +static rsRetVal qDestructLinkedList(qqueue_t __attribute__((unused)) *pThis) { DEFiRet; @@ -598,11 +604,11 @@ static rsRetVal qDestructLinkedList(queue_t __attribute__((unused)) *pThis) RETiRet; } -static rsRetVal qAddLinkedList(queue_t *pThis, void* pUsr) +static rsRetVal qAddLinkedList(qqueue_t *pThis, void* pUsr) { DEFiRet; - iRet = queueAddLinkedList(&pThis->tVars.linklist.pRoot, &pThis->tVars.linklist.pLast, pUsr); + iRet = qqueueAddLinkedList(&pThis->tVars.linklist.pRoot, &pThis->tVars.linklist.pLast, pUsr); #if 0 qLinkedList_t *pEntry; @@ -626,10 +632,10 @@ finalize_it: RETiRet; } -static rsRetVal qDelLinkedList(queue_t *pThis, obj_t **ppUsr) +static rsRetVal qDelLinkedList(qqueue_t *pThis, obj_t **ppUsr) { DEFiRet; - iRet = queueDelLinkedList(&pThis->tVars.linklist.pRoot, &pThis->tVars.linklist.pLast, ppUsr); + iRet = qqueueDelLinkedList(&pThis->tVars.linklist.pRoot, &pThis->tVars.linklist.pLast, ppUsr); #if 0 qLinkedList_t *pEntry; @@ -656,11 +662,11 @@ static rsRetVal qDelLinkedList(queue_t *pThis, obj_t **ppUsr) static rsRetVal -queueLoadPersStrmInfoFixup(strm_t *pStrm, queue_t __attribute__((unused)) *pThis) +qqueueLoadPersStrmInfoFixup(strm_t *pStrm, qqueue_t __attribute__((unused)) *pThis) { DEFiRet; ISOBJ_TYPE_assert(pStrm, strm); - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); CHKiRet(strmSetDir(pStrm, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir()))); finalize_it: RETiRet; @@ -672,14 +678,14 @@ finalize_it: * rgerhards, 2008-01-15 */ static rsRetVal -queueHaveQIF(queue_t *pThis) +qqueueHaveQIF(qqueue_t *pThis) { DEFiRet; uchar pszQIFNam[MAXFNAME]; size_t lenQIFNam; struct stat stat_buf; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); if(pThis->pszFilePrefix == NULL) ABORT_FINALIZE(RS_RET_NO_FILEPREFIX); @@ -709,7 +715,7 @@ finalize_it: * rgerhards, 2008-01-11 */ static rsRetVal -queueTryLoadPersistedInfo(queue_t *pThis) +qqueueTryLoadPersistedInfo(qqueue_t *pThis) { DEFiRet; strm_t *psQIF = NULL; @@ -719,7 +725,7 @@ queueTryLoadPersistedInfo(queue_t *pThis) int iUngottenObjs; obj_t *pUsr; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); /* Construct file name */ lenQIFNam = snprintf((char*)pszQIFNam, sizeof(pszQIFNam) / sizeof(uchar), "%s/%s.qi", @@ -754,15 +760,15 @@ queueTryLoadPersistedInfo(queue_t *pThis) while(iUngottenObjs > 0) { /* fill the queue from disk */ CHKiRet(obj.Deserialize((void*) &pUsr, (uchar*)"msg", psQIF, NULL, NULL)); - queueUngetObj(pThis, pUsr, MUTEX_ALREADY_LOCKED); + qqueueUngetObj(pThis, pUsr, MUTEX_ALREADY_LOCKED); --iUngottenObjs; /* one less */ } /* and now the stream objects (some order as when persisted!) */ CHKiRet(obj.Deserialize(&pThis->tVars.disk.pWrite, (uchar*) "strm", psQIF, - (rsRetVal(*)(obj_t*,void*))queueLoadPersStrmInfoFixup, pThis)); + (rsRetVal(*)(obj_t*,void*))qqueueLoadPersStrmInfoFixup, pThis)); CHKiRet(obj.Deserialize(&pThis->tVars.disk.pRead, (uchar*) "strm", psQIF, - (rsRetVal(*)(obj_t*,void*))queueLoadPersStrmInfoFixup, pThis)); + (rsRetVal(*)(obj_t*,void*))qqueueLoadPersStrmInfoFixup, pThis)); CHKiRet(strmSeekCurrOffs(pThis->tVars.disk.pWrite)); CHKiRet(strmSeekCurrOffs(pThis->tVars.disk.pRead)); @@ -792,7 +798,7 @@ finalize_it: * allowed file size at this point - that should be a config setting... * rgerhards, 2008-01-10 */ -static rsRetVal qConstructDisk(queue_t *pThis) +static rsRetVal qConstructDisk(qqueue_t *pThis) { DEFiRet; int bRestarted = 0; @@ -800,7 +806,7 @@ static rsRetVal qConstructDisk(queue_t *pThis) ASSERT(pThis != NULL); /* and now check if there is some persistent information that needs to be read in */ - iRet = queueTryLoadPersistedInfo(pThis); + iRet = qqueueTryLoadPersistedInfo(pThis); if(iRet == RS_RET_OK) bRestarted = 1; else if(iRet != RS_RET_FILE_NOT_FOUND) @@ -842,7 +848,7 @@ finalize_it: } -static rsRetVal qDestructDisk(queue_t *pThis) +static rsRetVal qDestructDisk(qqueue_t *pThis) { DEFiRet; @@ -854,7 +860,7 @@ static rsRetVal qDestructDisk(queue_t *pThis) RETiRet; } -static rsRetVal qAddDisk(queue_t *pThis, void* pUsr) +static rsRetVal qAddDisk(qqueue_t *pThis, void* pUsr) { DEFiRet; number_t nWriteCount; @@ -881,7 +887,7 @@ finalize_it: RETiRet; } -static rsRetVal qDelDisk(queue_t *pThis, void **ppUsr) +static rsRetVal qDelDisk(qqueue_t *pThis, void **ppUsr) { DEFiRet; @@ -912,18 +918,18 @@ finalize_it: } /* -------------------- direct (no queueing) -------------------- */ -static rsRetVal qConstructDirect(queue_t __attribute__((unused)) *pThis) +static rsRetVal qConstructDirect(qqueue_t __attribute__((unused)) *pThis) { return RS_RET_OK; } -static rsRetVal qDestructDirect(queue_t __attribute__((unused)) *pThis) +static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis) { return RS_RET_OK; } -static rsRetVal qAddDirect(queue_t *pThis, void* pUsr) +static rsRetVal qAddDirect(qqueue_t *pThis, void* pUsr) { DEFiRet; @@ -940,7 +946,7 @@ static rsRetVal qAddDirect(queue_t *pThis, void* pUsr) RETiRet; } -static rsRetVal qDelDirect(queue_t __attribute__((unused)) *pThis, __attribute__((unused)) void **out) +static rsRetVal qDelDirect(qqueue_t __attribute__((unused)) *pThis, __attribute__((unused)) void **out) { return RS_RET_OK; } @@ -955,12 +961,12 @@ static rsRetVal qDelDirect(queue_t __attribute__((unused)) *pThis, __attribute__ * rgerhards, 2008-01-20 */ static rsRetVal -queueUngetObj(queue_t *pThis, obj_t *pUsr, int bLockMutex) +qqueueUngetObj(qqueue_t *pThis, obj_t *pUsr, int bLockMutex) { DEFiRet; DEFVARS_mutexProtection; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); ISOBJ_assert(pUsr); /* TODO: we aborted right at this place at least 3 times -- race? 2008-02-28, -03-10, -03-15 The second time I noticed it the queue was in destruction with NO worker threads running. The pUsr ptr was totally off and provided no clue what it may be pointing @@ -969,7 +975,7 @@ queueUngetObj(queue_t *pThis, obj_t *pUsr, int bLockMutex) dbgoprint((obj_t*) pThis, "ungetting user object %s\n", obj.GetName(pUsr)); BEGIN_MTX_PROTECTED_OPERATIONS(pThis->mut, bLockMutex); - iRet = queueAddLinkedList(&pThis->pUngetRoot, &pThis->pUngetLast, pUsr); + iRet = qqueueAddLinkedList(&pThis->pUngetRoot, &pThis->pUngetLast, pUsr); ++pThis->iUngottenObjs; /* indicate one more */ END_MTX_PROTECTED_OPERATIONS(pThis->mut); @@ -985,14 +991,14 @@ queueUngetObj(queue_t *pThis, obj_t *pUsr, int bLockMutex) * rgerhards, 2008-01-29 */ static rsRetVal -queueGetUngottenObj(queue_t *pThis, obj_t **ppUsr) +qqueueGetUngottenObj(qqueue_t *pThis, obj_t **ppUsr) { DEFiRet; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); ASSERT(ppUsr != NULL); - iRet = queueDelLinkedList(&pThis->pUngetRoot, &pThis->pUngetLast, ppUsr); + iRet = qqueueDelLinkedList(&pThis->pUngetRoot, &pThis->pUngetLast, ppUsr); --pThis->iUngottenObjs; /* indicate one less */ dbgoprint((obj_t*) pThis, "dequeued ungotten user object %s\n", obj.GetName(*ppUsr)); @@ -1006,7 +1012,7 @@ queueGetUngottenObj(queue_t *pThis, obj_t **ppUsr) * things truely different. -- rgerhards, 2008-02-12 */ static rsRetVal -queueAdd(queue_t *pThis, void *pUsr) +qqueueAdd(qqueue_t *pThis, void *pUsr) { DEFiRet; @@ -1015,7 +1021,7 @@ queueAdd(queue_t *pThis, void *pUsr) CHKiRet(pThis->qAdd(pThis, pUsr)); if(pThis->qType != QUEUETYPE_DIRECT) { - ++pThis->iQueueSize; + ATOMIC_INC(pThis->iQueueSize); dbgoprint((obj_t*) pThis, "entry added, size now %d entries\n", pThis->iQueueSize); } @@ -1029,7 +1035,7 @@ finalize_it: * ungotten list and, if so, dequeue it first. */ static rsRetVal -queueDel(queue_t *pThis, void *pUsr) +qqueueDel(qqueue_t *pThis, void *pUsr) { DEFiRet; @@ -1041,10 +1047,10 @@ queueDel(queue_t *pThis, void *pUsr) * losing the whole process because it loops... -- rgerhards, 2008-01-03 */ if(pThis->iUngottenObjs > 0) { - iRet = queueGetUngottenObj(pThis, (obj_t**) pUsr); + iRet = qqueueGetUngottenObj(pThis, (obj_t**) pUsr); } else { iRet = pThis->qDel(pThis, pUsr); - --pThis->iQueueSize; + ATOMIC_DEC(pThis->iQueueSize); } dbgoprint((obj_t*) pThis, "entry deleted, state %d, size now %d entries\n", @@ -1065,14 +1071,14 @@ queueDel(queue_t *pThis, void *pUsr) * complex) if each would have its own shutdown. The function does not self check * this condition - the caller must make sure it is not called with a parent. */ -static rsRetVal queueShutdownWorkers(queue_t *pThis) +static rsRetVal qqueueShutdownWorkers(qqueue_t *pThis) { DEFiRet; DEFVARS_mutexProtection; struct timespec tTimeout; rsRetVal iRetLocal; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); ASSERT(pThis->pqParent == NULL); /* detect invalid calling sequence */ dbgoprint((obj_t*) pThis, "initiating worker thread shutdown sequence\n"); @@ -1086,7 +1092,7 @@ static rsRetVal queueShutdownWorkers(queue_t *pThis) /* first try to shutdown the queue within the regular shutdown period */ BEGIN_MTX_PROTECTED_OPERATIONS(pThis->mut, LOCK_MUTEX); /* some workers may be running in parallel! */ - if(queueGetOverallQueueSize(pThis) > 0) { + if(qqueueGetOverallQueueSize(pThis) > 0) { if(pThis->bRunsDA) { /* We may have waited on the low water mark. As it may have changed, we * see if we reactivate the worker. @@ -1124,7 +1130,7 @@ static rsRetVal queueShutdownWorkers(queue_t *pThis) if(pThis->bRunsDA) { END_MTX_PROTECTED_OPERATIONS(pThis->mut); dbgoprint((obj_t*) pThis, "we have a DA queue (0x%lx), requesting its shutdown.\n", - queueGetID(pThis->pqDA)); + qqueueGetID(pThis->pqDA)); /* we use the same absolute timeout as above, so we do not use more than the configured * timeout interval! */ @@ -1153,19 +1159,19 @@ static rsRetVal queueShutdownWorkers(queue_t *pThis) /* at this stage, we need to have the DA worker properly initialized and running (if there is one) */ if(pThis->bRunsDA) - queueWaitDAModeInitialized(pThis); + qqueueWaitDAModeInitialized(pThis); BEGIN_MTX_PROTECTED_OPERATIONS(pThis->mut, LOCK_MUTEX); /* some workers may be running in parallel! */ /* optimize parameters for shutdown of DA-enabled queues */ - if(pThis->bIsDA && queueGetOverallQueueSize(pThis) > 0 && pThis->bSaveOnShutdown) { + if(pThis->bIsDA && qqueueGetOverallQueueSize(pThis) > 0 && pThis->bSaveOnShutdown) { /* switch to enqueue-only mode so that no more actions happen */ if(pThis->bRunsDA == 0) { - queueInitDA(pThis, QUEUE_MODE_ENQONLY, MUTEX_ALREADY_LOCKED); /* switch to DA mode */ + qqueueInitDA(pThis, QUEUE_MODE_ENQONLY, MUTEX_ALREADY_LOCKED); /* switch to DA mode */ } else { /* TODO: RACE: we may reach this point when the DA worker has been initialized (state 1) * but is not yet running (state 2). In this case, pThis->pqDA is NULL! rgerhards, 2008-02-27 */ - queueSetEnqOnly(pThis->pqDA, QUEUE_MODE_ENQONLY, MUTEX_ALREADY_LOCKED); /* switch to enqueue-only mode */ + qqueueSetEnqOnly(pThis->pqDA, QUEUE_MODE_ENQONLY, MUTEX_ALREADY_LOCKED); /* switch to enqueue-only mode */ } END_MTX_PROTECTED_OPERATIONS(pThis->mut); /* make sure we do not timeout before we are done */ @@ -1187,7 +1193,7 @@ static rsRetVal queueShutdownWorkers(queue_t *pThis) * they will automatically terminate as there no longer is any message left to process. */ BEGIN_MTX_PROTECTED_OPERATIONS(pThis->mut, LOCK_MUTEX); /* some workers may be running in parallel! */ - if(queueGetOverallQueueSize(pThis) > 0) { + if(qqueueGetOverallQueueSize(pThis) > 0) { timeoutComp(&tTimeout, pThis->toActShutdown); if(wtpGetCurNumWrkr(pThis->pWtpReg, LOCK_MUTEX) > 0) { END_MTX_PROTECTED_OPERATIONS(pThis->mut); @@ -1256,7 +1262,7 @@ static rsRetVal queueShutdownWorkers(queue_t *pThis) * Well, more precisely, they *are in termination*. Some cancel cleanup handlers * may still be running. */ - dbgoprint((obj_t*) pThis, "worker threads terminated, remaining queue size %d.\n", queueGetOverallQueueSize(pThis)); + dbgoprint((obj_t*) pThis, "worker threads terminated, remaining queue size %d.\n", qqueueGetOverallQueueSize(pThis)); RETiRet; } @@ -1268,22 +1274,23 @@ static rsRetVal queueShutdownWorkers(queue_t *pThis) * is done by queueStart(). The reason is that we want to give the caller a chance * to modify some parameters before the queue is actually started. */ -rsRetVal queueConstruct(queue_t **ppThis, queueType_t qType, int iWorkerThreads, +rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThreads, int iMaxQueueSize, rsRetVal (*pConsumer)(void*,void*)) { DEFiRet; - queue_t *pThis; + qqueue_t *pThis; ASSERT(ppThis != NULL); ASSERT(pConsumer != NULL); ASSERT(iWorkerThreads >= 0); - if((pThis = (queue_t *)calloc(1, sizeof(queue_t))) == NULL) { + if((pThis = (qqueue_t *)calloc(1, sizeof(qqueue_t))) == NULL) { ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); } /* we have an object, so let's fill the properties */ objConstructSetObjInfo(pThis); + pThis->bOptimizeUniProc = glbl.GetOptimizeUniProc(); if((pThis->pszSpoolDir = (uchar*) strdup((char*)glbl.GetWorkDir())) == NULL) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); @@ -1314,7 +1321,7 @@ rsRetVal queueConstruct(queue_t **ppThis, queueType_t qType, int iWorkerThreads, pThis->qConstruct = qConstructLinkedList; pThis->qDestruct = qDestructLinkedList; pThis->qAdd = qAddLinkedList; - pThis->qDel = (rsRetVal (*)(queue_t*,void**)) qDelLinkedList; + pThis->qDel = (rsRetVal (*)(qqueue_t*,void**)) qDelLinkedList; break; case QUEUETYPE_DISK: pThis->qConstruct = qConstructDisk; @@ -1341,25 +1348,25 @@ finalize_it: /* cancellation cleanup handler for queueWorker () * Updates admin structure and frees ressources. * Params: - * arg1 - user pointer (in this case a queue_t) + * arg1 - user pointer (in this case a qqueue_t) * arg2 - user data pointer (in this case a queue data element, any object [queue's pUsr ptr!]) * Note that arg2 may be NULL, in which case no dequeued but unprocessed pUsr exists! * rgerhards, 2008-01-16 */ static rsRetVal -queueConsumerCancelCleanup(void *arg1, void *arg2) +qqueueConsumerCancelCleanup(void *arg1, void *arg2) { DEFiRet; - queue_t *pThis = (queue_t*) arg1; + qqueue_t *pThis = (qqueue_t*) arg1; obj_t *pUsr = (obj_t*) arg2; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); if(pUsr != NULL) { /* make sure the data element is not lost */ dbgoprint((obj_t*) pThis, "cancelation cleanup handler consumer called, we need to unget one user data element\n"); - CHKiRet(queueUngetObj(pThis, pUsr, LOCK_MUTEX)); + CHKiRet(qqueueUngetObj(pThis, pUsr, LOCK_MUTEX)); } finalize_it: @@ -1381,13 +1388,13 @@ finalize_it: * the return state! * rgerhards, 2008-01-24 */ -static int queueChkDiscardMsg(queue_t *pThis, int iQueueSize, int bRunsDA, void *pUsr) +static int qqueueChkDiscardMsg(qqueue_t *pThis, int iQueueSize, int bRunsDA, void *pUsr) { DEFiRet; rsRetVal iRetLocal; int iSeverity; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); ISOBJ_assert(pUsr); if(pThis->iDiscardMrk > 0 && iQueueSize >= pThis->iDiscardMrk && bRunsDA == 0) { @@ -1412,7 +1419,7 @@ finalize_it: * rgerhards, 2008-10-21 */ static rsRetVal -queueDequeueConsumable(queue_t *pThis, wti_t *pWti, int iCancelStateSave) +qqueueDequeueConsumable(qqueue_t *pThis, wti_t *pWti, int iCancelStateSave) { DEFiRet; void *pUsr; @@ -1420,9 +1427,9 @@ queueDequeueConsumable(queue_t *pThis, wti_t *pWti, int iCancelStateSave) int bRunsDA; /* cache for early mutex release */ /* dequeue element (still protected from mutex) */ - iRet = queueDel(pThis, &pUsr); - queueChkPersist(pThis); - iQueueSize = queueGetOverallQueueSize(pThis); /* cache this for after mutex release */ + iRet = qqueueDel(pThis, &pUsr); + qqueueChkPersist(pThis); + iQueueSize = qqueueGetOverallQueueSize(pThis); /* cache this for after mutex release */ bRunsDA = pThis->bRunsDA; /* cache this for after mutex release */ /* We now need to save the user pointer for the cancel cleanup handler, BUT ONLY @@ -1473,7 +1480,7 @@ queueDequeueConsumable(queue_t *pThis, wti_t *pWti, int iCancelStateSave) * provide real-time creation of spool files. * Note: It is OK to use the cached iQueueSize here, because it does not hurt if it is slightly wrong. */ - CHKiRet(queueChkDiscardMsg(pThis, iQueueSize, bRunsDA, pUsr)); + CHKiRet(qqueueChkDiscardMsg(pThis, iQueueSize, bRunsDA, pUsr)); finalize_it: if(iRet != RS_RET_OK && iRet != RS_RET_DISCARDMSG) { @@ -1522,7 +1529,7 @@ finalize_it: * but you get the idea from the code above. */ static rsRetVal -queueRateLimiter(queue_t *pThis) +qqueueRateLimiter(qqueue_t *pThis) { DEFiRet; int iDelay; @@ -1530,9 +1537,7 @@ queueRateLimiter(queue_t *pThis) time_t tCurr; struct tm m; - ISOBJ_TYPE_assert(pThis, queue); - - dbgoprint((obj_t*) pThis, "entering rate limiter\n"); + ISOBJ_TYPE_assert(pThis, qqueue); iDelay = 0; if(pThis->iDeqtWinToHr != 25) { /* 25 means disabled */ @@ -1587,14 +1592,14 @@ queueRateLimiter(queue_t *pThis) * rgerhards, 2008-01-21 */ static rsRetVal -queueConsumerReg(queue_t *pThis, wti_t *pWti, int iCancelStateSave) +qqueueConsumerReg(qqueue_t *pThis, wti_t *pWti, int iCancelStateSave) { DEFiRet; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); ISOBJ_TYPE_assert(pWti, wti); - CHKiRet(queueDequeueConsumable(pThis, pWti, iCancelStateSave)); + CHKiRet(qqueueDequeueConsumable(pThis, pWti, iCancelStateSave)); CHKiRet(pThis->pConsumer(pThis->pUsr, pWti->pUsrp)); /* we now need to check if we should deliberately delay processing a bit @@ -1621,15 +1626,15 @@ finalize_it: * rgerhards, 2008-01-14 */ static rsRetVal -queueConsumerDA(queue_t *pThis, wti_t *pWti, int iCancelStateSave) +qqueueConsumerDA(qqueue_t *pThis, wti_t *pWti, int iCancelStateSave) { DEFiRet; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); ISOBJ_TYPE_assert(pWti, wti); - CHKiRet(queueDequeueConsumable(pThis, pWti, iCancelStateSave)); - CHKiRet(queueEnqObj(pThis->pqDA, eFLOWCTL_NO_DELAY, pWti->pUsrp)); + CHKiRet(qqueueDequeueConsumable(pThis, pWti, iCancelStateSave)); + CHKiRet(qqueueEnqObj(pThis->pqDA, eFLOWCTL_NO_DELAY, pWti->pUsrp)); finalize_it: dbgoprint((obj_t*) pThis, "DAConsumer returns with iRet %d\n", iRet); @@ -1645,7 +1650,7 @@ finalize_it: * the DA queue */ static int -queueChkStopWrkrDA(queue_t *pThis) +qqueueChkStopWrkrDA(qqueue_t *pThis) { /* if our queue is in destruction, we drain to the DA queue and so we shall not terminate * until we have done so. @@ -1664,7 +1669,7 @@ queueChkStopWrkrDA(queue_t *pThis) && pThis->pqDA->tVars.disk.sizeOnDisk > pThis->pqDA->sizeOnDiskMax) { /* this queue can never grow, so we can give up... */ bStopWrkr = 1; - } else if(queueGetOverallQueueSize(pThis) < pThis->iHighWtrMrk && pThis->bQueueStarted == 1) { + } else if(qqueueGetOverallQueueSize(pThis) < pThis->iHighWtrMrk && pThis->bQueueStarted == 1) { bStopWrkr = 1; } else { bStopWrkr = 0; @@ -1687,9 +1692,9 @@ queueChkStopWrkrDA(queue_t *pThis) * the DA queue */ static int -queueChkStopWrkrReg(queue_t *pThis) +qqueueChkStopWrkrReg(qqueue_t *pThis) { - return pThis->bEnqOnly || pThis->bRunsDA || (pThis->pqParent != NULL && queueGetOverallQueueSize(pThis) == 0); + return pThis->bEnqOnly || pThis->bRunsDA || (pThis->pqParent != NULL && qqueueGetOverallQueueSize(pThis) == 0); } @@ -1697,26 +1702,26 @@ queueChkStopWrkrReg(queue_t *pThis) * are not stable! DA queue version */ static int -queueIsIdleDA(queue_t *pThis) +qqueueIsIdleDA(qqueue_t *pThis) { /* remember: iQueueSize is the DA queue size, not the main queue! */ /* TODO: I think we need just a single function for DA and non-DA mode - but I leave it for now as is */ - return(queueGetOverallQueueSize(pThis) == 0 || (pThis->bRunsDA && queueGetOverallQueueSize(pThis) <= pThis->iLowWtrMrk)); + return(qqueueGetOverallQueueSize(pThis) == 0 || (pThis->bRunsDA && qqueueGetOverallQueueSize(pThis) <= pThis->iLowWtrMrk)); } /* must only be called when the queue mutex is locked, else results * are not stable! Regular queue version */ static int -queueIsIdleReg(queue_t *pThis) +qqueueIsIdleReg(qqueue_t *pThis) { #if 0 /* enable for performance testing */ int ret; - ret = queueGetOverallQueueSize(pThis) == 0 || (pThis->bRunsDA && queueGetOverallQueueSize(pThis) <= pThis->iLowWtrMrk); + ret = qqueueGetOverallQueueSize(pThis) == 0 || (pThis->bRunsDA && qqueueGetOverallQueueSize(pThis) <= pThis->iLowWtrMrk); if(ret) fprintf(stderr, "queue is idle\n"); return ret; #else /* regular code! */ - return(queueGetOverallQueueSize(pThis) == 0 || (pThis->bRunsDA && queueGetOverallQueueSize(pThis) <= pThis->iLowWtrMrk)); + return(qqueueGetOverallQueueSize(pThis) == 0 || (pThis->bRunsDA && qqueueGetOverallQueueSize(pThis) <= pThis->iLowWtrMrk)); #endif } @@ -1735,11 +1740,11 @@ queueIsIdleReg(queue_t *pThis) * I am telling this, because I, too, always get confused by those... */ static rsRetVal -queueRegOnWrkrShutdown(queue_t *pThis) +qqueueRegOnWrkrShutdown(qqueue_t *pThis) { DEFiRet; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); if(pThis->pqParent != NULL) { pThis->pqParent->bChildIsDone = 1; /* indicate we are done */ @@ -1756,11 +1761,11 @@ queueRegOnWrkrShutdown(queue_t *pThis) * hook to indicate in the parent queue (if we are a child) that we are not done yet. */ static rsRetVal -queueRegOnWrkrStartup(queue_t *pThis) +qqueueRegOnWrkrStartup(qqueue_t *pThis) { DEFiRet; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); if(pThis->pqParent != NULL) { pThis->pqParent->bChildIsDone = 0; @@ -1773,7 +1778,7 @@ queueRegOnWrkrStartup(queue_t *pThis) /* start up the queue - it must have been constructed and parameters defined * before. */ -rsRetVal queueStart(queue_t *pThis) /* this is the ConstructionFinalizer */ +rsRetVal qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */ { DEFiRet; rsRetVal iRetLocal; @@ -1812,7 +1817,7 @@ rsRetVal queueStart(queue_t *pThis) /* this is the ConstructionFinalizer */ dbgoprint((obj_t*) pThis, "type %d, enq-only %d, disk assisted %d, maxFileSz %lld, qsize %d, child %d, " "full delay %d, light delay %d starting\n", pThis->qType, pThis->bEnqOnly, pThis->bIsDA, pThis->iMaxFileSize, - queueGetOverallQueueSize(pThis), pThis->pqParent == NULL ? 0 : 1, + qqueueGetOverallQueueSize(pThis), pThis->pqParent == NULL ? 0 : 1, pThis->iFullDlyMrk, pThis->iLightDlyMrk); if(pThis->qType == QUEUETYPE_DIRECT) @@ -1824,13 +1829,13 @@ rsRetVal queueStart(queue_t *pThis) /* this is the ConstructionFinalizer */ lenBuf = snprintf((char*)pszBuf, sizeof(pszBuf), "%s:Reg", obj.GetName((obj_t*) pThis)); CHKiRet(wtpConstruct (&pThis->pWtpReg)); CHKiRet(wtpSetDbgHdr (pThis->pWtpReg, pszBuf, lenBuf)); - CHKiRet(wtpSetpfRateLimiter (pThis->pWtpReg, (rsRetVal (*)(void *pUsr)) queueRateLimiter)); - CHKiRet(wtpSetpfChkStopWrkr (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, int)) queueChkStopWrkrReg)); - CHKiRet(wtpSetpfIsIdle (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, int)) queueIsIdleReg)); - CHKiRet(wtpSetpfDoWork (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, void *pWti, int)) queueConsumerReg)); - CHKiRet(wtpSetpfOnWorkerCancel (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, void*pWti))queueConsumerCancelCleanup)); - CHKiRet(wtpSetpfOnWorkerStartup (pThis->pWtpReg, (rsRetVal (*)(void *pUsr)) queueRegOnWrkrStartup)); - CHKiRet(wtpSetpfOnWorkerShutdown(pThis->pWtpReg, (rsRetVal (*)(void *pUsr)) queueRegOnWrkrShutdown)); + CHKiRet(wtpSetpfRateLimiter (pThis->pWtpReg, (rsRetVal (*)(void *pUsr)) qqueueRateLimiter)); + CHKiRet(wtpSetpfChkStopWrkr (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, int)) qqueueChkStopWrkrReg)); + CHKiRet(wtpSetpfIsIdle (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, int)) qqueueIsIdleReg)); + CHKiRet(wtpSetpfDoWork (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, void *pWti, int)) qqueueConsumerReg)); + CHKiRet(wtpSetpfOnWorkerCancel (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, void*pWti))qqueueConsumerCancelCleanup)); + CHKiRet(wtpSetpfOnWorkerStartup (pThis->pWtpReg, (rsRetVal (*)(void *pUsr)) qqueueRegOnWrkrStartup)); + CHKiRet(wtpSetpfOnWorkerShutdown(pThis->pWtpReg, (rsRetVal (*)(void *pUsr)) qqueueRegOnWrkrShutdown)); CHKiRet(wtpSetpmutUsr (pThis->pWtpReg, pThis->mut)); CHKiRet(wtpSetpcondBusy (pThis->pWtpReg, &pThis->notEmpty)); CHKiRet(wtpSetiNumWorkerThreads (pThis->pWtpReg, pThis->iNumWorkerThreads)); @@ -1843,10 +1848,10 @@ rsRetVal queueStart(queue_t *pThis) /* this is the ConstructionFinalizer */ /* If we are disk-assisted, we need to check if there is a QIF file * which we need to load. -- rgerhards, 2008-01-15 */ - iRetLocal = queueHaveQIF(pThis); + iRetLocal = qqueueHaveQIF(pThis); if(iRetLocal == RS_RET_OK) { dbgoprint((obj_t*) pThis, "on-disk queue present, needs to be reloaded\n"); - queueInitDA(pThis, QUEUE_MODE_ENQDEQ, LOCK_MUTEX); /* initiate DA mode */ + qqueueInitDA(pThis, QUEUE_MODE_ENQDEQ, LOCK_MUTEX); /* initiate DA mode */ bInitialized = 1; /* we are done */ } else { /* TODO: use logerror? -- rgerhards, 2008-01-16 */ @@ -1863,7 +1868,7 @@ rsRetVal queueStart(queue_t *pThis) /* this is the ConstructionFinalizer */ /* if the queue already contains data, we need to start the correct number of worker threads. This can be * the case when a disk queue has been loaded. If we did not start it here, it would never start. */ - queueAdviseMaxWorkers(pThis); + qqueueAdviseMaxWorkers(pThis); pThis->bQueueStarted = 1; finalize_it: @@ -1878,7 +1883,7 @@ finalize_it: * and 0 otherwise. * rgerhards, 2008-01-10 */ -static rsRetVal queuePersist(queue_t *pThis, int bIsCheckpoint) +static rsRetVal qqueuePersist(qqueue_t *pThis, int bIsCheckpoint) { DEFiRet; strm_t *psQIF = NULL; /* Queue Info File */ @@ -1889,7 +1894,7 @@ static rsRetVal queuePersist(queue_t *pThis, int bIsCheckpoint) ASSERT(pThis != NULL); if(pThis->qType != QUEUETYPE_DISK) { - if(queueGetOverallQueueSize(pThis) > 0) { + if(qqueueGetOverallQueueSize(pThis) > 0) { /* This error code is OK, but we will probably not implement this any time * The reason is that persistence happens via DA queues. But I would like to * leave the code as is, as we so have a hook in case we need one. @@ -1900,13 +1905,13 @@ static rsRetVal queuePersist(queue_t *pThis, int bIsCheckpoint) FINALIZE; /* if the queue is empty, we are happy and done... */ } - dbgoprint((obj_t*) pThis, "persisting queue to disk, %d entries...\n", queueGetOverallQueueSize(pThis)); + dbgoprint((obj_t*) pThis, "persisting queue to disk, %d entries...\n", qqueueGetOverallQueueSize(pThis)); /* Construct file name */ lenQIFNam = snprintf((char*)pszQIFNam, sizeof(pszQIFNam) / sizeof(uchar), "%s/%s.qi", (char*) glbl.GetWorkDir(), (char*)pThis->pszFilePrefix); - if((bIsCheckpoint != QUEUE_CHECKPOINT) && (queueGetOverallQueueSize(pThis) == 0)) { + if((bIsCheckpoint != QUEUE_CHECKPOINT) && (qqueueGetOverallQueueSize(pThis) == 0)) { if(pThis->bNeedDelQIF) { unlink((char*)pszQIFNam); pThis->bNeedDelQIF = 0; @@ -1940,7 +1945,7 @@ static rsRetVal queuePersist(queue_t *pThis, int bIsCheckpoint) * to the regular files. -- rgerhards, 2008-01-29 */ while(pThis->iUngottenObjs > 0) { - CHKiRet(queueGetUngottenObj(pThis, &pUsr)); + CHKiRet(qqueueGetUngottenObj(pThis, &pUsr)); CHKiRet((objSerialize(pUsr))(pUsr, psQIF)); objDestruct(pUsr); } @@ -1974,14 +1979,14 @@ finalize_it: * abide to our regular call interface)... * rgerhards, 2008-01-13 */ -rsRetVal queueChkPersist(queue_t *pThis) +rsRetVal qqueueChkPersist(qqueue_t *pThis) { DEFiRet; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); if(pThis->iPersistUpdCnt && ++pThis->iUpdsSincePersist >= pThis->iPersistUpdCnt) { - queuePersist(pThis, QUEUE_CHECKPOINT); + qqueuePersist(pThis, QUEUE_CHECKPOINT); pThis->iUpdsSincePersist = 0; } @@ -1990,8 +1995,8 @@ rsRetVal queueChkPersist(queue_t *pThis) /* destructor for the queue object */ -BEGINobjDestruct(queue) /* be sure to specify the object type also in END and CODESTART macros! */ -CODESTARTobjDestruct(queue) +BEGINobjDestruct(qqueue) /* be sure to specify the object type also in END and CODESTART macros! */ +CODESTARTobjDestruct(qqueue) pThis->bQueueInDestruction = 1; /* indicate we are in destruction (modifies some behaviour) */ /* shut down all workers (handles *all* of the persistence logic) @@ -2001,7 +2006,7 @@ CODESTARTobjDestruct(queue) * with a child! -- rgerhards, 2008-01-28 */ if(pThis->qType != QUEUETYPE_DIRECT && !pThis->bEnqOnly && pThis->pqParent == NULL) - queueShutdownWorkers(pThis); + qqueueShutdownWorkers(pThis); /* finally destruct our (regular) worker thread pool * Note: currently pWtpReg is never NULL, but if we optimize our logic, this may happen, @@ -2026,7 +2031,7 @@ CODESTARTobjDestruct(queue) wtpDestruct(&pThis->pWtpDA); } if(pThis->pqDA != NULL) { - queueDestruct(&pThis->pqDA); + qqueueDestruct(&pThis->pqDA); } /* persist the queue (we always do that - queuePersits() does cleanup if the queue is empty) @@ -2036,7 +2041,7 @@ CODESTARTobjDestruct(queue) * disk queues and DA mode. Anyhow, it doesn't hurt to know that we could extend it here * if need arises (what I doubt...) -- rgerhards, 2008-01-25 */ - CHKiRet_Hdlr(queuePersist(pThis, QUEUE_NO_CHECKPOINT)) { + CHKiRet_Hdlr(qqueuePersist(pThis, QUEUE_NO_CHECKPOINT)) { dbgoprint((obj_t*) pThis, "error %d persisting queue - data lost!\n", iRet); } @@ -2061,7 +2066,7 @@ CODESTARTobjDestruct(queue) if(pThis->pszSpoolDir != NULL) free(pThis->pszSpoolDir); -ENDobjDestruct(queue) +ENDobjDestruct(qqueue) /* set the queue's file prefix @@ -2070,7 +2075,7 @@ ENDobjDestruct(queue) * rgerhards, 2008-01-09 */ rsRetVal -queueSetFilePrefix(queue_t *pThis, uchar *pszPrefix, size_t iLenPrefix) +qqueueSetFilePrefix(qqueue_t *pThis, uchar *pszPrefix, size_t iLenPrefix) { DEFiRet; @@ -2093,11 +2098,11 @@ finalize_it: * rgerhards, 2008-01-09 */ rsRetVal -queueSetMaxFileSize(queue_t *pThis, size_t iMaxFileSize) +qqueueSetMaxFileSize(qqueue_t *pThis, size_t iMaxFileSize) { DEFiRet; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); if(iMaxFileSize < 1024) { ABORT_FINALIZE(RS_RET_VALUE_TOO_LOW); @@ -2114,13 +2119,22 @@ finalize_it: * Enqueues the new element and awakes worker thread. */ rsRetVal -queueEnqObj(queue_t *pThis, flowControl_t flowCtlType, void *pUsr) +qqueueEnqObj(qqueue_t *pThis, flowControl_t flowCtlType, void *pUsr) { DEFiRet; int iCancelStateSave; struct timespec t; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); + + /* first check if we need to discard this message (which will cause CHKiRet() to exit) + * rgerhards, 2008-10-07: It is OK to do this outside of mutex protection. The iQueueSize + * and bRunsDA parameters may not reflect the correct settings here, but they are + * "good enough" in the sense that they can be used to drive the decision. Valgrind's + * threading tools may point this access to be an error, but this is done + * intentional. I do not see this causes problems to us. + */ + CHKiRet(qqueueChkDiscardMsg(pThis, pThis->iQueueSize, pThis->bRunsDA, pUsr)); /* Please note that this function is not cancel-safe and consequently * sets the calling thread's cancelibility state to PTHREAD_CANCEL_DISABLE @@ -2133,12 +2147,9 @@ queueEnqObj(queue_t *pThis, flowControl_t flowCtlType, void *pUsr) d_pthread_mutex_lock(pThis->mut); } - /* first check if we need to discard this message (which will cause CHKiRet() to exit) */ - CHKiRet(queueChkDiscardMsg(pThis, pThis->iQueueSize, pThis->bRunsDA, pUsr)); - /* then check if we need to add an assistance disk queue */ if(pThis->bIsDA) - CHKiRet(queueChkStrtDA(pThis)); + CHKiRet(qqueueChkStrtDA(pThis)); /* handle flow control * There are two different flow control mechanisms: basic and advanced flow control. @@ -2191,17 +2202,24 @@ queueEnqObj(queue_t *pThis, flowControl_t flowCtlType, void *pUsr) } /* and finally enqueue the message */ - CHKiRet(queueAdd(pThis, pUsr)); - queueChkPersist(pThis); + CHKiRet(qqueueAdd(pThis, pUsr)); + qqueueChkPersist(pThis); finalize_it: if(pThis->qType != QUEUETYPE_DIRECT) { /* make sure at least one worker is running. */ - queueAdviseMaxWorkers(pThis); - dbgoprint((obj_t*) pThis, "EnqueueMsg advised worker start\n"); + qqueueAdviseMaxWorkers(pThis); /* and release the mutex */ d_pthread_mutex_unlock(pThis->mut); pthread_setcancelstate(iCancelStateSave, NULL); + dbgoprint((obj_t*) pThis, "EnqueueMsg advised worker start\n"); + /* the following pthread_yield is experimental, but brought us performance + * benefit. For details, please see http://kb.monitorware.com/post14216.html#p14216 + * rgerhards, 2008-10-09 + * but this is only true for uniprocessors, so we guard it with an optimize flag -- rgerhards, 2008-10-22 + */ + if(pThis->bOptimizeUniProc) + pthread_yield(); } RETiRet; @@ -2217,12 +2235,12 @@ finalize_it: * rgerhards, 2008-01-16 */ static rsRetVal -queueSetEnqOnly(queue_t *pThis, int bEnqOnly, int bLockMutex) +qqueueSetEnqOnly(qqueue_t *pThis, int bEnqOnly, int bLockMutex) { DEFiRet; DEFVARS_mutexProtection; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); /* for simplicity, we do one big mutex lock. This method is extremely seldom * called, so that doesn't matter... -- rgerhards, 2008-01-16 @@ -2261,24 +2279,24 @@ finalize_it: /* some simple object access methods */ -DEFpropSetMeth(queue, iPersistUpdCnt, int) -DEFpropSetMeth(queue, iDeqtWinFromHr, int) -DEFpropSetMeth(queue, iDeqtWinToHr, int) -DEFpropSetMeth(queue, toQShutdown, long) -DEFpropSetMeth(queue, toActShutdown, long) -DEFpropSetMeth(queue, toWrkShutdown, long) -DEFpropSetMeth(queue, toEnq, long) -DEFpropSetMeth(queue, iHighWtrMrk, int) -DEFpropSetMeth(queue, iLowWtrMrk, int) -DEFpropSetMeth(queue, iDiscardMrk, int) -DEFpropSetMeth(queue, iFullDlyMrk, int) -DEFpropSetMeth(queue, iDiscardSeverity, int) -DEFpropSetMeth(queue, bIsDA, int) -DEFpropSetMeth(queue, iMinMsgsPerWrkr, int) -DEFpropSetMeth(queue, bSaveOnShutdown, int) -DEFpropSetMeth(queue, pUsr, void*) -DEFpropSetMeth(queue, iDeqSlowdown, int) -DEFpropSetMeth(queue, sizeOnDiskMax, int64) +DEFpropSetMeth(qqueue, iPersistUpdCnt, int) +DEFpropSetMeth(qqueue, iDeqtWinFromHr, int) +DEFpropSetMeth(qqueue, iDeqtWinToHr, int) +DEFpropSetMeth(qqueue, toQShutdown, long) +DEFpropSetMeth(qqueue, toActShutdown, long) +DEFpropSetMeth(qqueue, toWrkShutdown, long) +DEFpropSetMeth(qqueue, toEnq, long) +DEFpropSetMeth(qqueue, iHighWtrMrk, int) +DEFpropSetMeth(qqueue, iLowWtrMrk, int) +DEFpropSetMeth(qqueue, iDiscardMrk, int) +DEFpropSetMeth(qqueue, iFullDlyMrk, int) +DEFpropSetMeth(qqueue, iDiscardSeverity, int) +DEFpropSetMeth(qqueue, bIsDA, int) +DEFpropSetMeth(qqueue, iMinMsgsPerWrkr, int) +DEFpropSetMeth(qqueue, bSaveOnShutdown, int) +DEFpropSetMeth(qqueue, pUsr, void*) +DEFpropSetMeth(qqueue, iDeqSlowdown, int) +DEFpropSetMeth(qqueue, sizeOnDiskMax, int64) /* This function can be used as a generic way to set properties. Only the subset @@ -2287,11 +2305,11 @@ DEFpropSetMeth(queue, sizeOnDiskMax, int64) * rgerhards, 2008-01-11 */ #define isProp(name) !rsCStrSzStrCmp(pProp->pcsName, (uchar*) name, sizeof(name) - 1) -static rsRetVal queueSetProperty(queue_t *pThis, var_t *pProp) +static rsRetVal qqueueSetProperty(qqueue_t *pThis, var_t *pProp) { DEFiRet; - ISOBJ_TYPE_assert(pThis, queue); + ISOBJ_TYPE_assert(pThis, qqueue); ASSERT(pProp != NULL); if(isProp("iQueueSize")) { @@ -2313,19 +2331,19 @@ finalize_it: #undef isProp /* dummy */ -rsRetVal queueQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; } +rsRetVal qqueueQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; } /* Initialize the stream class. Must be called as the very first method * before anything else is called inside this class. * rgerhards, 2008-01-09 */ -BEGINObjClassInit(queue, 1, OBJ_IS_CORE_MODULE) +BEGINObjClassInit(qqueue, 1, OBJ_IS_CORE_MODULE) /* request objects we use */ CHKiRet(objUse(glbl, CORE_COMPONENT)); /* now set our own handlers */ - OBJSetMethodHandler(objMethod_SETPROPERTY, queueSetProperty); -ENDObjClassInit(queue) + OBJSetMethodHandler(objMethod_SETPROPERTY, qqueueSetProperty); +ENDObjClassInit(qqueue) /* vi:set ai: */ diff --git a/runtime/queue.h b/runtime/queue.h index 9e75b31b..a267862d 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -58,6 +58,7 @@ typedef struct qWrkThrd_s { typedef struct queue_s { BEGINobjInstance; queueType_t qType; + int bOptimizeUniProc; /* cache for the equally-named global setting, pulled at time of queue creation */ int bEnqOnly; /* does queue run in enqueue-only mode (1) or not (0)? */ int bSaveOnShutdown;/* persists everthing on shutdown (if DA!)? 1-yes, 0-no */ int bQueueStarted; /* has queueStart() been called on this queue? 1-yes, 0-no */ @@ -159,7 +160,7 @@ typedef struct queue_s { strm_t *pRead; /* current file to be read */ } disk; } tVars; -} queue_t; +} qqueue_t; /* some symbolic constants for easier reference */ #define QUEUE_MODE_ENQDEQ 0 @@ -176,30 +177,30 @@ typedef struct queue_s { #define QUEUE_TIMEOUT_ETERNAL 24 * 60 * 60 * 1000 /* prototypes */ -rsRetVal queueDestruct(queue_t **ppThis); -rsRetVal queueEnqObj(queue_t *pThis, flowControl_t flwCtlType, void *pUsr); -rsRetVal queueStart(queue_t *pThis); -rsRetVal queueSetMaxFileSize(queue_t *pThis, size_t iMaxFileSize); -rsRetVal queueSetFilePrefix(queue_t *pThis, uchar *pszPrefix, size_t iLenPrefix); -rsRetVal queueConstruct(queue_t **ppThis, queueType_t qType, int iWorkerThreads, +rsRetVal qqueueDestruct(qqueue_t **ppThis); +rsRetVal qqueueEnqObj(qqueue_t *pThis, flowControl_t flwCtlType, void *pUsr); +rsRetVal qqueueStart(qqueue_t *pThis); +rsRetVal qqueueSetMaxFileSize(qqueue_t *pThis, size_t iMaxFileSize); +rsRetVal qqueueSetFilePrefix(qqueue_t *pThis, uchar *pszPrefix, size_t iLenPrefix); +rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThreads, int iMaxQueueSize, rsRetVal (*pConsumer)(void*,void*)); -PROTOTYPEObjClassInit(queue); -PROTOTYPEpropSetMeth(queue, iPersistUpdCnt, int); -PROTOTYPEpropSetMeth(queue, iDeqtWinFromHr, int); -PROTOTYPEpropSetMeth(queue, iDeqtWinToHr, int); -PROTOTYPEpropSetMeth(queue, toQShutdown, long); -PROTOTYPEpropSetMeth(queue, toActShutdown, long); -PROTOTYPEpropSetMeth(queue, toWrkShutdown, long); -PROTOTYPEpropSetMeth(queue, toEnq, long); -PROTOTYPEpropSetMeth(queue, iHighWtrMrk, int); -PROTOTYPEpropSetMeth(queue, iLowWtrMrk, int); -PROTOTYPEpropSetMeth(queue, iDiscardMrk, int); -PROTOTYPEpropSetMeth(queue, iDiscardSeverity, int); -PROTOTYPEpropSetMeth(queue, iMinMsgsPerWrkr, int); -PROTOTYPEpropSetMeth(queue, bSaveOnShutdown, int); -PROTOTYPEpropSetMeth(queue, pUsr, void*); -PROTOTYPEpropSetMeth(queue, iDeqSlowdown, int); -PROTOTYPEpropSetMeth(queue, sizeOnDiskMax, int64); -#define queueGetID(pThis) ((unsigned long) pThis) +PROTOTYPEObjClassInit(qqueue); +PROTOTYPEpropSetMeth(qqueue, iPersistUpdCnt, int); +PROTOTYPEpropSetMeth(qqueue, iDeqtWinFromHr, int); +PROTOTYPEpropSetMeth(qqueue, iDeqtWinToHr, int); +PROTOTYPEpropSetMeth(qqueue, toQShutdown, long); +PROTOTYPEpropSetMeth(qqueue, toActShutdown, long); +PROTOTYPEpropSetMeth(qqueue, toWrkShutdown, long); +PROTOTYPEpropSetMeth(qqueue, toEnq, long); +PROTOTYPEpropSetMeth(qqueue, iHighWtrMrk, int); +PROTOTYPEpropSetMeth(qqueue, iLowWtrMrk, int); +PROTOTYPEpropSetMeth(qqueue, iDiscardMrk, int); +PROTOTYPEpropSetMeth(qqueue, iDiscardSeverity, int); +PROTOTYPEpropSetMeth(qqueue, iMinMsgsPerWrkr, int); +PROTOTYPEpropSetMeth(qqueue, bSaveOnShutdown, int); +PROTOTYPEpropSetMeth(qqueue, pUsr, void*); +PROTOTYPEpropSetMeth(qqueue, iDeqSlowdown, int); +PROTOTYPEpropSetMeth(qqueue, sizeOnDiskMax, int64); +#define qqueueGetID(pThis) ((unsigned long) pThis) #endif /* #ifndef QUEUE_H_INCLUDED */ diff --git a/runtime/rsyslog.c b/runtime/rsyslog.c index 54db12c2..8df100a1 100644 --- a/runtime/rsyslog.c +++ b/runtime/rsyslog.c @@ -157,7 +157,7 @@ rsrtInit(char **ppErrObj, obj_if_t *pObjIF) if(ppErrObj != NULL) *ppErrObj = "wtp"; CHKiRet(wtpClassInit(NULL)); if(ppErrObj != NULL) *ppErrObj = "queue"; - CHKiRet(queueClassInit(NULL)); + CHKiRet(qqueueClassInit(NULL)); if(ppErrObj != NULL) *ppErrObj = "vmstk"; CHKiRet(vmstkClassInit(NULL)); if(ppErrObj != NULL) *ppErrObj = "sysvar"; diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h index 9ce58210..32177a9f 100644 --- a/runtime/rsyslog.h +++ b/runtime/rsyslog.h @@ -84,6 +84,14 @@ typedef rsRetVal (*errLogFunc_t)(uchar*); /* this is a trick to store a function typedef struct permittedPeers_s permittedPeers_t; /* this should go away in the long term -- rgerhards, 2008-05-19 */ typedef struct permittedPeerWildcard_s permittedPeerWildcard_t; /* this should go away in the long term -- rgerhards, 2008-05-19 */ typedef struct tcpsrv_s tcpsrv_t; +typedef struct tcps_sess_s tcps_sess_t; +typedef struct strmsrv_s strmsrv_t; +typedef struct strms_sess_s strms_sess_t; +typedef struct vmstk_s vmstk_t; +typedef rsRetVal (*prsf_t)(struct vmstk_s*, int); /* pointer to a RainerScript function */ + +typedef struct tcpLstnPortList_s tcpLstnPortList_t; // TODO: rename? +typedef struct strmLstnPortList_s strmLstnPortList_t; // TODO: rename? /* some universal 64 bit define... */ typedef long long int64; @@ -95,6 +103,8 @@ typedef unsigned int u_int32_t; /* TODO: is this correct? */ typedef int socklen_t; #endif +typedef char bool; /* I intentionally use char, to keep it slim so that many fit into the CPU cache! */ + /* settings for flow control * TODO: is there a better place for them? -- rgerhards, 2008-03-14 */ @@ -252,8 +262,24 @@ enum rsRetVal_ /** return value. All methods return this if not specified oth RS_RET_QUEUE_FULL = -2105, /**< queue is full, operation could not be completed */ RS_RET_ACCEPT_ERR = -2106, /**< error during accept() system call */ RS_RET_INVLD_TIME = -2107, /**< invalid timestamp (e.g. could not be parsed) */ + RS_RET_NO_ZIP = -2108, /**< ZIP functionality is not present */ RS_RET_CODE_ERR = -2109, /**< program code (internal) error */ - RS_RET_NONFATAL_CONFIG_ERR = -2123, /**< non-fatal error during config processing */ + RS_RET_FUNC_NO_LPAREN = -2110, /**< left parenthesis missing after function call (rainerscript) */ + RS_RET_FUNC_MISSING_EXPR = -2111, /**< no expression after comma in function call (rainerscript) */ + RS_RET_INVLD_NBR_ARGUMENTS = -2112, /**< invalid number of arguments for function call (rainerscript) */ + RS_RET_INVLD_FUNC = -2113, /**< invalid function name for function call (rainerscript) */ + RS_RET_DUP_FUNC_NAME = -2114, /**< duplicate function name (rainerscript) */ + RS_RET_UNKNW_FUNC = -2115, /**< unkown function name (rainerscript) */ + RS_RET_ERR_RLIM_NOFILE = -2116, /**< error setting max. nbr open files process limit */ + RS_RET_ERR_CREAT_PIPE = -2117, /**< error during pipe creation */ + RS_RET_ERR_FORK = -2118, /**< error during fork() */ + RS_RET_ERR_WRITE_PIPE = -2119, /**< error writing to pipe */ + RS_RET_RSCORE_TOO_OLD = -2120, /**< rsyslog core is too old for ... (eg this plugin) */ + RS_RET_DEFER_COMMIT = -2121, /**< output plugin status: not yet committed (an OK state!) */ + RS_RET_PREVIOUS_COMMITTED = -2122, /**< output plugin status: previous record was committed (an OK state!) */ + RS_RET_ACTION_FAILED = -2123, /**< action failed and is now suspended (consider this permanent for the time being) */ + RS_RET_NONFATAL_CONFIG_ERR = -2124, /**< non-fatal error during config processing */ + RS_RET_FILENAME_INVALID = -2140, /**< filename invalid, not found, no access, ... */ /* RainerScript error messages (range 1000.. 1999) */ RS_RET_SYSVAR_NOT_FOUND = 1001, /**< system variable could not be found (maybe misspelled) */ @@ -337,9 +363,15 @@ typedef enum rsObjectID rsObjID; # define __attribute__(x) /*NOTHING*/ #endif +#ifndef O_CLOEXEC +/* of course, this limits the functionality... */ +# define O_CLOEXEC 0 +#endif + /* The following prototype is convenient, even though it may not be the 100% correct place.. -- rgerhards 2008-01-07 */ void dbgprintf(char *, ...) __attribute__((format(printf, 1, 2))); + #include "debug.h" #include "obj.h" diff --git a/runtime/srutils.c b/runtime/srutils.c index 97cc3252..d01ca20d 100644 --- a/runtime/srutils.c +++ b/runtime/srutils.c @@ -371,6 +371,7 @@ int getNumberDigits(long lNum) rsRetVal timeoutComp(struct timespec *pt, long iTimeout) { + BEGINfunc assert(pt != NULL); /* compute timeout */ clock_gettime(CLOCK_REALTIME, pt); @@ -379,6 +380,7 @@ timeoutComp(struct timespec *pt, long iTimeout) pt->tv_nsec -= 1000000000; } pt->tv_sec += iTimeout / 1000; + ENDfunc return RS_RET_OK; /* so far, this is static... */ } @@ -393,6 +395,7 @@ timeoutVal(struct timespec *pt) { struct timespec t; long iTimeout; + BEGINfunc assert(pt != NULL); /* compute timeout */ @@ -403,6 +406,7 @@ timeoutVal(struct timespec *pt) if(iTimeout < 0) iTimeout = 0; + ENDfunc return iTimeout; } @@ -454,7 +458,7 @@ srSleep(int iSeconds, int iuSeconds) * Added 2008-01-30 */ char *rs_strerror_r(int errnum, char *buf, size_t buflen) { -#ifdef __hpux +#ifndef HAVE_STRERROR_R char *pszErr; pszErr = strerror(errnum); snprintf(buf, buflen, "%s", pszErr); diff --git a/runtime/stream.c b/runtime/stream.c index f1f69cc8..1cff2da6 100644 --- a/runtime/stream.c +++ b/runtime/stream.c @@ -368,7 +368,7 @@ strmReadLine(strm_t *pThis, cstr_t **ppCStr) CHKiRet(rsCStrAppendChar(*ppCStr, c)); CHKiRet(strmReadChar(pThis, &c)); } - CHKiRet(rsCStrFinish(*ppCStr)); + CHKiRet(cstrFinalize(*ppCStr)); finalize_it: if(iRet != RS_RET_OK && *ppCStr != NULL) diff --git a/runtime/stringbuf.c b/runtime/stringbuf.c index 63b42348..a2d9c599 100644 --- a/runtime/stringbuf.c +++ b/runtime/stringbuf.c @@ -6,8 +6,9 @@ * Please see syslogd.c for license information. * All functions in this "class" start with rsCStr (rsyslog Counted String). * begun 2005-09-07 rgerhards + * did some optimization (read: bugs!) rgerhards, 2009-06-16 * - * Copyright (C) 2007-2008 by Rainer Gerhards and Adiscon GmbH + * Copyright (C) 2007-2009 by Rainer Gerhards and Adiscon GmbH * * This file is part of the rsyslog runtime library. * @@ -40,6 +41,7 @@ #include "regexp.h" #include "obj.h" +uchar* rsCStrGetSzStr(cstr_t *pThis); /* ################################################################# * * private members * @@ -54,15 +56,14 @@ DEFobjCurrIf(regexp) * ################################################################# */ -rsRetVal rsCStrConstruct(cstr_t **ppThis) +rsRetVal cstrConstruct(cstr_t **ppThis) { DEFiRet; cstr_t *pThis; ASSERT(ppThis != NULL); - if((pThis = (cstr_t*) calloc(1, sizeof(cstr_t))) == NULL) - ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); + CHKmalloc(pThis = (cstr_t*) calloc(1, sizeof(cstr_t))); rsSETOBJTYPE(pThis, OIDrsCStr); pThis->pBuf = NULL; @@ -89,7 +90,7 @@ rsRetVal rsCStrConstructFromszStr(cstr_t **ppThis, uchar *sz) CHKiRet(rsCStrConstruct(&pThis)); - pThis->iBufSize = pThis->iStrLen = strlen((char*)(char *) sz); + pThis->iBufSize = pThis->iStrLen = strlen((char *) sz); if((pThis->pBuf = (uchar*) malloc(sizeof(uchar) * pThis->iStrLen)) == NULL) { RSFREEOBJ(pThis); ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); @@ -137,24 +138,8 @@ void rsCStrDestruct(cstr_t **ppThis) { cstr_t *pThis = *ppThis; - /* rgerhards 2005-10-19: The free of pBuf was contained in conditional compilation. - * The code was only compiled if STRINGBUF_TRIM_ALLOCSIZE was set to 1. I honestly - * do not know why it was so, I think it was an artifact. Anyhow, I have changed this - * now. Should there any issue occur, this comment hopefully will shed some light - * on what happened. I re-verified, and this function has never before been called - * by anyone. So changing it can have no impact for obvious reasons... - * - * rgerhards, 2008-02-20: I changed the interface to the new calling conventions, where - * the destructor receives a pointer to the object, so that it can set it to NULL. - */ - if(pThis->pBuf != NULL) { - free(pThis->pBuf); - } - - if(pThis->pszBuf != NULL) { - free(pThis->pszBuf); - } - + free(pThis->pBuf); + free(pThis->pszBuf); RSFREEOBJ(pThis); *ppThis = NULL; } @@ -166,12 +151,14 @@ void rsCStrDestruct(cstr_t **ppThis) * allocated. In practice, a bit more is allocated because we envision that * some more characters may be added after these. * rgerhards, 2008-01-07 + * changed to utilized realloc() -- rgerhards, 2009-06-16 */ -static rsRetVal rsCStrExtendBuf(cstr_t *pThis, size_t iMinNeeded) +static rsRetVal +rsCStrExtendBuf(cstr_t *pThis, size_t iMinNeeded) { - DEFiRet; uchar *pNewBuf; size_t iNewSize; + DEFiRet; /* first compute the new size needed */ if(iMinNeeded > pThis->iAllocIncrement) { @@ -187,15 +174,9 @@ static rsRetVal rsCStrExtendBuf(cstr_t *pThis, size_t iMinNeeded) } iNewSize += pThis->iBufSize; /* add current size */ - /* and then allocate and copy over */ /* DEV debugging only: dbgprintf("extending string buffer, old %d, new %d\n", pThis->iBufSize, iNewSize); */ - if((pNewBuf = (uchar*) malloc(iNewSize * sizeof(uchar))) == NULL) - ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); - memcpy(pNewBuf, pThis->pBuf, pThis->iBufSize); + CHKmalloc(pNewBuf = (uchar*) realloc(pThis->pBuf, iNewSize * sizeof(uchar))); pThis->iBufSize = iNewSize; - if(pThis->pBuf != NULL) { - free(pThis->pBuf); - } pThis->pBuf = pNewBuf; finalize_it: @@ -288,6 +269,29 @@ finalize_it: } +/* NEW VARIANT + * Append a character to the current string object. This may only be done until + * cstrFinalize() is called. + * rgerhards, 2009-06-16 + */ +rsRetVal cstrAppendChar(cstr_t *pThis, uchar c) +{ + DEFiRet; + + rsCHECKVALIDOBJECT(pThis, OIDrsCStr); + + if(pThis->iStrLen >= pThis->iBufSize) { + CHKiRet(rsCStrExtendBuf(pThis, 1)); /* need more memory! */ + } + + /* ok, when we reach this, we have sufficient memory */ + *(pThis->pBuf + pThis->iStrLen++) = c; + +finalize_it: + RETiRet; +} + + /* Sets the string object to the classigal sz-string provided. * Any previously stored vlaue is discarded. If a NULL pointer * the the new value (pszNew) is provided, an empty string is @@ -299,10 +303,8 @@ rsRetVal rsCStrSetSzStr(cstr_t *pThis, uchar *pszNew) { rsCHECKVALIDOBJECT(pThis, OIDrsCStr); - if(pThis->pBuf != NULL) - free(pThis->pBuf); - if(pThis->pszBuf != NULL) - free(pThis->pszBuf); + free(pThis->pBuf); + free(pThis->pszBuf); if(pszNew == NULL) { pThis->iStrLen = 0; pThis->iBufSize = 0; @@ -390,35 +392,42 @@ uchar* rsCStrGetSzStr(cstr_t *pThis) } +/* NEW VERSION for interface without separate psz buffer! */ +/* Returns the cstr data as a classical C sz string. We use that the + * Finalizer did properly terminate our string (but we may stil be NULL). + * So it is vital that the finalizer is called BEFORe this function here! + * The caller must not free or otherwise manipulate the returned string and must not + * destroy the CStr object as long as the ascii string is used. + * This function may return NULL, if the string is currently NULL. This + * is a feature, not a bug. If you need non-NULL in any case, use + * cstrGetSzStrNoNULL() instead. + * Note that due to the new single-buffer interface this function almost does nothing! + * rgerhards, 2006-09-16 + */ +uchar* cstrGetSzStr(cstr_t *pThis) +{ + rsCHECKVALIDOBJECT(pThis, OIDrsCStr); + return(pThis->pBuf); +} + + /* Converts the CStr object to a classical zero-terminated C string, * returns that string and destroys the CStr object. The returned string * MUST be freed by the caller. The function might return NULL if * no memory can be allocated. * - * TODO: - * This function should at some time become special. The base idea is to - * add one extra byte to the end of the regular buffer, so that we can - * convert it to an szString without the need to copy. The extra memory - * footprint is not hefty, but the performance gain is potentially large. - * To get it done now, I am not doing the optimiziation right now. - * rgerhards, 2005-09-07 + * This is the NEW replacement for rsCStrConvSzStrAndDestruct which does + * no longer utilize a special buffer but soley works on pBuf (and also + * assumes that cstrFinalize had been called). * - * rgerhards, 2007-09-04: I have changed the interface of this function. It now - * returns an rsRetVal, so that we can communicate back if we have an error. - * Using the standard method is much better than returning NULL. Secondly, NULL - * was not actually an error - it was in indication if the string was empty. - * This was needed in some parts of the code, in others not. I have now added - * a second parameter to specify what the caller needs. I hope these changes - * will make it less likely that the function is called incorrectly, what - * previously happend quite often and was the cause of a number of program - * aborts. So the parameters are now: + * Parameters are as follows: * pointer to the object, pointer to string-pointer to receive string and * bRetNULL: 0 - must not return NULL on empty string, return "" in that * case, 1 - return NULL instead of an empty string. * PLEASE NOTE: the caller must free the memory returned in ppSz in any case * (except, of course, if it is NULL). */ -rsRetVal rsCStrConvSzStrAndDestruct(cstr_t *pThis, uchar **ppSz, int bRetNULL) +rsRetVal cstrConvSzStrAndDestruct(cstr_t *pThis, uchar **ppSz, int bRetNULL) { DEFiRet; uchar* pRetBuf; @@ -429,14 +438,13 @@ rsRetVal rsCStrConvSzStrAndDestruct(cstr_t *pThis, uchar **ppSz, int bRetNULL) if(pThis->pBuf == NULL) { if(bRetNULL == 0) { - if((pRetBuf = malloc(sizeof(uchar))) == NULL) - ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); + CHKmalloc(pRetBuf = malloc(sizeof(uchar))); *pRetBuf = '\0'; } else { pRetBuf = NULL; } } else - pRetBuf = rsCStrGetSzStr(pThis); + pRetBuf = pThis->pBuf; *ppSz = pRetBuf; @@ -445,60 +453,39 @@ finalize_it: * that we can NOT use the rsCStrDestruct function as it would * also free the sz String buffer, which we pass on to the user. */ - if(pThis->pBuf != NULL) - free(pThis->pBuf); RSFREEOBJ(pThis); - RETiRet; } -#if STRINGBUF_TRIM_ALLOCSIZE == 1 - /* Only in this mode, we need to trim the string. To do - * so, we must allocate a new buffer of the exact - * string size, and then copy the old one over. - */ - /* WARNING - * STRINGBUF_TRIM_ALLOCSIZE can, in theory, be used to trim - * memory buffers. This part of the code was inherited from - * liblogging (where it is used in a different context) but - * never put to use in rsyslog. The reason is that it is hardly - * imaginable where the extra performance cost is worth the save - * in memory alloc. Then Anders Blomdel rightfully pointed out that - * the code does not work at all - and nobody even know that it - * probably shouldn't. Rather than removing, I deciced to somewhat - * fix the code, so that this feature may be enabled if somebody - * really has a need for it. Be warned, however, that I NEVER - * tested the fix. So if you intend to use this feature, you must - * do full testing before you rely on it. -- rgerhards, 2008-02-12 - */ -rsRetVal rsCStrFinish(cstr_t __attribute__((unused)) *pThis) +/* Finalize the string object. This must be called after all data is added to it + * but before that data is used. + * rgerhards, 2009-06-16 + */ +rsRetVal +cstrFinalize(cstr_t *pThis) { DEFiRet; - uchar* pBuf; rsCHECKVALIDOBJECT(pThis, OIDrsCStr); + + assert(pThis->bIsFinalized == 0); - if((pBuf = malloc((pThis->iStrLen) * sizeof(uchar))) == NULL) - { /* OK, in this case we use the previous buffer. At least - * we have it ;) - */ - } - else - { /* got the new buffer, so let's use it */ - memcpy(pBuf, pThis->pBuf, pThis->iStrLen); - pThis->pBuf = pBuf; + if(pThis->iStrLen > 0) { + /* terminate string only if one exists */ + CHKiRet(cstrAppendChar(pThis, '\0')); + --pThis->iStrLen; /* do NOT count the \0 byte */ } + pThis->bIsFinalized = 1; +finalize_it: RETiRet; } -#endif /* #if STRINGBUF_TRIM_ALLOCSIZE == 1 */ void rsCStrSetAllocIncrement(cstr_t *pThis, int iNewIncrement) { rsCHECKVALIDOBJECT(pThis, OIDrsCStr); assert(iNewIncrement > 0); - pThis->iAllocIncrement = iNewIncrement; } @@ -694,6 +681,7 @@ int rsCStrCaseInsensitveStartsWithSzStr(cstr_t *pCS1, uchar *psz, size_t iLenSz) return -1; /* pCS1 is less then psz */ } + /* check if a CStr object matches a regex. * msamia@redhat.com 2007-07-12 * @return returns 0 if matched @@ -701,25 +689,54 @@ int rsCStrCaseInsensitveStartsWithSzStr(cstr_t *pCS1, uchar *psz, size_t iLenSz) * rgerhards, 2007-07-16: bug is no real bug, because rsyslogd ensures there * never is a \0 *inside* a property string. * Note that the function returns -1 if regexp functionality is not available. - * TODO: change calling interface! -- rgerhards, 2008-03-07 + * rgerhards: 2009-03-04: ERE support added, via parameter iType: 0 - BRE, 1 - ERE + * Arnaud Cornet/rgerhards: 2009-04-02: performance improvement by caching compiled regex + * If a caller does not need the cached version, it must still provide memory for it + * and must call rsCStrRegexDestruct() afterwards. */ -int rsCStrSzStrMatchRegex(cstr_t *pCS1, uchar *psz) +rsRetVal rsCStrSzStrMatchRegex(cstr_t *pCS1, uchar *psz, int iType, void *rc) { - regex_t preq; + regex_t **cache = (regex_t**) rc; int ret; + DEFiRet; - BEGINfunc + assert(pCS1 != NULL); + assert(psz != NULL); + assert(cache != NULL); if(objUse(regexp, LM_REGEXP_FILENAME) == RS_RET_OK) { - regexp.regcomp(&preq, (char*) rsCStrGetSzStr(pCS1), 0); - ret = regexp.regexec(&preq, (char*) psz, 0, NULL, 0); - regexp.regfree(&preq); + if (*cache == NULL) { + *cache = calloc(sizeof(regex_t), 1); + regexp.regcomp(*cache, (char*) rsCStrGetSzStr(pCS1), (iType == 1 ? REG_EXTENDED : 0) | REG_NOSUB); + } + ret = regexp.regexec(*cache, (char*) psz, 0, NULL, 0); + if(ret != 0) + ABORT_FINALIZE(RS_RET_NOT_FOUND); } else { - ret = 1; /* simulate "not found" */ + ABORT_FINALIZE(RS_RET_NOT_FOUND); } - ENDfunc - return ret; +finalize_it: + RETiRet; +} + + +/* free a cached compiled regex + * Caller must provide a pointer to a buffer that was created by + * rsCStrSzStrMatchRegexCache() + */ +void rsCStrRegexDestruct(void *rc) +{ + regex_t **cache = rc; + + assert(cache != NULL); + assert(*cache != NULL); + + if(objUse(regexp, LM_REGEXP_FILENAME) == RS_RET_OK) { + regexp.regfree(*cache); + free(*cache); + *cache = NULL; + } } @@ -997,56 +1014,6 @@ int rsCStrCaseInsensitiveLocateInSzStr(cstr_t *pThis, uchar *sz) } -#if 0 /* read comment below why this is commented out. In short: for future use! */ -/* locate the first occurence of a standard sz string inside a rsCStr object. - * Returns the offset (0-bound) of this first occurrence. If not found, -1 is - * returned. - * rgerhards 2005-09-19 - * WARNING: I accidently created this function (I later noticed I didn't relly - * need it... I will not remove the function, as it probably is useful - * some time later. However, it is not fully tested, so start with testing - * it before you put it to first use). - */ -int rsCStrLocateSzStr(cstr_t *pThis, uchar *sz) -{ - int iLenSz; - int i; - int iMax; - int bFound; - rsCHECKVALIDOBJECT(pThis, OIDrsCStr); - - if(sz == NULL) - return 0; - - iLenSz = strlen((char*)sz); - if(iLenSz == 0) - return 0; - - /* compute the largest index where a match could occur - after all, - * the to-be-located string must be able to be present in the - * searched string (it needs its size ;)). - */ - iMax = pThis->iStrLen - iLenSz; - - bFound = 0; - i = 0; - while(i < iMax && !bFound) { - int iCheck; - uchar *pComp = pThis->pBuf + i; - for(iCheck = 0 ; iCheck < iLenSz ; ++iCheck) - if(*(pComp + iCheck) != *(sz + iCheck)) - break; - if(iCheck == iLenSz) - bFound = 1; /* found! - else it wouldn't be equal */ - else - ++i; /* on to the next try */ - } - - return(bFound ? i : -1); -} -#endif /* end comment out */ - - /* our exit function. TODO: remove once converted to a class * rgerhards, 2008-03-11 */ @@ -1070,11 +1037,5 @@ finalize_it: } -/* - * Local variables: - * c-indent-level: 8 - * c-basic-offset: 8 - * tab-width: 8 - * End: - * vi:set ai: +/* vi:set ai: */ diff --git a/runtime/stringbuf.h b/runtime/stringbuf.h index c1966449..d28aee26 100644 --- a/runtime/stringbuf.h +++ b/runtime/stringbuf.h @@ -49,13 +49,15 @@ typedef struct cstr_s size_t iBufSize; /**< current maximum size of the string buffer */ size_t iStrLen; /**< length of the string in characters. */ size_t iAllocIncrement; /**< the amount of bytes the string should be expanded if it needs to */ + bool bIsFinalized; /**< is this object finished and ready for use? (a debug aid, may be removed later TODO 2009-06-16) */ } cstr_t; /** * Construct a rsCStr object. */ -rsRetVal rsCStrConstruct(cstr_t **ppThis); +rsRetVal cstrConstruct(cstr_t **ppThis); +#define rsCStrConstruct(x) cstrConstruct((x)) rsRetVal rsCStrConstructFromszStr(cstr_t **ppThis, uchar *sz); rsRetVal rsCStrConstructFromCStr(cstr_t **ppThis, cstr_t *pFrom); @@ -63,6 +65,7 @@ rsRetVal rsCStrConstructFromCStr(cstr_t **ppThis, cstr_t *pFrom); * Destruct the string buffer object. */ void rsCStrDestruct(cstr_t **ppThis); +#define cstrDestruct(x) rsCStrDestruct((x)) /** * Append a character to an existing string. If necessary, the @@ -71,6 +74,7 @@ void rsCStrDestruct(cstr_t **ppThis); * \param c Character to append to string. */ rsRetVal rsCStrAppendChar(cstr_t *pThis, uchar c); +rsRetVal cstrAppendChar(cstr_t *pThis, uchar c); /** * Truncate "n" number of characters from the end of the @@ -123,10 +127,9 @@ rsRetVal rsCStrAppendInt(cstr_t *pThis, long i); rsRetVal strExit(void); /* TODO: remove once we have a real object interface! */ -uchar* rsCStrGetSzStr(cstr_t *pThis); +uchar* __attribute__((deprecated)) rsCStrGetSzStr(cstr_t *pThis); uchar* rsCStrGetSzStrNoNULL(cstr_t *pThis); rsRetVal rsCStrSetSzStr(cstr_t *pThis, uchar *pszNew); -rsRetVal rsCStrConvSzStrAndDestruct(cstr_t *pThis, uchar **ppSz, int bRetNULL); int rsCStrCStrCmp(cstr_t *pCS1, cstr_t *pCS2); int rsCStrSzStrCmp(cstr_t *pCS1, uchar *psz, size_t iLenSz); int rsCStrOffsetSzStrCmp(cstr_t *pCS1, size_t iOffset, uchar *psz, size_t iLenSz); @@ -136,11 +139,17 @@ int rsCStrCaseInsensitiveLocateInSzStr(cstr_t *pThis, uchar *sz); int rsCStrStartsWithSzStr(cstr_t *pCS1, uchar *psz, size_t iLenSz); int rsCStrCaseInsensitveStartsWithSzStr(cstr_t *pCS1, uchar *psz, size_t iLenSz); int rsCStrSzStrStartsWithCStr(cstr_t *pCS1, uchar *psz, size_t iLenSz); -int rsCStrSzStrMatchRegex(cstr_t *pCS1, uchar *psz); +rsRetVal rsCStrSzStrMatchRegex(cstr_t *pCS1, uchar *psz, int iType, void *cache); +void rsCStrRegexDestruct(void *rc); rsRetVal rsCStrConvertToNumber(cstr_t *pStr, number_t *pNumber); rsRetVal rsCStrConvertToBool(cstr_t *pStr, number_t *pBool); rsRetVal rsCStrAppendCStr(cstr_t *pThis, cstr_t *pstrAppend); +/* new calling interface */ +rsRetVal cstrFinalize(cstr_t *pThis); +rsRetVal cstrConvSzStrAndDestruct(cstr_t *pThis, uchar **ppSz, int bRetNULL); +uchar* cstrGetSzStr(cstr_t *pThis); + /* now come inline-like functions */ #ifdef NDEBUG # define rsCStrLen(x) ((int)((x)->iStrLen)) @@ -148,19 +157,6 @@ rsRetVal rsCStrAppendCStr(cstr_t *pThis, cstr_t *pstrAppend); int rsCStrLen(cstr_t *pThis); #endif -#if STRINGBUF_TRIM_ALLOCSIZE != 1 -/* This is the normal case (see comment in rsCStrFinish!). In those cases, the function - * simply needs to do nothing, so that we can save us the function call. - * rgerhards, 2008-02-12 - */ -# define rsCStrFinish(pThis) RS_RET_OK -#else - /** - * Finish the string buffer dynamic allocation. - */ - rsRetVal rsCStrFinish(cstr_t *pThis); -#endif - #define rsCStrGetBufBeg(x) ((x)->pBuf) rsRetVal strInit(); diff --git a/runtime/strms_sess.c b/runtime/strms_sess.c new file mode 100644 index 00000000..0aeebe03 --- /dev/null +++ b/runtime/strms_sess.c @@ -0,0 +1,300 @@ +/* strms_sess.c + * + * This implements a session of the strmsrv object. For general + * comments, see header of strmsrv.c. + * + * Copyright 2007, 2008, 2009 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ +#include "config.h" +#include <stdlib.h> +#include <string.h> +#include <assert.h> +#include <errno.h> +#include <ctype.h> + +#include "rsyslog.h" +#include "dirty.h" +#include "module-template.h" +#include "net.h" +#include "strmsrv.h" +#include "strms_sess.h" +#include "obj.h" +#include "errmsg.h" +#include "netstrm.h" +#include "msg.h" +#include "datetime.h" + + +/* static data */ +DEFobjStaticHelpers +DEFobjCurrIf(glbl) +DEFobjCurrIf(errmsg) +DEFobjCurrIf(netstrm) +DEFobjCurrIf(datetime) + +static int iMaxLine; /* maximum size of a single message */ + +/* forward definitions */ +static rsRetVal Close(strms_sess_t *pThis); + + +/* Standard-Constructor */ +BEGINobjConstruct(strms_sess) /* be sure to specify the object type also in END macro! */ +ENDobjConstruct(strms_sess) + + +/* ConstructionFinalizer + */ +static rsRetVal +strms_sessConstructFinalize(strms_sess_t *pThis) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, strms_sess); + if(pThis->pSrv->OnSessConstructFinalize != NULL) { + CHKiRet(pThis->pSrv->OnSessConstructFinalize(&pThis->pUsr)); + } + +finalize_it: + RETiRet; +} + + +/* destructor for the strms_sess object */ +BEGINobjDestruct(strms_sess) /* be sure to specify the object type also in END and CODESTART macros! */ +CODESTARTobjDestruct(strms_sess) + if(pThis->pStrm != NULL) + netstrm.Destruct(&pThis->pStrm); + + if(pThis->pSrv->pOnSessDestruct != NULL) { + pThis->pSrv->pOnSessDestruct(&pThis->pUsr); + } + /* now destruct our own properties */ + free(pThis->fromHost); + free(pThis->fromHostIP); +ENDobjDestruct(strms_sess) + + +/* debugprint for the strms_sess object */ +BEGINobjDebugPrint(strms_sess) /* be sure to specify the object type also in END and CODESTART macros! */ +CODESTARTobjDebugPrint(strms_sess) +ENDobjDebugPrint(strms_sess) + + +/* set property functions */ +/* set the hostname. Note that the caller *hands over* the string. That is, + * the caller no longer controls it once SetHost() has received it. Most importantly, + * the caller must not free it. -- rgerhards, 2008-04-24 + */ +static rsRetVal +SetHost(strms_sess_t *pThis, uchar *pszHost) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, strms_sess); + free(pThis->fromHost); + pThis->fromHost = pszHost; + RETiRet; +} + +/* set the remote host's IP. Note that the caller *hands over* the string. That is, + * the caller no longer controls it once SetHostIP() has received it. Most importantly, + * the caller must not free it. -- rgerhards, 2008-05-16 + */ +static rsRetVal +SetHostIP(strms_sess_t *pThis, uchar *pszHostIP) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, strms_sess); + free(pThis->fromHostIP); + pThis->fromHostIP = pszHostIP; + RETiRet; +} + +static rsRetVal +SetStrm(strms_sess_t *pThis, netstrm_t *pStrm) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, strms_sess); + pThis->pStrm = pStrm; + RETiRet; +} + + +/* set our parent, the strmsrv object */ +static rsRetVal +SetStrmsrv(strms_sess_t *pThis, strmsrv_t *pSrv) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, strms_sess); + ISOBJ_TYPE_assert(pSrv, strmsrv); + pThis->pSrv = pSrv; + RETiRet; +} + + +/* set our parent listener info*/ +static rsRetVal +SetLstnInfo(strms_sess_t *pThis, strmLstnPortList_t *pLstnInfo) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, strms_sess); + assert(pLstnInfo != NULL); + pThis->pLstnInfo = pLstnInfo; + RETiRet; +} + + +static rsRetVal +SetUsrP(strms_sess_t *pThis, void *pUsr) +{ + DEFiRet; + pThis->pUsr = pUsr; + RETiRet; +} + + +static void * +GetUsrP(strms_sess_t *pThis) +{ + return pThis->pUsr; +} + + +/* Closes a STRM session + * No attention is paid to the return code + * of close, so potential-double closes are not detected. + */ +static rsRetVal +Close(strms_sess_t *pThis) +{ + DEFiRet; + + ISOBJ_TYPE_assert(pThis, strms_sess); + netstrm.Destruct(&pThis->pStrm); + free(pThis->fromHost); + pThis->fromHost = NULL; /* not really needed, but... */ + free(pThis->fromHostIP); + pThis->fromHostIP = NULL; /* not really needed, but... */ + + RETiRet; +} + + + +/* Processes the data received via a STRM session. If there + * is no other way to handle it, data is discarded. + * Input parameter data is the data received, iLen is its + * len as returned from recv(). iLen must be 1 or more (that + * is errors must be handled by caller!). iSTRMSess must be + * the index of the STRM session that received the data. + * rgerhards 2005-07-04 + * And another change while generalizing. We now return either + * RS_RET_OK, which means the session should be kept open + * or anything else, which means it must be closed. + * rgerhards, 2008-03-01 + */ +static rsRetVal +DataRcvd(strms_sess_t *pThis, char *pData, size_t iLen) +{ + DEFiRet; + char *pEnd; + + ISOBJ_TYPE_assert(pThis, strms_sess); + assert(pData != NULL); + assert(iLen > 0); + + /* We now copy the message to the session buffer. */ + pEnd = pData + iLen; /* this is one off, which is intensional */ + + while(pData < pEnd) { + CHKiRet(pThis->pSrv->OnCharRcvd(pThis, (uchar)*pData++)); + } + +finalize_it: + RETiRet; +} + + +/* queryInterface function + * rgerhards, 2008-02-29 + */ +BEGINobjQueryInterface(strms_sess) +CODESTARTobjQueryInterface(strms_sess) + if(pIf->ifVersion != strms_sessCURR_IF_VERSION) { /* check for current version, increment on each change */ + ABORT_FINALIZE(RS_RET_INTERFACE_NOT_SUPPORTED); + } + + /* ok, we have the right interface, so let's fill it + * Please note that we may also do some backwards-compatibility + * work here (if we can support an older interface version - that, + * of course, also affects the "if" above). + */ + pIf->DebugPrint = strms_sessDebugPrint; + pIf->Construct = strms_sessConstruct; + pIf->ConstructFinalize = strms_sessConstructFinalize; + pIf->Destruct = strms_sessDestruct; + + pIf->Close = Close; + pIf->DataRcvd = DataRcvd; + + pIf->SetUsrP = SetUsrP; + pIf->GetUsrP = GetUsrP; + pIf->SetStrmsrv = SetStrmsrv; + pIf->SetLstnInfo = SetLstnInfo; + pIf->SetHost = SetHost; + pIf->SetHostIP = SetHostIP; + pIf->SetStrm = SetStrm; +finalize_it: +ENDobjQueryInterface(strms_sess) + + +/* exit our class + * rgerhards, 2008-03-10 + */ +BEGINObjClassExit(strms_sess, OBJ_IS_LOADABLE_MODULE) /* CHANGE class also in END MACRO! */ +CODESTARTObjClassExit(strms_sess) + /* release objects we no longer need */ + objRelease(errmsg, CORE_COMPONENT); + objRelease(netstrm, LM_NETSTRMS_FILENAME); + objRelease(datetime, CORE_COMPONENT); +ENDObjClassExit(strms_sess) + + +/* Initialize our class. Must be called as the very first method + * before anything else is called inside this class. + * rgerhards, 2008-02-29 + */ +BEGINObjClassInit(strms_sess, 1, OBJ_IS_CORE_MODULE) /* class, version - CHANGE class also in END MACRO! */ + /* request objects we use */ + CHKiRet(objUse(errmsg, CORE_COMPONENT)); + CHKiRet(objUse(netstrm, LM_NETSTRMS_FILENAME)); + CHKiRet(objUse(datetime, CORE_COMPONENT)); + + CHKiRet(objUse(glbl, CORE_COMPONENT)); + iMaxLine = glbl.GetMaxLine(); /* get maximum size we currently support */ + objRelease(glbl, CORE_COMPONENT); + + /* set our own handlers */ + OBJSetMethodHandler(objMethod_DEBUGPRINT, strms_sessDebugPrint); + OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, strms_sessConstructFinalize); +ENDObjClassInit(strms_sess) + +/* vim:set ai: + */ diff --git a/runtime/strms_sess.h b/runtime/strms_sess.h new file mode 100644 index 00000000..6483c0c3 --- /dev/null +++ b/runtime/strms_sess.h @@ -0,0 +1,75 @@ +/* Definitions for strms_sess class. This implements a session of the + * generic stream server. + * + * Copyright 2008, 2009 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ +#ifndef INCLUDED_STRMS_SESS_H +#define INCLUDED_STRMS_SESS_H + +#include "obj.h" + +/* a forward-definition, we are somewhat cyclic */ +struct strmsrv_s; + +/* the strms_sess object */ +struct strms_sess_s { + BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */ + strmsrv_t *pSrv; /* pointer back to my server (e.g. for callbacks) */ + strmLstnPortList_t *pLstnInfo; /* pointer back to listener info */ + netstrm_t *pStrm; +// uchar *pMsg; /* message (fragment) received */ + uchar *fromHost; + uchar *fromHostIP; + void *pUsr; /* a user-pointer */ +}; + + +/* interfaces */ +BEGINinterface(strms_sess) /* name must also be changed in ENDinterface macro! */ + INTERFACEObjDebugPrint(strms_sess); + rsRetVal (*Construct)(strms_sess_t **ppThis); + rsRetVal (*ConstructFinalize)(strms_sess_t __attribute__((unused)) *pThis); + rsRetVal (*Destruct)(strms_sess_t **ppThis); + rsRetVal (*Close)(strms_sess_t *pThis); + rsRetVal (*DataRcvd)(strms_sess_t *pThis, char *pData, size_t iLen); + /* set methods */ + rsRetVal (*SetStrmsrv)(strms_sess_t *pThis, struct strmsrv_s *pSrv); + rsRetVal (*SetLstnInfo)(strms_sess_t *pThis, strmLstnPortList_t *pLstnInfo); + rsRetVal (*SetUsrP)(strms_sess_t*, void*); + void* (*GetUsrP)(strms_sess_t*); + rsRetVal (*SetHost)(strms_sess_t *pThis, uchar*); + rsRetVal (*SetHostIP)(strms_sess_t *pThis, uchar*); + rsRetVal (*SetStrm)(strms_sess_t *pThis, netstrm_t*); + rsRetVal (*SetOnMsgReceive)(strms_sess_t *pThis, rsRetVal (*OnMsgReceive)(strms_sess_t*, uchar*, int)); +ENDinterface(strms_sess) +#define strms_sessCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */ +/* interface changes + * to version v2, rgerhards, 2009-05-22 + * - Data structures changed + * - SetLstnInfo entry point added + */ + + +/* prototypes */ +PROTOTYPEObj(strms_sess); + + +#endif /* #ifndef INCLUDED_STRMS_SESS_H */ diff --git a/runtime/strmsrv.c b/runtime/strmsrv.c new file mode 100644 index 00000000..3dc53a97 --- /dev/null +++ b/runtime/strmsrv.c @@ -0,0 +1,966 @@ +/* strmsrv.c + * + * This builds a basic stream server. It handles connection creation but + * not any protocol. Instead, it calls a "data received" entry point of the + * caller with any data received, in which case the caller must react accordingly. + * This module works together with the netstream drivers. + * + * There are actually two classes within the stream server code: one is + * the strmsrv itself, the other one is its sessions. This is a helper + * class to strmsrv. + * + * File begun on 2009-06-01 by RGerhards based on strmsrv.c. Note that strmsrv is + * placed under LGPL, which is possible because I carefully evaluated and + * eliminated all those parts of strmsrv which were not written by me. + * + * TODO: I would consider it useful to migrate tcpsrv.c/tcps_sess.c to this stream + * class here. The requires a little bit redesign, but should not be too hard. The + * core idea, already begun here, is that we still support lots of callbacks, but + * provide "canned" implementations for standard cases. That way, most upper-layer + * modules can be kept rather simple and without any extra overhead. Note that + * to support this, tcps_sess.c would need to extract the message reception state + * machine to a separate module which then is called via the DoCharRcvd() interface + * of this class here. -- rgerhards, 2009-06-01 + * + * Copyright 2007, 2008, 2009 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ + +#include "config.h" +#include <stdlib.h> +#include <assert.h> +#include <string.h> +#include <errno.h> +#include <unistd.h> +#include <stdarg.h> +#include <ctype.h> +#include <netinet/in.h> +#include <netdb.h> +#include <sys/types.h> +#include <sys/socket.h> +#if HAVE_FCNTL_H +#include <fcntl.h> +#endif +#include "rsyslog.h" +#include "dirty.h" +#include "cfsysline.h" +#include "module-template.h" +#include "net.h" +#include "srUtils.h" +#include "conf.h" +#include "strmsrv.h" +#include "obj.h" +#include "glbl.h" +#include "netstrms.h" +#include "netstrm.h" +#include "nssel.h" +#include "errmsg.h" +#include "unicode-helper.h" + +MODULE_TYPE_LIB + +/* defines */ +#define STRMSESS_MAX_DEFAULT 200 /* default for nbr of strm sessions if no number is given */ +#define STRMLSTN_MAX_DEFAULT 20 /* default for nbr of listeners */ + +/* static data */ +DEFobjStaticHelpers +DEFobjCurrIf(conf) +DEFobjCurrIf(glbl) +DEFobjCurrIf(strms_sess) +DEFobjCurrIf(errmsg) +DEFobjCurrIf(net) +DEFobjCurrIf(netstrms) +DEFobjCurrIf(netstrm) +DEFobjCurrIf(nssel) + +/* forward definitions */ +static rsRetVal create_strm_socket(strmsrv_t *pThis); + +/* standard callbacks, if the caller did not provide us with them (this helps keep us + * flexible while at the same time permits very simple upper-layer modules) + */ +/* this shall go into a specific ACL module! */ +static int +isPermittedHost(struct sockaddr __attribute__((unused)) *addr, char __attribute__((unused)) *fromHostFQDN, + void __attribute__((unused)) *pUsrSrv, void __attribute__((unused)) *pUsrSess) +{ + return 1; +} + + +static rsRetVal +doOpenLstnSocks(strmsrv_t *pSrv) +{ + ISOBJ_TYPE_assert(pSrv, strmsrv); + return create_strm_socket(pSrv); +} + + +static rsRetVal +doRcvData(strms_sess_t *pSess, char *buf, size_t lenBuf, ssize_t *piLenRcvd) +{ + DEFiRet; + assert(pSess != NULL); + assert(piLenRcvd != NULL); + + *piLenRcvd = lenBuf; + CHKiRet(netstrm.Rcv(pSess->pStrm, (uchar*) buf, piLenRcvd)); +finalize_it: + RETiRet; +} + +static rsRetVal +onRegularClose(strms_sess_t *pSess) +{ + DEFiRet; + assert(pSess != NULL); + + /* process any incomplete frames left over */ + //strms_sess.PrepareClose(pSess); + /* Session closed */ + strms_sess.Close(pSess); + RETiRet; +} + + +static rsRetVal +onErrClose(strms_sess_t *pSess) +{ + DEFiRet; + assert(pSess != NULL); + + strms_sess.Close(pSess); + RETiRet; +} + +/* ------------------------------ end callbacks ------------------------------ */ + +/* add new listener port to listener port list + * rgerhards, 2009-05-21 + */ +static inline rsRetVal +addNewLstnPort(strmsrv_t *pThis, uchar *pszPort) +{ + strmLstnPortList_t *pEntry; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, strmsrv); + + /* create entry */ + CHKmalloc(pEntry = malloc(sizeof(strmLstnPortList_t))); + pEntry->pszPort = pszPort; + pEntry->pSrv = pThis; + CHKmalloc(pEntry->pszInputName = ustrdup(pThis->pszInputName)); + + /* and add to list */ + pEntry->pNext = pThis->pLstnPorts; + pThis->pLstnPorts = pEntry; + +finalize_it: + RETiRet; +} + + +/* configure STRM listener settings. + * Note: pszPort is handed over to us - the caller MUST NOT free it! + * rgerhards, 2008-03-20 + */ +static rsRetVal +configureSTRMListen(strmsrv_t *pThis, uchar *pszPort) +{ + int i; + uchar *pPort = pszPort; + DEFiRet; + + assert(pszPort != NULL); + ISOBJ_TYPE_assert(pThis, strmsrv); + + /* extract port */ + i = 0; + while(isdigit((int) *pPort)) { + i = i * 10 + *pPort++ - '0'; + } + + if(i >= 0 && i <= 65535) { + CHKiRet(addNewLstnPort(pThis, pszPort)); + } else { + errmsg.LogError(0, NO_ERRCODE, "Invalid STRM listen port %s - ignored.\n", pszPort); + } + +finalize_it: + RETiRet; +} + + +/* Initialize the session table + * returns 0 if OK, somewhat else otherwise + */ +static rsRetVal +STRMSessTblInit(strmsrv_t *pThis) +{ + DEFiRet; + + ISOBJ_TYPE_assert(pThis, strmsrv); + assert(pThis->pSessions == NULL); + + dbgprintf("Allocating buffer for %d STRM sessions.\n", pThis->iSessMax); + if((pThis->pSessions = (strms_sess_t **) calloc(pThis->iSessMax, sizeof(strms_sess_t *))) == NULL) { + dbgprintf("Error: STRMSessInit() could not alloc memory for STRM session table.\n"); + ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); + } + +finalize_it: + RETiRet; +} + + +/* find a free spot in the session table. If the table + * is full, -1 is returned, else the index of the free + * entry (0 or higher). + */ +static int +STRMSessTblFindFreeSpot(strmsrv_t *pThis) +{ + register int i; + + ISOBJ_TYPE_assert(pThis, strmsrv); + + for(i = 0 ; i < pThis->iSessMax ; ++i) { + if(pThis->pSessions[i] == NULL) + break; + } + + return((i < pThis->iSessMax) ? i : -1); +} + + +/* Get the next session index. Free session tables entries are + * skipped. This function is provided the index of the last + * session entry, or -1 if no previous entry was obtained. It + * returns the index of the next session or -1, if there is no + * further entry in the table. Please note that the initial call + * might as well return -1, if there is no session at all in the + * session table. + */ +static int +STRMSessGetNxtSess(strmsrv_t *pThis, int iCurr) +{ + register int i; + + BEGINfunc + ISOBJ_TYPE_assert(pThis, strmsrv); + assert(pThis->pSessions != NULL); + for(i = iCurr + 1 ; i < pThis->iSessMax ; ++i) { + if(pThis->pSessions[i] != NULL) + break; + } + + ENDfunc + return((i < pThis->iSessMax) ? i : -1); +} + + +/* De-Initialize STRM listner sockets. + * This function deinitializes everything, including freeing the + * session table. No STRM listen receive operations are permitted + * unless the subsystem is reinitialized. + * rgerhards, 2007-06-21 + */ +static void deinit_strm_listener(strmsrv_t *pThis) +{ + int i; + strmLstnPortList_t *pEntry; + strmLstnPortList_t *pDel; + + ISOBJ_TYPE_assert(pThis, strmsrv); + + if(pThis->pSessions != NULL) { + /* close all STRM connections! */ + i = STRMSessGetNxtSess(pThis, -1); + while(i != -1) { + strms_sess.Destruct(&pThis->pSessions[i]); + /* now get next... */ + i = STRMSessGetNxtSess(pThis, i); + } + + /* we are done with the session table - so get rid of it... */ + free(pThis->pSessions); + pThis->pSessions = NULL; /* just to make sure... */ + } + + /* free list of strm listen ports */ + pEntry = pThis->pLstnPorts; + while(pEntry != NULL) { + free(pEntry->pszPort); + free(pEntry->pszInputName); + pDel = pEntry; + pEntry = pEntry->pNext; + free(pDel); + } + + /* finally close our listen streams */ + for(i = 0 ; i < pThis->iLstnMax ; ++i) { + netstrm.Destruct(pThis->ppLstn + i); + } +} + + +/* add a listen socket to our listen socket array. This is a callback + * invoked from the netstrm class. -- rgerhards, 2008-04-23 + */ +static rsRetVal +addStrmLstn(void *pUsr, netstrm_t *pLstn) +{ + strmLstnPortList_t *pPortList = (strmLstnPortList_t *) pUsr; + strmsrv_t *pThis = pPortList->pSrv; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, strmsrv); + ISOBJ_TYPE_assert(pLstn, netstrm); + + if(pThis->iLstnMax >= STRMLSTN_MAX_DEFAULT) + ABORT_FINALIZE(RS_RET_MAX_LSTN_REACHED); + + pThis->ppLstn[pThis->iLstnMax] = pLstn; + pThis->ppLstnPort[pThis->iLstnMax] = pPortList; + ++pThis->iLstnMax; + +finalize_it: + RETiRet; +} + + +/* Initialize STRM listener socket for a single port + * rgerhards, 2009-05-21 + */ +static inline rsRetVal +initSTRMListener(strmsrv_t *pThis, strmLstnPortList_t *pPortEntry) +{ + DEFiRet; + + ISOBJ_TYPE_assert(pThis, strmsrv); + assert(pPortEntry != NULL); + + /* TODO: add capability to specify local listen address! */ + CHKiRet(netstrm.LstnInit(pThis->pNS, (void*)pPortEntry, addStrmLstn, pPortEntry->pszPort, NULL, pThis->iSessMax)); + +finalize_it: + RETiRet; +} + + +/* Initialize STRM sockets (for listener) and listens on them */ +static rsRetVal +create_strm_socket(strmsrv_t *pThis) +{ + strmLstnPortList_t *pEntry; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, strmsrv); + + /* init all configured ports */ + pEntry = pThis->pLstnPorts; + while(pEntry != NULL) { + CHKiRet(initSTRMListener(pThis, pEntry)); + pEntry = pEntry->pNext; + } + + /* OK, we had success. Now it is also time to + * initialize our connections + */ + if(STRMSessTblInit(pThis) != 0) { + /* OK, we are in some trouble - we could not initialize the + * session table, so we can not continue. We need to free all + * we have assigned so far, because we can not really use it... + */ + errmsg.LogError(0, RS_RET_ERR, "Could not initialize STRM session table, suspending STRM message reception."); + ABORT_FINALIZE(RS_RET_ERR); + } + +finalize_it: + RETiRet; +} + + +/* Accept new STRM connection; make entry in session table. If there + * is no more space left in the connection table, the new STRM + * connection is immediately dropped. + * ppSess has a pointer to the newly created session, if it succeeds. + * If it does not succeed, no session is created and ppSess is + * undefined. If the user has provided an OnSessAccept Callback, + * this one is executed immediately after creation of the + * session object, so that it can do its own initialization. + * rgerhards, 2008-03-02 + */ +static rsRetVal +SessAccept(strmsrv_t *pThis, strmLstnPortList_t *pLstnInfo, strms_sess_t **ppSess, netstrm_t *pStrm) +{ + DEFiRet; + strms_sess_t *pSess = NULL; + netstrm_t *pNewStrm = NULL; + int iSess = -1; + struct sockaddr_storage *addr; + uchar *fromHostFQDN = NULL; + uchar *fromHostIP = NULL; + + ISOBJ_TYPE_assert(pThis, strmsrv); + assert(pLstnInfo != NULL); + + CHKiRet(netstrm.AcceptConnReq(pStrm, &pNewStrm)); + + /* Add to session list */ + iSess = STRMSessTblFindFreeSpot(pThis); + if(iSess == -1) { + errno = 0; + errmsg.LogError(0, RS_RET_MAX_SESS_REACHED, "too many strm sessions - dropping incoming request"); + ABORT_FINALIZE(RS_RET_MAX_SESS_REACHED); + } + + if(pThis->bUseKeepAlive) { + CHKiRet(netstrm.EnableKeepAlive(pNewStrm)); + } + + /* we found a free spot and can construct our session object */ + CHKiRet(strms_sess.Construct(&pSess)); + CHKiRet(strms_sess.SetStrmsrv(pSess, pThis)); + CHKiRet(strms_sess.SetLstnInfo(pSess, pLstnInfo)); + + /* get the host name */ + CHKiRet(netstrm.GetRemoteHName(pNewStrm, &fromHostFQDN)); + CHKiRet(netstrm.GetRemoteIP(pNewStrm, &fromHostIP)); + CHKiRet(netstrm.GetRemAddr(pNewStrm, &addr)); + /* TODO: check if we need to strip the domain name here -- rgerhards, 2008-04-24 */ + + /* Here we check if a host is permitted to send us messages. If it isn't, we do not further + * process the message but log a warning (if we are configured to do this). + * rgerhards, 2005-09-26 + */ + if(pThis->pIsPermittedHost != NULL + && !pThis->pIsPermittedHost((struct sockaddr*) addr, (char*) fromHostFQDN, pThis->pUsr, pSess->pUsr)) { + dbgprintf("%s is not an allowed sender\n", fromHostFQDN); + if(glbl.GetOption_DisallowWarning()) { + errno = 0; + errmsg.LogError(0, RS_RET_HOST_NOT_PERMITTED, "STRM message from disallowed sender %s discarded", fromHostFQDN); + } + ABORT_FINALIZE(RS_RET_HOST_NOT_PERMITTED); + } + + /* OK, we have an allowed sender, so let's continue, what + * means we can finally fill in the session object. + */ + CHKiRet(strms_sess.SetHost(pSess, fromHostFQDN)); + fromHostFQDN = NULL; /* we handed this string over */ + CHKiRet(strms_sess.SetHostIP(pSess, fromHostIP)); + fromHostIP = NULL; /* we handed this string over */ + CHKiRet(strms_sess.SetStrm(pSess, pNewStrm)); + pNewStrm = NULL; /* prevent it from being freed in error handler, now done in strms_sess! */ + CHKiRet(strms_sess.ConstructFinalize(pSess)); + + /* check if we need to call our callback */ + if(pThis->pOnSessAccept != NULL) { + CHKiRet(pThis->pOnSessAccept(pThis, pSess)); + } + + *ppSess = pSess; + pThis->pSessions[iSess] = pSess; + pSess = NULL; /* this is now also handed over */ + +finalize_it: + if(iRet != RS_RET_OK) { + if(pSess != NULL) + strms_sess.Destruct(&pSess); + if(pNewStrm != NULL) + netstrm.Destruct(&pNewStrm); + free(fromHostFQDN); + free(fromHostIP); + } + + RETiRet; +} + + +static void +RunCancelCleanup(void *arg) +{ + nssel_t **ppSel = (nssel_t**) arg; + + if(*ppSel != NULL) + nssel.Destruct(ppSel); +} + + +/* This function is called to gather input. */ +#pragma GCC diagnostic ignored "-Wempty-body" +static rsRetVal +Run(strmsrv_t *pThis) +{ + DEFiRet; + int nfds; + int i; + int iSTRMSess; + int bIsReady; + strms_sess_t *pNewSess; + nssel_t *pSel; + ssize_t iRcvd; + + ISOBJ_TYPE_assert(pThis, strmsrv); + + /* this is an endless loop - it is terminated by the framework canelling + * this thread. Thus, we also need to instantiate a cancel cleanup handler + * to prevent us from leaking anything. -- rgerharsd, 20080-04-24 + */ + pthread_cleanup_push(RunCancelCleanup, (void*) &pSel); + while(1) { + CHKiRet(nssel.Construct(&pSel)); + // TODO: set driver + CHKiRet(nssel.ConstructFinalize(pSel)); + + /* Add the STRM listen sockets to the list of read descriptors. */ + for(i = 0 ; i < pThis->iLstnMax ; ++i) { + CHKiRet(nssel.Add(pSel, pThis->ppLstn[i], NSDSEL_RD)); + } + + /* do the sessions */ + iSTRMSess = STRMSessGetNxtSess(pThis, -1); + while(iSTRMSess != -1) { + /* TODO: access to pNsd is NOT really CLEAN, use method... */ + CHKiRet(nssel.Add(pSel, pThis->pSessions[iSTRMSess]->pStrm, NSDSEL_RD)); + /* now get next... */ + iSTRMSess = STRMSessGetNxtSess(pThis, iSTRMSess); + } + + /* wait for io to become ready */ + CHKiRet(nssel.Wait(pSel, &nfds)); + + for(i = 0 ; i < pThis->iLstnMax ; ++i) { + CHKiRet(nssel.IsReady(pSel, pThis->ppLstn[i], NSDSEL_RD, &bIsReady, &nfds)); + if(bIsReady) { + dbgprintf("New connect on NSD %p.\n", pThis->ppLstn[i]); + SessAccept(pThis, pThis->ppLstnPort[i], &pNewSess, pThis->ppLstn[i]); + --nfds; /* indicate we have processed one */ + } + } + + /* now check the sessions */ + iSTRMSess = STRMSessGetNxtSess(pThis, -1); + while(nfds && iSTRMSess != -1) { + CHKiRet(nssel.IsReady(pSel, pThis->pSessions[iSTRMSess]->pStrm, NSDSEL_RD, &bIsReady, &nfds)); + if(bIsReady) { + char buf[8*1024]; /* reception buffer - may hold a partial or multiple messages */ + dbgprintf("netstream %p with new data\n", pThis->pSessions[iSTRMSess]->pStrm); + + /* Receive message */ + iRet = pThis->pRcvData(pThis->pSessions[iSTRMSess], buf, sizeof(buf), &iRcvd); + switch(iRet) { + case RS_RET_CLOSED: + pThis->pOnRegularClose(pThis->pSessions[iSTRMSess]); + strms_sess.Destruct(&pThis->pSessions[iSTRMSess]); + break; + case RS_RET_RETRY: + /* we simply ignore retry - this is not an error, but we also have not received anything */ + break; + case RS_RET_OK: + /* valid data received, process it! */ + if(strms_sess.DataRcvd(pThis->pSessions[iSTRMSess], buf, iRcvd) != RS_RET_OK) { + /* in this case, something went awfully wrong. + * We are instructed to terminate the session. + */ + errmsg.LogError(0, NO_ERRCODE, "Tearing down STRM Session %d - see " + "previous messages for reason(s)\n", iSTRMSess); + pThis->pOnErrClose(pThis->pSessions[iSTRMSess]); + strms_sess.Destruct(&pThis->pSessions[iSTRMSess]); + } + break; + default: + errno = 0; + errmsg.LogError(0, iRet, "netstream session %p will be closed due to error\n", + pThis->pSessions[iSTRMSess]->pStrm); + pThis->pOnErrClose(pThis->pSessions[iSTRMSess]); + strms_sess.Destruct(&pThis->pSessions[iSTRMSess]); + break; + } + --nfds; /* indicate we have processed one */ + } + iSTRMSess = STRMSessGetNxtSess(pThis, iSTRMSess); + } + CHKiRet(nssel.Destruct(&pSel)); +finalize_it: /* this is a very special case - this time only we do not exit the function, + * because that would not help us either. So we simply retry it. Let's see + * if that actually is a better idea. Exiting the loop wasn't we always + * crashed, which made sense (the rest of the engine was not prepared for + * that) -- rgerhards, 2008-05-19 + */ + /*EMPTY*/; + } + + /* note that this point is usually not reached */ + pthread_cleanup_pop(0); /* remove cleanup handler */ + + RETiRet; +} +#pragma GCC diagnostic warning "-Wempty-body" + + +/* Standard-Constructor */ +BEGINobjConstruct(strmsrv) /* be sure to specify the object type also in END macro! */ + pThis->iSessMax = STRMSESS_MAX_DEFAULT; /* TODO: useful default ;) */ + /* set default callbacks (used if caller does not overwrite them) */ + pThis->pIsPermittedHost = isPermittedHost; + pThis->OpenLstnSocks = doOpenLstnSocks; + pThis->pRcvData = doRcvData; + pThis->pOnRegularClose = onRegularClose; + pThis->pOnErrClose = onErrClose; + /* session specific callbacks */ + //pThis->OnSessConstructFinalize = + //pThis->pOnSessDestruct = +ENDobjConstruct(strmsrv) + + +/* ConstructionFinalizer */ +static rsRetVal +strmsrvConstructFinalize(strmsrv_t *pThis) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, strmsrv); + + /* prepare network stream subsystem */ + CHKiRet(netstrms.Construct(&pThis->pNS)); + CHKiRet(netstrms.SetDrvrMode(pThis->pNS, pThis->iDrvrMode)); + if(pThis->pszDrvrAuthMode != NULL) + CHKiRet(netstrms.SetDrvrAuthMode(pThis->pNS, pThis->pszDrvrAuthMode)); + if(pThis->pPermPeers != NULL) + CHKiRet(netstrms.SetDrvrPermPeers(pThis->pNS, pThis->pPermPeers)); + // TODO: set driver! + CHKiRet(netstrms.ConstructFinalize(pThis->pNS)); + + /* set up listeners */ + CHKmalloc(pThis->ppLstn = calloc(STRMLSTN_MAX_DEFAULT, sizeof(netstrm_t*))); + CHKmalloc(pThis->ppLstnPort = calloc(STRMLSTN_MAX_DEFAULT, sizeof(strmLstnPortList_t*))); + iRet = pThis->OpenLstnSocks(pThis); + +finalize_it: + if(iRet != RS_RET_OK) { + if(pThis->pNS != NULL) + netstrms.Destruct(&pThis->pNS); + } + RETiRet; +} + + +/* destructor for the strmsrv object */ +BEGINobjDestruct(strmsrv) /* be sure to specify the object type also in END and CODESTART macros! */ +CODESTARTobjDestruct(strmsrv) + if(pThis->OnDestruct != NULL) + pThis->OnDestruct(pThis->pUsr); + + deinit_strm_listener(pThis); + + if(pThis->pNS != NULL) + netstrms.Destruct(&pThis->pNS); + free(pThis->pszDrvrAuthMode); + free(pThis->ppLstn); + free(pThis->ppLstnPort); + free(pThis->pszInputName); +ENDobjDestruct(strmsrv) + + +/* debugprint for the strmsrv object */ +BEGINobjDebugPrint(strmsrv) /* be sure to specify the object type also in END and CODESTART macros! */ +CODESTARTobjDebugPrint(strmsrv) +ENDobjDebugPrint(strmsrv) + +/* set functions */ +static rsRetVal +SetCBIsPermittedHost(strmsrv_t *pThis, int (*pCB)(struct sockaddr *addr, char *fromHostFQDN, void*, void*)) +{ + DEFiRet; + pThis->pIsPermittedHost = pCB; + RETiRet; +} + +static rsRetVal +SetCBOnSessAccept(strmsrv_t *pThis, rsRetVal (*pCB)(strmsrv_t*, strms_sess_t*)) +{ + DEFiRet; + pThis->pOnSessAccept = pCB; + RETiRet; +} + +static rsRetVal +SetCBOnDestruct(strmsrv_t *pThis, rsRetVal (*pCB)(void*)) +{ + DEFiRet; + pThis->OnDestruct = pCB; + RETiRet; +} + +static rsRetVal +SetCBOnSessConstructFinalize(strmsrv_t *pThis, rsRetVal (*pCB)(void*)) +{ + DEFiRet; + pThis->OnSessConstructFinalize = pCB; + RETiRet; +} + +static rsRetVal +SetCBOnSessDestruct(strmsrv_t *pThis, rsRetVal (*pCB)(void*)) +{ + DEFiRet; + pThis->pOnSessDestruct = pCB; + RETiRet; +} + +static rsRetVal +SetCBOnRegularClose(strmsrv_t *pThis, rsRetVal (*pCB)(strms_sess_t*)) +{ + DEFiRet; + pThis->pOnRegularClose = pCB; + RETiRet; +} + +static rsRetVal +SetCBOnErrClose(strmsrv_t *pThis, rsRetVal (*pCB)(strms_sess_t*)) +{ + DEFiRet; + pThis->pOnErrClose = pCB; + RETiRet; +} + +static rsRetVal +SetCBOpenLstnSocks(strmsrv_t *pThis, rsRetVal (*pCB)(strmsrv_t*)) +{ + DEFiRet; + pThis->OpenLstnSocks = pCB; + RETiRet; +} + +static rsRetVal +SetUsrP(strmsrv_t *pThis, void *pUsr) +{ + DEFiRet; + pThis->pUsr = pUsr; + RETiRet; +} + +static rsRetVal +SetKeepAlive(strmsrv_t *pThis, int iVal) +{ + DEFiRet; + dbgprintf("keep-alive set to %d\n", iVal); + pThis->bUseKeepAlive = iVal; + RETiRet; +} + +static rsRetVal +SetOnCharRcvd(strmsrv_t *pThis, rsRetVal (*OnCharRcvd)(strms_sess_t*, uchar)) +{ + DEFiRet; + assert(OnCharRcvd != NULL); + pThis->OnCharRcvd = OnCharRcvd; + RETiRet; +} + +/* Set the input name to use -- rgerhards, 2008-12-10 */ +static rsRetVal +SetInputName(strmsrv_t *pThis, uchar *name) +{ + uchar *pszName; + DEFiRet; + ISOBJ_TYPE_assert(pThis, strmsrv); + if(name == NULL) + pszName = NULL; + else + CHKmalloc(pszName = ustrdup(name)); + free(pThis->pszInputName); + pThis->pszInputName = pszName; +finalize_it: + RETiRet; +} + + +/* here follows a number of methods that shuffle authentication settings down + * to the drivers. Drivers not supporting these settings may return an error + * state. + * -------------------------------------------------------------------------- */ + +/* set the driver mode -- rgerhards, 2008-04-30 */ +static rsRetVal +SetDrvrMode(strmsrv_t *pThis, int iMode) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, strmsrv); + pThis->iDrvrMode = iMode; + RETiRet; +} + + +/* set the driver authentication mode -- rgerhards, 2008-05-19 */ +static rsRetVal +SetDrvrAuthMode(strmsrv_t *pThis, uchar *mode) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, strmsrv); + CHKmalloc(pThis->pszDrvrAuthMode = ustrdup(mode)); +finalize_it: + RETiRet; +} + + +/* set the driver's permitted peers -- rgerhards, 2008-05-19 */ +static rsRetVal +SetDrvrPermPeers(strmsrv_t *pThis, permittedPeers_t *pPermPeers) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, strmsrv); + pThis->pPermPeers = pPermPeers; + RETiRet; +} + + +/* End of methods to shuffle autentication settings to the driver.; + + * -------------------------------------------------------------------------- */ + + +/* set max number of sessions + * this must be called before ConstructFinalize, or it will have no effect! + * rgerhards, 2009-04-09 + */ +static rsRetVal +SetSessMax(strmsrv_t *pThis, int iMax) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, strmsrv); + pThis->iSessMax = iMax; + RETiRet; +} + + +/* queryInterface function + * rgerhards, 2008-02-29 + */ +BEGINobjQueryInterface(strmsrv) +CODESTARTobjQueryInterface(strmsrv) + if(pIf->ifVersion != strmsrvCURR_IF_VERSION) { /* check for current version, increment on each change */ + ABORT_FINALIZE(RS_RET_INTERFACE_NOT_SUPPORTED); + } + + /* ok, we have the right interface, so let's fill it + * Please note that we may also do some backwards-compatibility + * work here (if we can support an older interface version - that, + * of course, also affects the "if" above). + */ + pIf->DebugPrint = strmsrvDebugPrint; + pIf->Construct = strmsrvConstruct; + pIf->ConstructFinalize = strmsrvConstructFinalize; + pIf->Destruct = strmsrvDestruct; + + pIf->configureSTRMListen = configureSTRMListen; + pIf->create_strm_socket = create_strm_socket; + pIf->Run = Run; + + pIf->SetKeepAlive = SetKeepAlive; + pIf->SetUsrP = SetUsrP; + pIf->SetInputName = SetInputName; + pIf->SetSessMax = SetSessMax; + pIf->SetDrvrMode = SetDrvrMode; + pIf->SetDrvrAuthMode = SetDrvrAuthMode; + pIf->SetDrvrPermPeers = SetDrvrPermPeers; + pIf->SetCBIsPermittedHost = SetCBIsPermittedHost; + pIf->SetCBOpenLstnSocks = SetCBOpenLstnSocks; + pIf->SetCBOnSessAccept = SetCBOnSessAccept; + pIf->SetCBOnSessConstructFinalize = SetCBOnSessConstructFinalize; + pIf->SetCBOnSessDestruct = SetCBOnSessDestruct; + pIf->SetCBOnDestruct = SetCBOnDestruct; + pIf->SetCBOnRegularClose = SetCBOnRegularClose; + pIf->SetCBOnErrClose = SetCBOnErrClose; + pIf->SetOnCharRcvd = SetOnCharRcvd; + +finalize_it: +ENDobjQueryInterface(strmsrv) + + +/* exit our class + * rgerhards, 2008-03-10 + */ +BEGINObjClassExit(strmsrv, OBJ_IS_LOADABLE_MODULE) /* CHANGE class also in END MACRO! */ +CODESTARTObjClassExit(strmsrv) + /* release objects we no longer need */ + objRelease(strms_sess, DONT_LOAD_LIB); + objRelease(conf, CORE_COMPONENT); + objRelease(glbl, CORE_COMPONENT); + objRelease(errmsg, CORE_COMPONENT); + objRelease(netstrms, DONT_LOAD_LIB); + objRelease(nssel, DONT_LOAD_LIB); + objRelease(netstrm, LM_NETSTRMS_FILENAME); + objRelease(net, LM_NET_FILENAME); +ENDObjClassExit(strmsrv) + + +/* Initialize our class. Must be called as the very first method + * before anything else is called inside this class. + * rgerhards, 2008-02-29 + */ +BEGINObjClassInit(strmsrv, 1, OBJ_IS_LOADABLE_MODULE) /* class, version - CHANGE class also in END MACRO! */ + /* request objects we use */ + CHKiRet(objUse(errmsg, CORE_COMPONENT)); + CHKiRet(objUse(net, LM_NET_FILENAME)); + CHKiRet(objUse(netstrms, LM_NETSTRMS_FILENAME)); + CHKiRet(objUse(netstrm, DONT_LOAD_LIB)); + CHKiRet(objUse(nssel, DONT_LOAD_LIB)); + CHKiRet(objUse(strms_sess, DONT_LOAD_LIB)); + CHKiRet(objUse(conf, CORE_COMPONENT)); + CHKiRet(objUse(glbl, CORE_COMPONENT)); + + /* set our own handlers */ + OBJSetMethodHandler(objMethod_DEBUGPRINT, strmsrvDebugPrint); + OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, strmsrvConstructFinalize); +ENDObjClassInit(strmsrv) + + +/* --------------- here now comes the plumbing that makes as a library module --------------- */ + + +BEGINmodExit +CODESTARTmodExit + /* de-init in reverse order! */ + strmsrvClassExit(); + strms_sessClassExit(); +ENDmodExit + + +BEGINqueryEtryPt +CODESTARTqueryEtryPt +CODEqueryEtryPt_STD_LIB_QUERIES +ENDqueryEtryPt + + +BEGINmodInit() +CODESTARTmodInit + *ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */ + + /* Initialize all classes that are in our module - this includes ourselfs */ + CHKiRet(strms_sessClassInit(pModInfo)); + CHKiRet(strmsrvClassInit(pModInfo)); /* must be done after strms_sess, as we use it */ +ENDmodInit + +/* vim:set ai: + */ diff --git a/runtime/strmsrv.h b/runtime/strmsrv.h new file mode 100644 index 00000000..86e529c3 --- /dev/null +++ b/runtime/strmsrv.h @@ -0,0 +1,112 @@ +/* Definitions for strmsrv class. + * + * Copyright 2008, 2009 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ +#ifndef INCLUDED_STRMSRV_H +#define INCLUDED_STRMSRV_H + +#include "obj.h" +#include "strms_sess.h" + +/* list of strm listen ports */ +struct strmLstnPortList_s { + uchar *pszPort; /**< the ports the listener shall listen on */ + uchar *pszInputName; /**< value to be used as input name */ + strmsrv_t *pSrv; /**< pointer to higher-level server instance */ + strmLstnPortList_t *pNext; /**< next port or NULL */ +}; + + +/* the strmsrv object */ +struct strmsrv_s { + BEGINobjInstance; /**< Data to implement generic object - MUST be the first data element! */ + int bUseKeepAlive; /**< use socket layer KEEPALIVE handling? */ + netstrms_t *pNS; /**< pointer to network stream subsystem */ + int iDrvrMode; /**< mode of the stream driver to use */ + uchar *pszDrvrAuthMode; /**< auth mode of the stream driver to use */ + uchar *pszInputName; /**< value to be used as input name */ + permittedPeers_t *pPermPeers;/**< driver's permitted peers */ + int iLstnMax; /**< max nbr of listeners currently supported */ + netstrm_t **ppLstn; /**< our netstream listners */ + strmLstnPortList_t **ppLstnPort; /**< pointer to relevant listen port description */ + int iSessMax; /**< max number of sessions supported */ + strmLstnPortList_t *pLstnPorts; /**< head pointer for listen ports */ + int addtlFrameDelim; /**< additional frame delimiter for plain STRM syslog framing (e.g. to handle NetScreen) */ + strms_sess_t **pSessions;/**< array of all of our sessions */ + void *pUsr; /**< a user-settable pointer (provides extensibility for "derived classes")*/ + /* callbacks */ + int (*pIsPermittedHost)(struct sockaddr *addr, char *fromHostFQDN, void*pUsrSrv, void*pUsrSess); + rsRetVal (*pRcvData)(strms_sess_t*, char*, size_t, ssize_t *); + rsRetVal (*OpenLstnSocks)(struct strmsrv_s*); + rsRetVal (*pOnListenDeinit)(void*); + rsRetVal (*OnDestruct)(void*); + rsRetVal (*pOnRegularClose)(strms_sess_t *pSess); + rsRetVal (*pOnErrClose)(strms_sess_t *pSess); + /* session specific callbacks */ + rsRetVal (*pOnSessAccept)(strmsrv_t *, strms_sess_t*); + rsRetVal (*OnSessConstructFinalize)(void*); + rsRetVal (*pOnSessDestruct)(void*); + rsRetVal (*OnCharRcvd)(strms_sess_t*, uchar); +}; + + +/* interfaces */ +BEGINinterface(strmsrv) /* name must also be changed in ENDinterface macro! */ + INTERFACEObjDebugPrint(strmsrv); + rsRetVal (*Construct)(strmsrv_t **ppThis); + rsRetVal (*ConstructFinalize)(strmsrv_t __attribute__((unused)) *pThis); + rsRetVal (*Destruct)(strmsrv_t **ppThis); + rsRetVal (*configureSTRMListen)(strmsrv_t*, uchar *pszPort); + //rsRetVal (*SessAccept)(strmsrv_t *pThis, strmLstnPortList_t*, strms_sess_t **ppSess, netstrm_t *pStrm); + rsRetVal (*create_strm_socket)(strmsrv_t *pThis); + rsRetVal (*Run)(strmsrv_t *pThis); + /* set methods */ + rsRetVal (*SetAddtlFrameDelim)(strmsrv_t*, int); + rsRetVal (*SetInputName)(strmsrv_t*, uchar*); + rsRetVal (*SetKeepAlive)(strmsrv_t*, int); + rsRetVal (*SetUsrP)(strmsrv_t*, void*); + rsRetVal (*SetCBIsPermittedHost)(strmsrv_t*, int (*) (struct sockaddr *addr, char*, void*, void*)); + rsRetVal (*SetCBOpenLstnSocks)(strmsrv_t *, rsRetVal (*)(strmsrv_t*)); + rsRetVal (*SetCBOnDestruct)(strmsrv_t*, rsRetVal (*) (void*)); + rsRetVal (*SetCBOnRegularClose)(strmsrv_t*, rsRetVal (*) (strms_sess_t*)); + rsRetVal (*SetCBOnErrClose)(strmsrv_t*, rsRetVal (*) (strms_sess_t*)); + rsRetVal (*SetDrvrMode)(strmsrv_t *pThis, int iMode); + rsRetVal (*SetDrvrAuthMode)(strmsrv_t *pThis, uchar *pszMode); + rsRetVal (*SetDrvrPermPeers)(strmsrv_t *pThis, permittedPeers_t*); + /* session specifics */ + rsRetVal (*SetCBOnSessAccept)(strmsrv_t*, rsRetVal (*) (strmsrv_t*, strms_sess_t*)); + rsRetVal (*SetCBOnSessDestruct)(strmsrv_t*, rsRetVal (*) (void*)); + rsRetVal (*SetCBOnSessConstructFinalize)(strmsrv_t*, rsRetVal (*) (void*)); + rsRetVal (*SetSessMax)(strmsrv_t *pThis, int iMaxSess); + rsRetVal (*SetOnCharRcvd)(strmsrv_t *pThis, rsRetVal (*OnMsgCharRcvd)(strms_sess_t*, uchar)); +ENDinterface(strmsrv) +#define strmsrvCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */ +/* change for v?: + */ + + +/* prototypes */ +PROTOTYPEObj(strmsrv); + +/* the name of our library binary */ +#define LM_STRMSRV_FILENAME "lmstrmsrv" + +#endif /* #ifndef INCLUDED_STRMSRV_H */ diff --git a/runtime/sysvar.c b/runtime/sysvar.c index 5eec8f67..c102d1f5 100644 --- a/runtime/sysvar.c +++ b/runtime/sysvar.c @@ -84,7 +84,7 @@ getNOW(eNOWType eNow, cstr_t **ppStr) uchar szBuf[16]; struct syslogTime t; - datetime.getCurrTime(&t); + datetime.getCurrTime(&t, NULL); switch(eNow) { case NOW_NOW: snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "%4.4d-%2.2d-%2.2d", t.year, t.month, t.day); diff --git a/runtime/unicode-helper.h b/runtime/unicode-helper.h new file mode 100644 index 00000000..36d76a78 --- /dev/null +++ b/runtime/unicode-helper.h @@ -0,0 +1,54 @@ +/* This is the header file for unicode support. + * + * Currently, this is a dummy module. + * The following functions are wrappers which hopefully enable us to move + * from 8-bit chars to unicode with relative ease when we finally attack this + * + * Begun 2009-05-21 RGerhards + * + * Copyright (C) 2009 by Rainer Gerhards and Adiscon GmbH + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ +#ifndef INCLUDED_UNICODE_HELPER_H +#define INCLUDED_UNICODE_HELPER_H + +#include <string.h> + +static inline int ustrcmp(uchar *psz1, uchar *psz2) +{ + return strcmp((char*) psz1, (char*) psz2); +} + +static inline int ustrlen(uchar *psz) +{ + return strlen((char*) psz); +} + +static inline uchar* ustrdup(uchar *psz) +{ + return (uchar*) strdup((char*)psz); +} + + +#define UCHAR_CONSTANT(x) ((uchar*) (x)) + +#endif /* multi-include protection */ +/* vim:set ai: + */ diff --git a/runtime/vm.c b/runtime/vm.c index f1fbcc2e..8cbf9e12 100644 --- a/runtime/vm.c +++ b/runtime/vm.c @@ -27,6 +27,7 @@ #include <stdlib.h> #include <string.h> #include <assert.h> +#include <ctype.h> #include "rsyslog.h" #include "obj.h" @@ -40,6 +41,142 @@ DEFobjCurrIf(vmstk) DEFobjCurrIf(var) DEFobjCurrIf(sysvar) +/* ------------------------------ function registry code and structures ------------------------------ */ + +/* we maintain a registry of known functions */ +/* currently, this is a singly-linked list, this shall become a binary + * tree when we add the real call interface. So far, entries are added + * at the root, only. + */ +typedef struct s_rsf_entry { + cstr_t *pName; /* function name */ + prsf_t rsf; /* pointer to function code */ + struct s_rsf_entry *pNext; /* Pointer to next element or NULL */ +} rsf_entry_t; +rsf_entry_t *funcRegRoot = NULL; + + +/* add a function to the function registry. + * The handed-over cstr_t* object must no longer be used by the caller. + * A duplicate function name is an error. + * rgerhards, 2009-04-06 + */ +static rsRetVal +rsfrAddFunction(uchar *szName, prsf_t rsf) +{ + rsf_entry_t *pEntry; + size_t lenName; + DEFiRet; + + assert(szName != NULL); + assert(rsf != NULL); + + /* first check if we have a duplicate name, with the current approach this means + * we need to go through the whole list. + */ + lenName = strlen((char*)szName); + for(pEntry = funcRegRoot ; pEntry != NULL ; pEntry = pEntry->pNext) + if(!rsCStrSzStrCmp(pEntry->pName, szName, lenName)) + ABORT_FINALIZE(RS_RET_DUP_FUNC_NAME); + + /* unique name, so add to head of list */ + CHKmalloc(pEntry = calloc(1, sizeof(rsf_entry_t))); + CHKiRet(rsCStrConstructFromszStr(&pEntry->pName, szName)); + pEntry->rsf = rsf; + pEntry->pNext = funcRegRoot; + funcRegRoot = pEntry; + +finalize_it: + RETiRet; +} + + +/* find a function inside the function registry + * The caller provides a cstr_t with the function name and receives + * a function pointer back. If no function is found, an RS_RET_UNKNW_FUNC + * error is returned. So if the function returns with RS_RET_OK, the caller + * can savely assume the function pointer is valid. + * rgerhards, 2009-04-06 + */ +static rsRetVal +findRSFunction(cstr_t *pcsName, prsf_t *prsf) +{ + rsf_entry_t *pEntry; + rsf_entry_t *pFound; + DEFiRet; + + assert(prsf != NULL); + + /* find function by list walkthrough. */ + pFound = NULL; + for(pEntry = funcRegRoot ; pEntry != NULL && pFound == NULL ; pEntry = pEntry->pNext) + if(!rsCStrCStrCmp(pEntry->pName, pcsName)) + pFound = pEntry; + + if(pFound == NULL) + ABORT_FINALIZE(RS_RET_UNKNW_FUNC); + + *prsf = pFound->rsf; + +finalize_it: + RETiRet; +} + + +/* find the name of a RainerScript function whom's function pointer + * is known. This function returns the cstr_t object, which MUST NOT + * be modified by the caller. + * rgerhards, 2009-04-06 + */ +static rsRetVal +findRSFunctionName(prsf_t rsf, cstr_t **ppcsName) +{ + rsf_entry_t *pEntry; + rsf_entry_t *pFound; + DEFiRet; + + assert(rsf != NULL); + assert(ppcsName != NULL); + + /* find function by list walkthrough. */ + pFound = NULL; + for(pEntry = funcRegRoot ; pEntry != NULL && pFound == NULL ; pEntry = pEntry->pNext) + if(pEntry->rsf == rsf) + pFound = pEntry; + + if(pFound == NULL) + ABORT_FINALIZE(RS_RET_UNKNW_FUNC); + + *ppcsName = pFound->pName; + +finalize_it: + RETiRet; +} + + +/* free the whole function registry + */ +static void +rsfrRemoveAll(void) +{ + rsf_entry_t *pEntry; + rsf_entry_t *pEntryDel; + + BEGINfunc + pEntry = funcRegRoot; + while(pEntry != NULL) { + pEntryDel = pEntry; + pEntry = pEntry->pNext; + rsCStrDestruct(&pEntryDel->pName); + free(pEntryDel); + } + funcRegRoot = NULL; + ENDfunc +} + + +/* ------------------------------ end function registry code and structures ------------------------------ */ + /* ------------------------------ instruction set implementation ------------------------------ * * The following functions implement the VM's instruction set. @@ -331,10 +468,109 @@ CODESTARTop(PUSHSYSVAR) finalize_it: ENDop(PUSHSYSVAR) +/* The function call operation is only very roughly implemented. While the plumbing + * to reach this instruction is fine, the instruction itself currently supports only + * functions with a single argument AND with a name that we know. + * TODO: later, we can add here the real logic, that involves looking up function + * names, loading them dynamically ... and all that... + * implementation begun 2009-03-10 by rgerhards + */ +BEGINop(FUNC_CALL) /* remember to set the instruction also in the ENDop macro! */ + var_t *numOperands; +CODESTARTop(FUNC_CALL) + vmstk.PopNumber(pThis->pStk, &numOperands); + CHKiRet((*pOp->operand.rsf)(pThis->pStk, numOperands->val.num)); + var.Destruct(&numOperands); /* no longer needed */ +finalize_it: +ENDop(FUNC_CALL) + /* ------------------------------ end instruction set implementation ------------------------------ */ +/* ------------------------------ begin built-in function implementation ------------------------------ */ +/* note: this shall probably be moved to a separate module, but for the time being we do it directly + * in here. This is on our way to get from a dirty to a clean solution via baby steps that are + * a bit less dirty each time... + * + * The advantage of doing it here is that we do not yet need to think about how to handle the + * exit case, where we must not unload function modules which functions are still referenced. + * + * CALLING INTERFACE: + * The function must pop its parameters off the stack and pop its result onto + * the stack when it is finished. The number of parameters the function was + * called with is provided to it. If the argument count is less then what the function + * expected, it may handle the situation with defaults (or return an error). If the + * argument count is greater than expected, returnung an error is highly + * recommended (use RS_RET_INVLD_NBR_ARGUMENTS for these cases). + * + * All function names are prefixed with "rsf_" (RainerScript Function) to have + * a separate "name space". + * + * rgerhards, 2009-04-06 + */ + + +/* The strlen function, also probably a prototype of how all functions should be + * implemented. + * rgerhards, 2009-04-06 + */ +static rsRetVal +rsf_strlen(vmstk_t *pStk, int numOperands) +{ + DEFiRet; + var_t *operand1; + int iStrlen; + + if(numOperands != 1) + ABORT_FINALIZE(RS_RET_INVLD_NBR_ARGUMENTS); + + /* pop args and do operaton (trivial case here...) */ + vmstk.PopString(pStk, &operand1); + iStrlen = strlen((char*) rsCStrGetSzStr(operand1->val.pStr)); + + /* Store result and cleanup */ + var.SetNumber(operand1, iStrlen); + vmstk.Push(pStk, operand1); +finalize_it: + RETiRet; +} + + +/* The "tolower" function, which converts its sole argument to lower case. + * Quite honestly, currently this is primarily a test driver for me... + * rgerhards, 2009-04-06 + */ +static rsRetVal +rsf_tolower(vmstk_t *pStk, int numOperands) +{ + DEFiRet; + var_t *operand1; + uchar *pSrc; + cstr_t *pcstr; + int iStrlen; + + if(numOperands != 1) + ABORT_FINALIZE(RS_RET_INVLD_NBR_ARGUMENTS); + + /* pop args and do operaton */ + CHKiRet(rsCStrConstruct(&pcstr)); + vmstk.PopString(pStk, &operand1); + pSrc = rsCStrGetSzStr(operand1->val.pStr); + iStrlen = strlen((char*)pSrc); + while(iStrlen--) { + CHKiRet(rsCStrAppendChar(pcstr, tolower(*pSrc++))); + } + + /* Store result and cleanup */ + CHKiRet(cstrFinalize(pcstr)); + var.SetString(operand1, pcstr); + vmstk.Push(pStk, operand1); +finalize_it: + RETiRet; +} + + /* Standard-Constructor */ BEGINobjConstruct(vm) /* be sure to specify the object type also in END macro! */ @@ -413,6 +649,7 @@ execProg(vm_t *pThis, vmprg_t *pProg) doOP(DIV); doOP(MOD); doOP(UNARY_MINUS); + doOP(FUNC_CALL); default: ABORT_FINALIZE(RS_RET_INVALID_VMOP); dbgoprint((obj_t*) pThis, "invalid instruction %d in vmprg\n", pCurrOp->opcode); @@ -503,10 +740,23 @@ CODESTARTobjQueryInterface(vm) pIf->PopBoolFromStack = PopBoolFromStack; pIf->PopVarFromStack = PopVarFromStack; pIf->SetMsg = SetMsg; + pIf->FindRSFunction = findRSFunction; + pIf->FindRSFunctionName = findRSFunctionName; finalize_it: ENDobjQueryInterface(vm) +/* Exit the vm class. + * rgerhards, 2009-04-06 + */ +BEGINObjClassExit(vm, OBJ_IS_CORE_MODULE) /* class, version */ + rsfrRemoveAll(); + objRelease(sysvar, CORE_COMPONENT); + objRelease(var, CORE_COMPONENT); + objRelease(vmstk, CORE_COMPONENT); +ENDObjClassExit(vm) + + /* Initialize the vm class. Must be called as the very first method * before anything else is called inside this class. * rgerhards, 2008-02-19 @@ -520,6 +770,11 @@ BEGINObjClassInit(vm, 1, OBJ_IS_CORE_MODULE) /* class, version */ /* set our own handlers */ OBJSetMethodHandler(objMethod_DEBUGPRINT, vmDebugPrint); OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, vmConstructFinalize); + + /* register built-in functions // TODO: move to its own module */ + CHKiRet(rsfrAddFunction((uchar*)"strlen", rsf_strlen)); + CHKiRet(rsfrAddFunction((uchar*)"tolower", rsf_tolower)); + ENDObjClassInit(vm) /* vi:set ai: diff --git a/runtime/vm.h b/runtime/vm.h index d2458220..cb3c69d0 100644 --- a/runtime/vm.h +++ b/runtime/vm.h @@ -55,8 +55,11 @@ BEGINinterface(vm) /* name must also be changed in ENDinterface macro! */ rsRetVal (*PopBoolFromStack)(vm_t *pThis, var_t **ppVar); /* there are a few cases where we need this... */ rsRetVal (*PopVarFromStack)(vm_t *pThis, var_t **ppVar); /* there are a few cases where we need this... */ rsRetVal (*SetMsg)(vm_t *pThis, msg_t *pMsg); /* there are a few cases where we need this... */ + /* v2 (4.1.7) */ + rsRetVal (*FindRSFunction)(cstr_t *pcsName, prsf_t *prsf); /* 2009-06-04 */ + rsRetVal (*FindRSFunctionName)(prsf_t rsf, cstr_t **ppcsName); /* 2009-06-04 */ ENDinterface(vm) -#define vmCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */ +#define vmCURR_IF_VERSION 2 /* increment whenever you change the interface structure! */ /* prototypes */ diff --git a/runtime/vmop.c b/runtime/vmop.c index fcacb15b..acacfc9e 100644 --- a/runtime/vmop.c +++ b/runtime/vmop.c @@ -32,10 +32,12 @@ #include "rsyslog.h" #include "obj.h" #include "vmop.h" +#include "vm.h" /* static data */ DEFobjStaticHelpers DEFobjCurrIf(var) +DEFobjCurrIf(vm) /* forward definitions */ @@ -61,9 +63,7 @@ rsRetVal vmopConstructFinalize(vmop_t __attribute__((unused)) *pThis) /* destructor for the vmop object */ BEGINobjDestruct(vmop) /* be sure to specify the object type also in END and CODESTART macros! */ CODESTARTobjDestruct(vmop) - if( pThis->opcode == opcode_PUSHSYSVAR - || pThis->opcode == opcode_PUSHMSGVAR - || pThis->opcode == opcode_PUSHCONSTANT) { + if(pThis->opcode != opcode_FUNC_CALL) { if(pThis->operand.pVar != NULL) var.Destruct(&pThis->operand.pVar); } @@ -73,12 +73,23 @@ ENDobjDestruct(vmop) /* DebugPrint support for the vmop object */ BEGINobjDebugPrint(vmop) /* be sure to specify the object type also in END and CODESTART macros! */ uchar *pOpcodeName; + cstr_t *pStrVar; CODESTARTobjDebugPrint(vmop) vmopOpcode2Str(pThis, &pOpcodeName); - dbgoprint((obj_t*) pThis, "opcode: %d\t(%s), next %p, var in next line\n", (int) pThis->opcode, pOpcodeName, - pThis->pNext); - if(pThis->operand.pVar != NULL) - var.DebugPrint(pThis->operand.pVar); + if(pThis->opcode == opcode_FUNC_CALL) { + CHKiRet(vm.FindRSFunctionName(pThis->operand.rsf, &pStrVar)); + assert(pStrVar != NULL); + } else { + CHKiRet(rsCStrConstruct(&pStrVar)); + if(pThis->operand.pVar != NULL) { + CHKiRet(var.Obj2Str(pThis->operand.pVar, pStrVar)); + } + } + CHKiRet(cstrFinalize(pStrVar)); + dbgoprint((obj_t*) pThis, "%.12s\t%s\n", pOpcodeName, rsCStrGetSzStrNoNULL(pStrVar)); + if(pThis->opcode != opcode_FUNC_CALL) + rsCStrDestruct(&pStrVar); +finalize_it: ENDobjDebugPrint(vmop) @@ -97,6 +108,7 @@ static rsRetVal Obj2Str(vmop_t *pThis, cstr_t *pstrPrg) { uchar *pOpcodeName; + cstr_t *pcsFuncName; uchar szBuf[2048]; size_t lenBuf; DEFiRet; @@ -106,8 +118,13 @@ Obj2Str(vmop_t *pThis, cstr_t *pstrPrg) vmopOpcode2Str(pThis, &pOpcodeName); lenBuf = snprintf((char*) szBuf, sizeof(szBuf), "%s\t", pOpcodeName); CHKiRet(rsCStrAppendStrWithLen(pstrPrg, szBuf, lenBuf)); - if(pThis->operand.pVar != NULL) - CHKiRet(var.Obj2Str(pThis->operand.pVar, pstrPrg)); + if(pThis->opcode == opcode_FUNC_CALL) { + CHKiRet(vm.FindRSFunctionName(pThis->operand.rsf, &pcsFuncName)); + CHKiRet(rsCStrAppendCStr(pstrPrg, pcsFuncName)); + } else { + if(pThis->operand.pVar != NULL) + CHKiRet(var.Obj2Str(pThis->operand.pVar, pstrPrg)); + } CHKiRet(rsCStrAppendChar(pstrPrg, '\n')); finalize_it: @@ -115,6 +132,23 @@ finalize_it: } +/* set function + * rgerhards, 2009-04-06 + */ +static rsRetVal +vmopSetFunc(vmop_t *pThis, cstr_t *pcsFuncName) +{ + prsf_t rsf; /* pointer to function */ + DEFiRet; + ISOBJ_TYPE_assert(pThis, vmop); + CHKiRet(vm.FindRSFunction(pcsFuncName, &rsf)); /* check if function exists and obtain pointer to it */ + assert(rsf != NULL); /* just double-check, would be very hard to find! */ + pThis->operand.rsf = rsf; +finalize_it: + RETiRet; +} + + /* set operand (variant case) * rgerhards, 2008-02-20 */ @@ -158,37 +192,37 @@ vmopOpcode2Str(vmop_t *pThis, uchar **ppName) *ppName = (uchar*) "and"; break; case opcode_PLUS: - *ppName = (uchar*) "+"; + *ppName = (uchar*) "add"; break; case opcode_MINUS: - *ppName = (uchar*) "-"; + *ppName = (uchar*) "sub"; break; case opcode_TIMES: - *ppName = (uchar*) "*"; + *ppName = (uchar*) "mul"; break; case opcode_DIV: - *ppName = (uchar*) "/"; + *ppName = (uchar*) "div"; break; case opcode_MOD: - *ppName = (uchar*) "%"; + *ppName = (uchar*) "mod"; break; case opcode_NOT: *ppName = (uchar*) "not"; break; case opcode_CMP_EQ: - *ppName = (uchar*) "=="; + *ppName = (uchar*) "cmp_=="; break; case opcode_CMP_NEQ: - *ppName = (uchar*) "!="; + *ppName = (uchar*) "cmp_!="; break; case opcode_CMP_LT: - *ppName = (uchar*) "<"; + *ppName = (uchar*) "cmp_<"; break; case opcode_CMP_GT: - *ppName = (uchar*) ">"; + *ppName = (uchar*) "cmp_>"; break; case opcode_CMP_LTEQ: - *ppName = (uchar*) "<="; + *ppName = (uchar*) "cmp_<="; break; case opcode_CMP_CONTAINS: *ppName = (uchar*) "contains"; @@ -197,28 +231,31 @@ vmopOpcode2Str(vmop_t *pThis, uchar **ppName) *ppName = (uchar*) "startswith"; break; case opcode_CMP_GTEQ: - *ppName = (uchar*) ">="; + *ppName = (uchar*) "cmp_>="; break; case opcode_PUSHSYSVAR: - *ppName = (uchar*) "PUSHSYSVAR"; + *ppName = (uchar*) "push_sysvar"; break; case opcode_PUSHMSGVAR: - *ppName = (uchar*) "PUSHMSGVAR"; + *ppName = (uchar*) "push_msgvar"; break; case opcode_PUSHCONSTANT: - *ppName = (uchar*) "PUSHCONSTANT"; + *ppName = (uchar*) "push_const"; break; case opcode_POP: - *ppName = (uchar*) "POP"; + *ppName = (uchar*) "pop"; break; case opcode_UNARY_MINUS: - *ppName = (uchar*) "UNARY_MINUS"; + *ppName = (uchar*) "unary_minus"; break; case opcode_STRADD: - *ppName = (uchar*) "STRADD"; + *ppName = (uchar*) "strconcat"; + break; + case opcode_FUNC_CALL: + *ppName = (uchar*) "func_call"; break; default: - *ppName = (uchar*) "INVALID opcode"; + *ppName = (uchar*) "!invalid_opcode!"; break; } @@ -244,6 +281,7 @@ CODESTARTobjQueryInterface(vmop) pIf->ConstructFinalize = vmopConstructFinalize; pIf->Destruct = vmopDestruct; pIf->DebugPrint = vmopDebugPrint; + pIf->SetFunc = vmopSetFunc; pIf->SetOpcode = vmopSetOpcode; pIf->SetVar = vmopSetVar; pIf->Opcode2Str = vmopOpcode2Str; @@ -259,6 +297,7 @@ ENDobjQueryInterface(vmop) BEGINObjClassInit(vmop, 1, OBJ_IS_CORE_MODULE) /* class, version */ /* request objects we use */ CHKiRet(objUse(var, CORE_COMPONENT)); + CHKiRet(objUse(vm, CORE_COMPONENT)); OBJSetMethodHandler(objMethod_DEBUGPRINT, vmopDebugPrint); OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, vmopConstructFinalize); diff --git a/runtime/vmop.h b/runtime/vmop.h index c3d5d5f4..67048c26 100644 --- a/runtime/vmop.h +++ b/runtime/vmop.h @@ -26,6 +26,7 @@ #define INCLUDED_VMOP_H #include "ctok_token.h" +#include "vmstk.h" #include "stringbuf.h" /* machine instructions types */ @@ -59,17 +60,45 @@ typedef enum { /* do NOT start at 0 to detect uninitialized types after calloc( opcode_PUSHMSGVAR = 1002, /* requires var operand */ opcode_PUSHCONSTANT = 1003, /* requires var operand */ opcode_UNARY_MINUS = 1010, - opcode_END_PROG = 1011 + opcode_FUNC_CALL = 1012, + opcode_END_PROG = 2000 } opcode_t; +/* Additional doc, operation specific + + FUNC_CALL + All parameter passing is via the stack. Parameters are placed onto the stack in reverse order, + that means the last parameter is on top of the stack, the first at the bottom location. + At the actual top of the stack is the number of parameters. This permits functions to be + called with variable number of arguments. The function itself is responsible for poping + the right number of parameters of the stack and complaining if the number is incorrect. + On exit, a single return value must be pushed onto the stack. The FUNC_CALL operation + is generic. Its pVar argument contains the function name string (TODO: very slow, make + faster in later releases). + + Sample Function call: sampleFunc(p1, p2, p3) ; returns number 4711 (sample) + Stacklayout on entry (order is top to bottom): + 3 + p3 + p2 + p1 + ... other vars ... + + Stack on exit + 4711 + ... other vars ... + + */ + + /* the vmop object */ typedef struct vmop_s { BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */ opcode_t opcode; union { var_t *pVar; - /* TODO: add function pointer */ + prsf_t rsf; /* pointer to function for "call" instruction */ } operand; struct vmop_s *pNext; /* next operation or NULL, if end of program (logically this belongs to vmprg) */ } vmop_t; @@ -85,8 +114,13 @@ BEGINinterface(vmop) /* name must also be changed in ENDinterface macro! */ rsRetVal (*SetVar)(vmop_t *pThis, var_t *pVar); rsRetVal (*Opcode2Str)(vmop_t *pThis, uchar **ppName); rsRetVal (*Obj2Str)(vmop_t *pThis, cstr_t *pstr); + /* v2 */ + rsRetVal (*SetFunc)(vmop_t *pThis, cstr_t *pcsFuncName); ENDinterface(vmop) -#define vmopCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */ +#define vmopCURR_IF_VERSION 2 /* increment whenever you change the interface structure! */ +/* interface changes, v1 -> v2 + * added SetFuct after existing function pointers -- rgerhards, 2009-04-06 + */ /* the remaining prototypes */ PROTOTYPEObj(vmop); diff --git a/runtime/vmprg.c b/runtime/vmprg.c index 705e6948..07757b98 100644 --- a/runtime/vmprg.c +++ b/runtime/vmprg.c @@ -74,7 +74,7 @@ ENDobjDestruct(vmprg) BEGINobjDebugPrint(vmprg) /* be sure to specify the object type also in END and CODESTART macros! */ vmop_t *pOp; CODESTARTobjDebugPrint(vmprg) - dbgoprint((obj_t*) pThis, "program contents:\n"); + dbgoprint((obj_t*) pThis, "VM Program:\n"); for(pOp = pThis->vmopRoot ; pOp != NULL ; pOp = pOp->pNext) { vmop.DebugPrint(pOp); } @@ -155,7 +155,6 @@ vmprgAddVarOperation(vmprg_t *pThis, opcode_t opcode, var_t *pVar) /* construct and fill vmop */ CHKiRet(vmop.Construct(&pOp)); CHKiRet(vmop.ConstructFinalize(pOp)); - CHKiRet(vmop.ConstructFinalize(pOp)); CHKiRet(vmop.SetOpcode(pOp, opcode)); if(pVar != NULL) CHKiRet(vmop.SetVar(pOp, pVar)); @@ -168,6 +167,32 @@ finalize_it: } +/* this is another shortcut for high-level callers. It is similar to vmprgAddVarOperation + * but adds a call operation. Among others, this include a check if the function + * is known. + */ +static rsRetVal +vmprgAddCallOperation(vmprg_t *pThis, cstr_t *pcsName) +{ + DEFiRet; + vmop_t *pOp; + + ISOBJ_TYPE_assert(pThis, vmprg); + + /* construct and fill vmop */ + CHKiRet(vmop.Construct(&pOp)); + CHKiRet(vmop.ConstructFinalize(pOp)); + CHKiRet(vmop.SetFunc(pOp, pcsName)); + CHKiRet(vmop.SetOpcode(pOp, opcode_FUNC_CALL)); + + /* and add it to the program */ + CHKiRet(vmprgAddOperation(pThis, pOp)); + +finalize_it: + RETiRet; +} + + /* queryInterface function * rgerhards, 2008-02-21 */ @@ -189,6 +214,7 @@ CODESTARTobjQueryInterface(vmprg) pIf->Obj2Str = Obj2Str; pIf->AddOperation = vmprgAddOperation; pIf->AddVarOperation = vmprgAddVarOperation; + pIf->AddCallOperation = vmprgAddCallOperation; finalize_it: ENDobjQueryInterface(vmprg) diff --git a/runtime/vmprg.h b/runtime/vmprg.h index c1042f7d..66f03913 100644 --- a/runtime/vmprg.h +++ b/runtime/vmprg.h @@ -57,8 +57,10 @@ BEGINinterface(vmprg) /* name must also be changed in ENDinterface macro! */ rsRetVal (*AddOperation)(vmprg_t *pThis, vmop_t *pOp); rsRetVal (*AddVarOperation)(vmprg_t *pThis, opcode_t opcode, var_t *pVar); rsRetVal (*Obj2Str)(vmprg_t *pThis, cstr_t *pstr); + /* v2 (4.1.7) */ + rsRetVal (*AddCallOperation)(vmprg_t *pThis, cstr_t *pVar); /* added 2009-04-06 */ ENDinterface(vmprg) -#define vmprgCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */ +#define vmprgCURR_IF_VERSION 2 /* increment whenever you change the interface structure! */ /* prototypes */ diff --git a/runtime/vmstk.h b/runtime/vmstk.h index 2d45ee4d..06657cf4 100644 --- a/runtime/vmstk.h +++ b/runtime/vmstk.h @@ -27,11 +27,11 @@ #define VMSTK_SIZE 256 /* the vmstk object */ -typedef struct vmstk_s { +struct vmstk_s { BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */ var_t *vStk[VMSTK_SIZE];/* the actual stack */ int iStkPtr; /* stack pointer, points to next free location, grows from 0 --> topend */ -} vmstk_t; +}; /* interfaces */ diff --git a/runtime/wti.c b/runtime/wti.c index 343a7227..544bffa7 100644 --- a/runtime/wti.c +++ b/runtime/wti.c @@ -39,15 +39,22 @@ #include <pthread.h> #include <errno.h> +#ifdef OS_SOLARIS +# include <sched.h> +# define pthread_yield() sched_yield() +#endif + #include "rsyslog.h" #include "stringbuf.h" #include "srUtils.h" #include "wtp.h" #include "wti.h" #include "obj.h" +#include "glbl.h" /* static data */ DEFobjStaticHelpers +DEFobjCurrIf(glbl) /* forward-definitions */ @@ -202,6 +209,7 @@ ENDobjDestruct(wti) /* Standard-Constructor for the wti object */ BEGINobjConstruct(wti) /* be sure to specify the object type also in END macro! */ + pThis->bOptimizeUniProc = glbl.GetOptimizeUniProc(); pthread_cond_init(&pThis->condExitDone, NULL); pthread_mutex_init(&pThis->mut, NULL); ENDobjConstruct(wti) @@ -303,7 +311,7 @@ wtiWorkerCancelCleanup(void *arg) pWtp = pThis->pWtp; ISOBJ_TYPE_assert(pWtp, wtp); - dbgprintf("%s: cancelation cleanup handler called.\n", wtiGetDbgHdr(pThis)); + DBGPRINTF("%s: cancelation cleanup handler called.\n", wtiGetDbgHdr(pThis)); /* call user supplied handler (that one e.g. requeues the element) */ pWtp->pfOnWorkerCancel(pThis->pWtp->pUsr, pThis->pUsrp); @@ -371,7 +379,8 @@ wtiWorker(wti_t *pThis) wtpProcessThrdChanges(pWtp); pthread_testcancel(); /* see big comment in function header */ # if !defined(__hpux) /* pthread_yield is missing there! */ - pthread_yield(); /* see big comment in function header */ + if(pThis->bOptimizeUniProc) + pthread_yield(); /* see big comment in function header */ # endif /* if we have a rate-limiter set for this worker pool, let's call it. Please @@ -395,7 +404,7 @@ wtiWorker(wti_t *pThis) /* if we reach this point, we are still protected by the mutex */ if(pWtp->pfIsIdle(pWtp->pUsr, MUTEX_ALREADY_LOCKED)) { - dbgprintf("%s: worker IDLE, waiting for work.\n", wtiGetDbgHdr(pThis)); + DBGPRINTF("%s: worker IDLE, waiting for work.\n", wtiGetDbgHdr(pThis)); pWtp->pfOnIdle(pWtp->pUsr, MUTEX_ALREADY_LOCKED); if(pWtp->toWrkShutdown == -1) { @@ -404,7 +413,7 @@ wtiWorker(wti_t *pThis) } else { timeoutComp(&t, pWtp->toWrkShutdown);/* get absolute timeout */ if(d_pthread_cond_timedwait(pWtp->pcondBusy, pWtp->pmutUsr, &t) != 0) { - dbgprintf("%s: inactivity timeout, worker terminating...\n", wtiGetDbgHdr(pThis)); + DBGPRINTF("%s: inactivity timeout, worker terminating...\n", wtiGetDbgHdr(pThis)); bInactivityTOOccured = 1; /* indicate we had a timeout */ } } @@ -470,6 +479,14 @@ finalize_it: /* dummy */ rsRetVal wtiQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; } +/* exit our class + */ +BEGINObjClassExit(wti, OBJ_IS_CORE_MODULE) /* CHANGE class also in END MACRO! */ +CODESTARTObjClassExit(nsdsel_gtls) + /* release objects we no longer need */ + objRelease(glbl, CORE_COMPONENT); +ENDObjClassExit(wti) + /* Initialize the wti class. Must be called as the very first method * before anything else is called inside this class. @@ -477,6 +494,7 @@ rsRetVal wtiQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; } */ BEGINObjClassInit(wti, 1, OBJ_IS_CORE_MODULE) /* one is the object version (most important for persisting) */ /* request objects we use */ + CHKiRet(objUse(glbl, CORE_COMPONENT)); ENDObjClassInit(wti) /* diff --git a/runtime/wti.h b/runtime/wti.h index 0cd6744e..6b60b833 100644 --- a/runtime/wti.h +++ b/runtime/wti.h @@ -31,6 +31,7 @@ /* the worker thread instance class */ typedef struct wti_s { BEGINobjInstance; + int bOptimizeUniProc; /* cache for the equally-named global setting, pulled at time of queue creation */ pthread_t thrdID; /* thread ID */ qWrkCmd_t tCurrCmd; /* current command to be carried out by worker */ obj_t *pUsrp; /* pointer to an object meaningful for current user pointer (e.g. queue pUsr data elemt) */ diff --git a/runtime/wtp.c b/runtime/wtp.c index 3beae271..04eb974f 100644 --- a/runtime/wtp.c +++ b/runtime/wtp.c @@ -40,15 +40,22 @@ #include <unistd.h> #include <errno.h> +#ifdef OS_SOLARIS +# include <sched.h> +# define pthread_yield() sched_yield() +#endif + #include "rsyslog.h" #include "stringbuf.h" #include "srUtils.h" #include "wtp.h" #include "wti.h" #include "obj.h" +#include "glbl.h" /* static data */ DEFobjStaticHelpers +DEFobjCurrIf(glbl) /* forward-definitions */ @@ -75,6 +82,7 @@ static rsRetVal NotImplementedDummy() { return RS_RET_OK; } /* Standard-Constructor for the wtp object */ BEGINobjConstruct(wtp) /* be sure to specify the object type also in END macro! */ + pThis->bOptimizeUniProc = glbl.GetOptimizeUniProc(); pthread_mutex_init(&pThis->mut, NULL); pthread_mutex_init(&pThis->mutThrdShutdwn, NULL); pthread_cond_init(&pThis->condThrdTrm, NULL); @@ -478,7 +486,7 @@ wtpStartWrkr(wtp_t *pThis, int bLockMutex) ISOBJ_TYPE_assert(pThis, wtp); - wtpProcessThrdChanges(pThis); + wtpProcessThrdChanges(pThis); // TODO: Performance: this causes a lot of FUTEX calls BEGIN_MTX_PROTECTED_OPERATIONS(&pThis->mut, bLockMutex); @@ -506,7 +514,8 @@ wtpStartWrkr(wtp_t *pThis, int bLockMutex) * hold the queue's mutex, but at least it has a chance to start on a single-CPU system. */ # if !defined(__hpux) /* pthread_yield is missing there! */ - pthread_yield(); + if(pThis->bOptimizeUniProc) + pthread_yield(); # endif /* indicate we just started a worker and would like to see it running */ @@ -639,12 +648,22 @@ finalize_it: /* dummy */ rsRetVal wtpQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; } +/* exit our class + */ +BEGINObjClassExit(wtp, OBJ_IS_CORE_MODULE) /* CHANGE class also in END MACRO! */ +CODESTARTObjClassExit(nsdsel_gtls) + /* release objects we no longer need */ + objRelease(glbl, CORE_COMPONENT); +ENDObjClassExit(wtp) + + /* Initialize the stream class. Must be called as the very first method * before anything else is called inside this class. * rgerhards, 2008-01-09 */ BEGINObjClassInit(wtp, 1, OBJ_IS_CORE_MODULE) /* request objects we use */ + CHKiRet(objUse(glbl, CORE_COMPONENT)); ENDObjClassInit(wtp) /* diff --git a/runtime/wtp.h b/runtime/wtp.h index 0f21ac11..b9cb07c5 100644 --- a/runtime/wtp.h +++ b/runtime/wtp.h @@ -52,6 +52,7 @@ typedef enum { /* the worker thread pool (wtp) object */ typedef struct wtp_s { BEGINobjInstance; + int bOptimizeUniProc; /* cache for the equally-named global setting, pulled at time of queue creation */ wtpState_t wtpState; int iNumWorkerThreads;/* number of worker threads to use */ int iCurNumWrkThrd;/* current number of active worker threads */ |