summaryrefslogtreecommitdiffstats
path: root/runtime
diff options
context:
space:
mode:
Diffstat (limited to 'runtime')
-rw-r--r--runtime/Makefile.am1
-rw-r--r--runtime/atomic.h86
-rw-r--r--runtime/cfsysline.c11
-rw-r--r--runtime/debug.c25
-rw-r--r--runtime/debug.h5
-rw-r--r--runtime/glbl.c38
-rw-r--r--runtime/glbl.h11
-rw-r--r--runtime/msg.c15
-rw-r--r--runtime/msg.h4
-rw-r--r--runtime/net.c8
-rw-r--r--runtime/nsdsel_ptcp.c51
-rw-r--r--runtime/nsdsel_ptcp.h5
-rw-r--r--runtime/parser.c35
-rw-r--r--runtime/prop.c6
-rw-r--r--runtime/prop.h2
-rw-r--r--runtime/queue.c10
-rw-r--r--runtime/queue.h1
-rw-r--r--runtime/rsyslog.c2
-rw-r--r--runtime/rsyslog.h27
-rw-r--r--runtime/stream.c3
-rw-r--r--runtime/unlimited_select.h45
-rw-r--r--runtime/vm.c44
-rw-r--r--runtime/wti.c11
-rw-r--r--runtime/wti.h1
-rw-r--r--runtime/wtp.c24
-rw-r--r--runtime/wtp.h3
26 files changed, 392 insertions, 82 deletions
diff --git a/runtime/Makefile.am b/runtime/Makefile.am
index 14abe722..c1a15198 100644
--- a/runtime/Makefile.am
+++ b/runtime/Makefile.am
@@ -15,6 +15,7 @@ librsyslog_la_SOURCES = \
nsd.h \
glbl.h \
glbl.c \
+ unlimited_select.h \
conf.c \
conf.h \
parser.h \
diff --git a/runtime/atomic.h b/runtime/atomic.h
index d5aaf56b..fc3e0b2d 100644
--- a/runtime/atomic.h
+++ b/runtime/atomic.h
@@ -31,8 +31,6 @@
* 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" /* autotools! */
-
#ifndef INCLUDED_ATOMIC_H
#define INCLUDED_ATOMIC_H
@@ -41,17 +39,28 @@
* They simply came in too late. -- rgerhards, 2008-04-02
*/
#ifdef HAVE_ATOMIC_BUILTINS
-# define ATOMIC_INC(data) ((void) __sync_fetch_and_add(&(data), 1))
+# define ATOMIC_INC(data, phlpmut) ((void) __sync_fetch_and_add(data, 1))
# define ATOMIC_INC_AND_FETCH(data) __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_DEC(data, phlpmut) ((void) __sync_sub_and_fetch(data, 1))
+# define ATOMIC_DEC_AND_FETCH(data, phlpmut) __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)
-# define ATOMIC_STORE_0_TO_INT(data) __sync_fetch_and_and(&(data), 0)
-# define ATOMIC_STORE_1_TO_INT(data) __sync_fetch_and_or(&(data), 1)
+# define ATOMIC_STORE_0_TO_INT(data, phlpmut) __sync_fetch_and_and(data, 0)
+# define ATOMIC_STORE_1_TO_INT(data, phlpmut) __sync_fetch_and_or(data, 1)
# define ATOMIC_STORE_INT_TO_INT(data, val) __sync_fetch_and_or(&(data), (val))
# define ATOMIC_CAS(data, oldVal, newVal) __sync_bool_compare_and_swap(&(data), (oldVal), (newVal));
-# define ATOMIC_CAS_VAL(data, oldVal, newVal) __sync_val_compare_and_swap(&(data), (oldVal), (newVal));
+# define ATOMIC_CAS_VAL(data, oldVal, newVal, phlpmut) __sync_val_compare_and_swap(data, (oldVal), (newVal));
+
+ /* functions below are not needed if we have atomics */
+# define DEF_ATOMIC_HELPER_MUT(x)
+# define INIT_ATOMIC_HELPER_MUT(x)
+# define DESTROY_ATOMIC_HELPER_MUT(x)
+
+ /* the following operations should preferrably be done atomic, but it is
+ * not fatal if not -- that means we can live with some missed updates. So be
+ * sure to use these macros only if that really does not matter!
+ */
+# define PREFER_ATOMIC_INC(data) ((void) __sync_fetch_and_add(&(data), 1))
#else
/* note that we gained parctical proof that theoretical problems DO occur
* if we do not properly address them. See this blog post for details:
@@ -60,12 +69,63 @@
* simply go ahead and do without them - use mutexes or other things. The
* code needs to be checked against all those cases. -- rgerhards, 2009-01-30
*/
+ #include <pthread.h>
+# define ATOMIC_INC(data, phlpmut) { \
+ pthread_mutex_lock(phlpmut); \
+ ++(*(data)); \
+ pthread_mutex_unlock(phlpmut); \
+ }
+
+# define ATOMIC_STORE_0_TO_INT(data, hlpmut) { \
+ pthread_mutex_lock(&hlpmut); \
+ *(data) = 0; \
+ pthread_mutex_unlock(&hlpmut); \
+ }
+
+# define ATOMIC_STORE_1_TO_INT(data, hlpmut) { \
+ pthread_mutex_lock(&hlpmut); \
+ *(data) = 1; \
+ pthread_mutex_unlock(&hlpmut); \
+ }
+
+ static inline int
+ ATOMIC_CAS_VAL(int *data, int oldVal, int newVal, pthread_mutex_t *phlpmut) {
+ int val;
+ pthread_mutex_lock(phlpmut);
+ if(*data == oldVal) {
+ *data = newVal;
+ }
+ val = *data;
+ pthread_mutex_unlock(phlpmut);
+ return(val);
+ }
+
+# define ATOMIC_DEC(data, phlpmut) { \
+ pthread_mutex_lock(phlpmut); \
+ --(*(data)); \
+ pthread_mutex_unlock(phlpmut); \
+ }
+
+ static inline int
+ ATOMIC_DEC_AND_FETCH(int *data, pthread_mutex_t *phlpmut) {
+ int val;
+ pthread_mutex_lock(phlpmut);
+ val = --(*data);
+ pthread_mutex_unlock(phlpmut);
+ return(val);
+ }
+#if 0
# 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
+# define ATOMIC_INC_AND_FETCH(data) (++(data))
+# define ATOMIC_FETCH_32BIT(data) (data) // TODO: del
+# define ATOMIC_STORE_1_TO_32BIT(data) (data) = 1 // TODO: del
+#endif
+# define DEF_ATOMIC_HELPER_MUT(x) pthread_mutex_t x
+# define INIT_ATOMIC_HELPER_MUT(x) pthread_mutex_init(&(x), NULL)
+# define DESTROY_ATOMIC_HELPER_MUT(x) pthread_mutex_init(&(x), NULL)
+
+# define PREFER_ATOMIC_INC(data) ((void) ++data)
+
#endif
#endif /* #ifndef INCLUDED_ATOMIC_H */
diff --git a/runtime/cfsysline.c b/runtime/cfsysline.c
index 1dd5905e..037e9f84 100644
--- a/runtime/cfsysline.c
+++ b/runtime/cfsysline.c
@@ -217,9 +217,11 @@ static rsRetVal doGetSize(uchar **pp, rsRetVal (*pSetHdlr)(void*, uid_t), void *
case 'K': i *= 1000; ++(*pp); break;
case 'M': i *= 1000000; ++(*pp); break;
case 'G': i *= 1000000000; ++(*pp); break;
- case 'T': i *= (long long) 1000000000000; ++(*pp); break; /* tera */
- case 'P': i *= (long long) 1000000000000000; ++(*pp); break; /* peta */
- case 'E': i *= (long long) 1000000000000000000; ++(*pp); break; /* exa */
+ /* we need to use the multiplication below because otherwise
+ * the compiler gets an error during constant parsing */
+ case 'T': i *= (int64) 1000 * 1000000000; ++(*pp); break; /* tera */
+ case 'P': i *= (int64) 1000000 * 1000000000; ++(*pp); break; /* peta */
+ case 'E': i *= (int64) 1000000000 * 1000000000; ++(*pp); break; /* exa */
}
/* done */
@@ -951,8 +953,6 @@ finalize_it:
*/
void dbgPrintCfSysLineHandlers(void)
{
- DEFiRet;
-
cslCmd_t *pCmd;
cslCmdHdlr_t *pCmdHdlr;
linkedListCookie_t llCookieCmd;
@@ -974,7 +974,6 @@ void dbgPrintCfSysLineHandlers(void)
}
}
dbgprintf("\n");
- ENDfunc
}
diff --git a/runtime/debug.c b/runtime/debug.c
index 6bb8d743..81b45d41 100644
--- a/runtime/debug.c
+++ b/runtime/debug.c
@@ -154,7 +154,9 @@ static pthread_key_t keyCallStack;
*/
static void dbgMutexCancelCleanupHdlr(void *pmut)
{
- pthread_mutex_unlock((pthread_mutex_t*) pmut);
+ int ret;
+ ret = pthread_mutex_unlock((pthread_mutex_t*) pmut);
+ assert(ret == 0);
}
@@ -844,6 +846,7 @@ dbgprint(obj_t *pObj, char *pszMsg, size_t lenMsg)
size_t lenWriteBuf;
struct timespec t;
uchar *pszObjName = NULL;
+ int ret;
/* we must get the object name before we lock the mutex, because the object
* potentially calls back into us. If we locked the mutex, we would deadlock
@@ -855,7 +858,8 @@ dbgprint(obj_t *pObj, char *pszMsg, size_t lenMsg)
pszObjName = obj.GetName(pObj);
}
- pthread_mutex_lock(&mutdbgprint);
+ ret = pthread_mutex_lock(&mutdbgprint);
+ assert(ret == 0); /* make sure mutex operation does not fail */
pthread_cleanup_push(dbgMutexCancelCleanupHdlr, &mutdbgprint);
/* The bWasNL handler does not really work. It works if no thread
@@ -939,6 +943,15 @@ dbgoprint(obj_t *pObj, char *fmt, ...)
va_start(ap, fmt);
lenWriteBuf = vsnprintf(pszWriteBuf, sizeof(pszWriteBuf), fmt, ap);
va_end(ap);
+ if(lenWriteBuf >= sizeof(pszWriteBuf)) {
+ /* prevent buffer overrruns and garbagge display */
+ pszWriteBuf[sizeof(pszWriteBuf) - 5] = '.';
+ pszWriteBuf[sizeof(pszWriteBuf) - 4] = '.';
+ pszWriteBuf[sizeof(pszWriteBuf) - 3] = '.';
+ pszWriteBuf[sizeof(pszWriteBuf) - 2] = '\n';
+ pszWriteBuf[sizeof(pszWriteBuf) - 1] = '\0';
+ lenWriteBuf = sizeof(pszWriteBuf);
+ }
dbgprint(pObj, pszWriteBuf, lenWriteBuf);
}
@@ -950,7 +963,7 @@ void
dbgprintf(char *fmt, ...)
{
va_list ap;
- char pszWriteBuf[20480];
+ char pszWriteBuf[32*1024];
size_t lenWriteBuf;
if(!(Debug && debugging_on))
@@ -1058,7 +1071,7 @@ int dbgEntrFunc(dbgFuncDB_t **ppFuncDB, const char *file, const char *func, int
}
/* when we reach this point, we have a fully-initialized FuncDB! */
- ATOMIC_INC(pFuncDB->nTimesCalled);
+ PREFER_ATOMIC_INC(pFuncDB->nTimesCalled);
if(bLogFuncFlow && dbgPrintNameIsInList((const uchar*)pFuncDB->file, printNameFileRoot))
dbgprintf("%s:%d: %s: enter\n", pFuncDB->file, pFuncDB->line, pFuncDB->func);
if(pThrd->stackPtr >= (int) (sizeof(pThrd->callStack) / sizeof(dbgFuncDB_t*))) {
@@ -1278,11 +1291,11 @@ dbgGetRuntimeOptions(void)
/* this is earlier in the process than the -d option, as such it
* allows us to spit out debug messages from the very beginning.
*/
- Debug = 1;
+ Debug = DEBUG_FULL;
debugging_on = 1;
} else if(!strcasecmp((char*)optname, "debugondemand")) {
/* Enables debugging, but turns off debug output */
- Debug = 1;
+ Debug = DEBUG_ONDEMAND;
debugging_on = 1;
dbgprintf("Note: debug on demand turned on via configuraton file, "
"use USR1 signal to activate.\n");
diff --git a/runtime/debug.h b/runtime/debug.h
index dcbfb930..cfdf819c 100644
--- a/runtime/debug.h
+++ b/runtime/debug.h
@@ -29,6 +29,11 @@
#include <pthread.h>
#include "obj-types.h"
+/* some settings for various debug modes */
+#define DEBUG_OFF 0
+#define DEBUG_ONDEMAND 1
+#define DEBUG_FULL 2
+
/* external static data elements (some time to be replaced) */
extern int Debug; /* debug flag - read-only after startup */
extern int debugging_on; /* read-only, except on sig USR1 */
diff --git a/runtime/glbl.c b/runtime/glbl.c
index 7fa61963..59d1fb0f 100644
--- a/runtime/glbl.c
+++ b/runtime/glbl.c
@@ -64,6 +64,7 @@ static int option_DisallowWarning = 1; /* complain if message from disallowed se
static int bDisableDNS = 0; /* don't look up IP addresses of remote messages */
static prop_t *propLocalHostName = NULL;/* our hostname as FQDN - read-only after startup */
static uchar *LocalHostName = NULL;/* our hostname - read-only after startup */
+static uchar *LocalHostNameOverride = NULL;/* user-overridden 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 */
@@ -72,6 +73,9 @@ static uchar *pszDfltNetstrmDrvr = NULL; /* module name of default netstream dri
static uchar *pszDfltNetstrmDrvrCAF = NULL; /* default CA file for the netstrm driver */
static uchar *pszDfltNetstrmDrvrKeyFile = NULL; /* default key file for the netstrm driver (server) */
static uchar *pszDfltNetstrmDrvrCertFile = NULL; /* default cert file for the netstrm driver (server) */
+#ifdef USE_UNLIMITED_SELECT
+static int iFdSetSize = howmany(FD_SETSIZE, __NFDBITS) * sizeof (fd_mask); /* size of select() bitmask in bytes */
+#endif
/* define a macro for the simple properties' set and get functions
@@ -104,6 +108,9 @@ SIMP_PROP(DisableDNS, bDisableDNS, int)
SIMP_PROP(LocalDomain, LocalDomain, uchar*)
SIMP_PROP(StripDomains, StripDomains, char**)
SIMP_PROP(LocalHosts, LocalHosts, char**)
+#ifdef USE_UNLIMITED_SELECT
+SIMP_PROP(FdSetSize, iFdSetSize, int)
+#endif
SIMP_PROP_SET(LocalFQDNName, LocalFQDNName, uchar*)
SIMP_PROP_SET(LocalHostName, LocalHostName, uchar*)
@@ -150,14 +157,19 @@ GenerateLocalHostNameProperty(void)
prop.Destruct(&propLocalHostName);
CHKiRet(prop.Construct(&propLocalHostName));
- if(LocalHostName == NULL)
- pszName = (uchar*) "[localhost]";
- else {
- if(GetPreserveFQDN() == 1)
- pszName = LocalFQDNName;
- else
- pszName = LocalHostName;
+ if(LocalHostNameOverride == NULL) {
+ if(LocalHostName == NULL)
+ pszName = (uchar*) "[localhost]";
+ else {
+ if(GetPreserveFQDN() == 1)
+ pszName = LocalFQDNName;
+ else
+ pszName = LocalHostName;
+ }
+ } else { /* local hostname is overriden via config */
+ pszName = LocalHostNameOverride;
}
+ DBGPRINTF("GenerateLocalHostName uses '%s'\n", pszName);
CHKiRet(prop.SetString(propLocalHostName, pszName, ustrlen(pszName)));
CHKiRet(prop.ConstructFinalize(propLocalHostName));
@@ -261,6 +273,9 @@ CODESTARTobjQueryInterface(glbl)
SIMP_PROP(DfltNetstrmDrvrCAF)
SIMP_PROP(DfltNetstrmDrvrKeyFile)
SIMP_PROP(DfltNetstrmDrvrCertFile)
+#ifdef USE_UNLIMITED_SELECT
+ SIMP_PROP(FdSetSize)
+#endif
#undef SIMP_PROP
finalize_it:
ENDobjQueryInterface(glbl)
@@ -287,6 +302,10 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
free(pszDfltNetstrmDrvrCertFile);
pszDfltNetstrmDrvrCertFile = NULL;
}
+ if(LocalHostNameOverride != NULL) {
+ free(LocalHostNameOverride);
+ LocalHostNameOverride = NULL;
+ }
if(pszWorkDir != NULL) {
free(pszWorkDir);
pszWorkDir = NULL;
@@ -295,6 +314,9 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
bOptimizeUniProc = 1;
bHUPisRestart = 0;
bPreserveFQDN = 0;
+#ifdef USE_UNLIMITED_SELECT
+ iFdSetSize = howmany(FD_SETSIZE, __NFDBITS) * sizeof (fd_mask);
+#endif
return RS_RET_OK;
}
@@ -315,6 +337,7 @@ 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 *)"localhostname", 0, eCmdHdlrGetWord, NULL, &LocalHostNameOverride, 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));
@@ -338,6 +361,7 @@ BEGINObjClassExit(glbl, OBJ_IS_CORE_MODULE) /* class, version */
free(pszWorkDir);
if(LocalHostName != NULL)
free(LocalHostName);
+ free(LocalHostNameOverride);
if(LocalFQDNName != NULL)
free(LocalFQDNName);
objRelease(prop, CORE_COMPONENT);
diff --git a/runtime/glbl.h b/runtime/glbl.h
index dcfb6d5f..6a332576 100644
--- a/runtime/glbl.h
+++ b/runtime/glbl.h
@@ -62,9 +62,18 @@ BEGINinterface(glbl) /* name must also be changed in ENDinterface macro! */
/* added v3, 2009-06-30 */
rsRetVal (*GenerateLocalHostNameProperty)(void);
prop_t* (*GetLocalHostNameProp)(void);
+ /* note: v4, v5 are already used by more recent versions, so we need to skip them! */
+ /* added v6, 2009-11-16 as part of varmojfekoj's "unlimited select()" patch
+ * Note that it must be always present, otherwise the interface would have different
+ * versions depending on compile settings, what is not acceptable.
+ * Use this property with care, it is only truly available if UNLIMITED_SELECT is enabled
+ * (I did not yet further investigate the details, because that code hopefully can be removed
+ * at some later stage).
+ */
+ SIMP_PROP(FdSetSize, int)
#undef SIMP_PROP
ENDinterface(glbl)
-#define glblCURR_IF_VERSION 3 /* increment whenever you change the interface structure! */
+#define glblCURR_IF_VERSION 6 /* increment whenever you change the interface structure! */
/* version 2 had PreserveFQDN added - rgerhards, 2008-12-08 */
/* the remaining prototypes */
diff --git a/runtime/msg.c b/runtime/msg.c
index c8046d1b..6335f462 100644
--- a/runtime/msg.c
+++ b/runtime/msg.c
@@ -748,7 +748,7 @@ BEGINobjDestruct(msg) /* be sure to specify the object type also in END and CODE
CODESTARTobjDestruct(msg)
/* DEV Debugging only ! dbgprintf("msgDestruct\t0x%lx, Ref now: %d\n", (unsigned long)pThis, pThis->iRefCount - 1); */
# ifdef HAVE_ATOMIC_BUILTINS
- currRefCount = ATOMIC_DEC_AND_FETCH(pThis->iRefCount);
+ currRefCount = ATOMIC_DEC_AND_FETCH(&pThis->iRefCount, NULL);
# else
MsgLock(pThis);
currRefCount = --pThis->iRefCount;
@@ -800,9 +800,18 @@ CODESTARTobjDestruct(msg)
* that we trim too often when the counter wraps.
*/
static unsigned iTrimCtr = 1;
+# ifdef HAVE_ATOMICS
if(ATOMIC_INC_AND_FETCH(iTrimCtr) % 100000 == 0) {
malloc_trim(128*1024);
}
+# else
+static pthread_mutex_t mutTrimCtr = PTHREAD_MUTEX_INITIALIZER;
+ d_pthread_mutex_lock(&mutTrimCtr);
+ if(iTrimCtr++ % 100000 == 0) {
+ malloc_trim(128*1024);
+ }
+ d_pthread_mutex_unlock(&mutTrimCtr);
+# endif
}
# endif
} else {
@@ -1002,7 +1011,7 @@ msg_t *MsgAddRef(msg_t *pM)
{
assert(pM != NULL);
# ifdef HAVE_ATOMIC_BUILTINS
- ATOMIC_INC(pM->iRefCount);
+ ATOMIC_INC(&pM->iRefCount, NULL);
# else
MsgLock(pM);
pM->iRefCount++;
@@ -3046,7 +3055,7 @@ static rsRetVal msgConstructFinalizer(msg_t *pThis)
* rgerhards, 2008-01-14
*/
static rsRetVal
-MsgGetSeverity(obj_t *pThis, int *piSeverity)
+MsgGetSeverity(obj_t_ptr pThis, int *piSeverity)
{
ISOBJ_TYPE_assert(pThis, msg);
assert(piSeverity != NULL);
diff --git a/runtime/msg.h b/runtime/msg.h
index 3a02365b..a6923d52 100644
--- a/runtime/msg.h
+++ b/runtime/msg.h
@@ -32,6 +32,7 @@
#include "obj.h"
#include "syslogd-types.h"
#include "template.h"
+#include "atomic.h"
/* rgerhards 2004-11-08: The following structure represents a
@@ -59,9 +60,9 @@ struct msg {
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. */
pthread_mutex_t mut;
+ int iRefCount; /* reference counter (0 = unused) */
bool bDoLock; /* use the mutex? */
bool bParseHOSTNAME; /* should the hostname be parsed from the message? */
- short iRefCount; /* reference counter (0 = unused) */
/* background: the hostname is not present on "regular" messages
* received via UNIX domain sockets from the same machine. However,
* it is available when we have a forwarder (e.g. rfc3195d) using local
@@ -130,6 +131,7 @@ struct msg {
#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 */
+#define NO_PRI_IN_RAW 0x100 /* rawmsg does not include a PRI (Solaris!), but PRI is already set correctly in the msg object */
/* function prototypes
diff --git a/runtime/net.c b/runtime/net.c
index 0866efd7..8fd8cc2b 100644
--- a/runtime/net.c
+++ b/runtime/net.c
@@ -502,8 +502,8 @@ static inline void MaskIP4 (struct in_addr *addr, uint8_t bits) {
addr->s_addr &= htonl(0xffffffff << (32 - bits));
}
-#define SIN(sa) ((struct sockaddr_in *)(sa))
-#define SIN6(sa) ((struct sockaddr_in6 *)(sa))
+#define SIN(sa) ((struct sockaddr_in *)(void*)(sa))
+#define SIN6(sa) ((struct sockaddr_in6 *)(void*)(sa))
/* This is a cancel-safe getnameinfo() version, because we learned
@@ -1165,12 +1165,12 @@ void debugListenInfo(int fd, char *type)
switch(sa.sa_family) {
case PF_INET:
szFamily = "IPv4";
- ipv4 = (struct sockaddr_in*) &sa;
+ ipv4 = (struct sockaddr_in*)(void*) &sa;
port = ntohs(ipv4->sin_port);
break;
case PF_INET6:
szFamily = "IPv6";
- ipv6 = (struct sockaddr_in6*) &sa;
+ ipv6 = (struct sockaddr_in6*)(void*) &sa;
port = ntohs(ipv6->sin6_port);
break;
default:
diff --git a/runtime/nsdsel_ptcp.c b/runtime/nsdsel_ptcp.c
index 41b85e0c..e2cfca7c 100644
--- a/runtime/nsdsel_ptcp.c
+++ b/runtime/nsdsel_ptcp.c
@@ -36,6 +36,7 @@
#include "errmsg.h"
#include "nsd_ptcp.h"
#include "nsdsel_ptcp.h"
+#include "unlimited_select.h"
/* static data */
DEFobjStaticHelpers
@@ -47,14 +48,23 @@ DEFobjCurrIf(glbl)
*/
BEGINobjConstruct(nsdsel_ptcp) /* be sure to specify the object type also in END macro! */
pThis->maxfds = 0;
+#ifdef USE_UNLIMITED_SELECT
+ pThis->pReadfds = calloc(1, glbl.GetFdSetSize());
+ pThis->pWritefds = calloc(1, glbl.GetFdSetSize());
+#else
FD_ZERO(&pThis->readfds);
FD_ZERO(&pThis->writefds);
+#endif
ENDobjConstruct(nsdsel_ptcp)
/* destructor for the nsdsel_ptcp object */
BEGINobjDestruct(nsdsel_ptcp) /* be sure to specify the object type also in END and CODESTART macros! */
CODESTARTobjDestruct(nsdsel_ptcp)
+#ifdef USE_UNLIMITED_SELECT
+ freeFdSet(pThis->pReadfds);
+ freeFdSet(pThis->pWritefds);
+#endif
ENDobjDestruct(nsdsel_ptcp)
@@ -65,20 +75,27 @@ Add(nsdsel_t *pNsdsel, nsd_t *pNsd, nsdsel_waitOp_t waitOp)
DEFiRet;
nsdsel_ptcp_t *pThis = (nsdsel_ptcp_t*) pNsdsel;
nsd_ptcp_t *pSock = (nsd_ptcp_t*) pNsd;
+#ifdef USE_UNLIMITED_SELECT
+ fd_set *pReadfds = pThis->pReadfds;
+ fd_set *pWritefds = pThis->pWritefds;
+#else
+ fd_set *pReadfds = &pThis->readfds;
+ fd_set *pWritefds = &pThis->writefds;
+#endif
ISOBJ_TYPE_assert(pSock, nsd_ptcp);
ISOBJ_TYPE_assert(pThis, nsdsel_ptcp);
switch(waitOp) {
case NSDSEL_RD:
- FD_SET(pSock->sock, &pThis->readfds);
+ FD_SET(pSock->sock, pReadfds);
break;
case NSDSEL_WR:
- FD_SET(pSock->sock, &pThis->writefds);
+ FD_SET(pSock->sock, pWritefds);
break;
case NSDSEL_RDWR:
- FD_SET(pSock->sock, &pThis->readfds);
- FD_SET(pSock->sock, &pThis->writefds);
+ FD_SET(pSock->sock, pReadfds);
+ FD_SET(pSock->sock, pWritefds);
break;
}
@@ -98,6 +115,13 @@ Select(nsdsel_t *pNsdsel, int *piNumReady)
DEFiRet;
int i;
nsdsel_ptcp_t *pThis = (nsdsel_ptcp_t*) pNsdsel;
+#ifdef USE_UNLIMITED_SELECT
+ fd_set *pReadfds = pThis->pReadfds;
+ fd_set *pWritefds = pThis->pWritefds;
+#else
+ fd_set *pReadfds = &pThis->readfds;
+ fd_set *pWritefds = &pThis->writefds;
+#endif
ISOBJ_TYPE_assert(pThis, nsdsel_ptcp);
assert(piNumReady != NULL);
@@ -106,13 +130,13 @@ Select(nsdsel_t *pNsdsel, int *piNumReady)
// TODO: name in dbgprintf!
dbgprintf("--------<NSDSEL_PTCP> calling select, active fds (max %d): ", pThis->maxfds);
for(i = 0; i <= pThis->maxfds; ++i)
- if(FD_ISSET(i, &pThis->readfds) || FD_ISSET(i, &pThis->writefds))
+ if(FD_ISSET(i, pReadfds) || FD_ISSET(i, pWritefds))
dbgprintf("%d ", i);
dbgprintf("\n");
}
/* now do the select */
- *piNumReady = select(pThis->maxfds+1, &pThis->readfds, &pThis->writefds, NULL, NULL);
+ *piNumReady = select(pThis->maxfds+1, pReadfds, pWritefds, NULL, NULL);
RETiRet;
}
@@ -125,6 +149,13 @@ IsReady(nsdsel_t *pNsdsel, nsd_t *pNsd, nsdsel_waitOp_t waitOp, int *pbIsReady)
DEFiRet;
nsdsel_ptcp_t *pThis = (nsdsel_ptcp_t*) pNsdsel;
nsd_ptcp_t *pSock = (nsd_ptcp_t*) pNsd;
+#ifdef USE_UNLIMITED_SELECT
+ fd_set *pReadfds = pThis->pReadfds;
+ fd_set *pWritefds = pThis->pWritefds;
+#else
+ fd_set *pReadfds = &pThis->readfds;
+ fd_set *pWritefds = &pThis->writefds;
+#endif
ISOBJ_TYPE_assert(pThis, nsdsel_ptcp);
ISOBJ_TYPE_assert(pSock, nsd_ptcp);
@@ -132,14 +163,14 @@ IsReady(nsdsel_t *pNsdsel, nsd_t *pNsd, nsdsel_waitOp_t waitOp, int *pbIsReady)
switch(waitOp) {
case NSDSEL_RD:
- *pbIsReady = FD_ISSET(pSock->sock, &pThis->readfds);
+ *pbIsReady = FD_ISSET(pSock->sock, pReadfds);
break;
case NSDSEL_WR:
- *pbIsReady = FD_ISSET(pSock->sock, &pThis->writefds);
+ *pbIsReady = FD_ISSET(pSock->sock, pWritefds);
break;
case NSDSEL_RDWR:
- *pbIsReady = FD_ISSET(pSock->sock, &pThis->readfds)
- | FD_ISSET(pSock->sock, &pThis->writefds);
+ *pbIsReady = FD_ISSET(pSock->sock, pReadfds)
+ | FD_ISSET(pSock->sock, pWritefds);
break;
}
diff --git a/runtime/nsdsel_ptcp.h b/runtime/nsdsel_ptcp.h
index 6c0c7fa7..f9ec8210 100644
--- a/runtime/nsdsel_ptcp.h
+++ b/runtime/nsdsel_ptcp.h
@@ -31,8 +31,13 @@ typedef nsdsel_if_t nsdsel_ptcp_if_t; /* we just *implement* this interface */
struct nsdsel_ptcp_s {
BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */
int maxfds;
+#ifdef USE_UNLIMITED_SELECT
+ fd_set *pReadfds;
+ fd_set *pWritefds;
+#else
fd_set readfds;
fd_set writefds;
+#endif
};
/* interface is defined in nsd.h, we just implement it! */
diff --git a/runtime/parser.c b/runtime/parser.c
index d27f3e38..be9304d7 100644
--- a/runtime/parser.c
+++ b/runtime/parser.c
@@ -285,23 +285,28 @@ rsRetVal parseMsg(msg_t *pMsg)
lenMsg = pMsg->iLenRawMsg;
msg = pMsg->pszRawMsg;
pri = DEFUPRI;
- 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(--lenMsg > 0 && isdigit((int) *++msg)) {
- pri = 10 * pri + (*msg - '0');
+ if(pMsg->msgFlags & NO_PRI_IN_RAW) {
+ /* In this case, simply do so as if the pri would be right at top */
+ MsgSetAfterPRIOffs(pMsg, 0);
+ } else {
+ 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(--lenMsg > 0 && isdigit((int) *++msg)) {
+ pri = 10 * pri + (*msg - '0');
+ }
+ if(*msg == '>')
+ ++msg;
+ if(pri & ~(LOG_FACMASK|LOG_PRIMASK))
+ pri = DEFUPRI;
}
- 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);
}
- pMsg->iFacility = LOG_FAC(pri);
- pMsg->iSeverity = LOG_PRI(pri);
- MsgSetAfterPRIOffs(pMsg, msg - pMsg->pszRawMsg);
/* 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.
diff --git a/runtime/prop.c b/runtime/prop.c
index d188b2ed..7f2a56ff 100644
--- a/runtime/prop.c
+++ b/runtime/prop.c
@@ -53,6 +53,7 @@ DEFobjStaticHelpers
*/
BEGINobjConstruct(prop) /* be sure to specify the object type also in END macro! */
pThis->iRefCount = 1;
+ INIT_ATOMIC_HELPER_MUT(pThis->mutRefCount);
ENDobjConstruct(prop)
@@ -60,11 +61,12 @@ ENDobjConstruct(prop)
BEGINobjDestruct(prop) /* be sure to specify the object type also in END and CODESTART macros! */
int currRefCount;
CODESTARTobjDestruct(prop)
- currRefCount = ATOMIC_DEC_AND_FETCH(pThis->iRefCount);
+ currRefCount = ATOMIC_DEC_AND_FETCH(&pThis->iRefCount, &pThis->mutRefCount);
if(currRefCount == 0) {
/* (only) in this case we need to actually destruct the object */
if(pThis->len >= CONF_PROP_BUFSIZE)
free(pThis->szVal.psz);
+ DESTROY_ATOMIC_HELPER_MUT(pThis->mutRefCount);
} else {
pThis = NULL; /* tell framework NOT to destructing the object! */
}
@@ -132,7 +134,7 @@ propConstructFinalize(prop_t __attribute__((unused)) *pThis)
*/
static rsRetVal AddRef(prop_t *pThis)
{
- ATOMIC_INC(pThis->iRefCount);
+ ATOMIC_INC(&pThis->iRefCount, &pThis->mutRefCount);
return RS_RET_OK;
}
diff --git a/runtime/prop.h b/runtime/prop.h
index e3519664..07b2ab7e 100644
--- a/runtime/prop.h
+++ b/runtime/prop.h
@@ -24,6 +24,7 @@
*/
#ifndef INCLUDED_PROP_H
#define INCLUDED_PROP_H
+#include "atomic.h"
/* the prop object */
struct prop_s {
@@ -34,6 +35,7 @@ struct prop_s {
uchar sz[CONF_PROP_BUFSIZE];
} szVal;
int len; /* we use int intentionally, otherwise we may get some troubles... */
+ DEF_ATOMIC_HELPER_MUT(mutRefCount);
};
/* interfaces */
diff --git a/runtime/queue.c b/runtime/queue.c
index 4d94941a..9c7f96d0 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -110,7 +110,7 @@ static inline void queueDrain(qqueue_t *pThis)
ASSERT(pThis != NULL);
/* iQueueSize is not decremented by qDel(), so we need to do it ourselves */
- while(ATOMIC_DEC_AND_FETCH(pThis->iQueueSize) > 0) {
+ while(ATOMIC_DEC_AND_FETCH(&pThis->iQueueSize, &pThis->mutQueueSize) > 0) {
pThis->qDel(pThis, &pUsr);
if(pUsr != NULL) {
objDestruct(pUsr);
@@ -1027,7 +1027,7 @@ qqueueAdd(qqueue_t *pThis, void *pUsr)
CHKiRet(pThis->qAdd(pThis, pUsr));
if(pThis->qType != QUEUETYPE_DIRECT) {
- ATOMIC_INC(pThis->iQueueSize);
+ ATOMIC_INC(&pThis->iQueueSize, &pThis->mutQueueSize);
dbgoprint((obj_t*) pThis, "entry added, size now %d entries\n", pThis->iQueueSize);
}
@@ -1056,7 +1056,7 @@ qqueueDel(qqueue_t *pThis, void *pUsr)
iRet = qqueueGetUngottenObj(pThis, (obj_t**) pUsr);
} else {
iRet = pThis->qDel(pThis, pUsr);
- ATOMIC_DEC(pThis->iQueueSize);
+ ATOMIC_DEC(&pThis->iQueueSize, &pThis->mutQueueSize);
}
dbgoprint((obj_t*) pThis, "entry deleted, state %d, size now %d entries\n",
@@ -1344,6 +1344,8 @@ rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThread
break;
}
+ INIT_ATOMIC_HELPER_MUT(pThis->mutQueueSize);
+
finalize_it:
OBJCONSTRUCT_CHECK_SUCCESS_AND_CLEANUP
RETiRet;
@@ -2064,6 +2066,8 @@ CODESTARTobjDestruct(qqueue)
pthread_cond_destroy(&pThis->belowFullDlyWtrMrk);
pthread_cond_destroy(&pThis->belowLightDlyWtrMrk);
+ DESTROY_ATOMIC_HELPER_MUT(pThis->mutQueueSize);
+
/* type-specific destructor */
iRet = pThis->qDestruct(pThis);
diff --git a/runtime/queue.h b/runtime/queue.h
index 1d82d8d9..aafdaa45 100644
--- a/runtime/queue.h
+++ b/runtime/queue.h
@@ -160,6 +160,7 @@ typedef struct queue_s {
strm_t *pRead; /* current file to be read */
} disk;
} tVars;
+ DEF_ATOMIC_HELPER_MUT(mutQueueSize);
} qqueue_t;
/* some symbolic constants for easier reference */
diff --git a/runtime/rsyslog.c b/runtime/rsyslog.c
index 443d0f41..c209ae30 100644
--- a/runtime/rsyslog.c
+++ b/runtime/rsyslog.c
@@ -80,6 +80,7 @@
#include "prop.h"
#include "rule.h"
#include "ruleset.h"
+#include "atomic.h"
/* forward definitions */
static rsRetVal dfltErrLogger(int, uchar *errMsg);
@@ -215,6 +216,7 @@ rsrtExit(void)
glblClassExit();
rulesetClassExit();
ruleClassExit();
+
objClassExit(); /* *THIS* *MUST/SHOULD?* always be the first class initilizer being called (except debug)! */
}
diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h
index 8979893a..03f5120d 100644
--- a/runtime/rsyslog.h
+++ b/runtime/rsyslog.h
@@ -61,6 +61,7 @@
/* define some base data types */
+
typedef unsigned char uchar;/* get rid of the unhandy "unsigned char" */
typedef struct thrdInfo thrdInfo_t;
typedef struct obj_s obj_t;
@@ -78,8 +79,6 @@ typedef struct nsd_gsspi_s nsd_gsspi_t;
typedef struct nsd_nss_s nsd_nss_t;
typedef struct nsdsel_ptcp_s nsdsel_ptcp_t;
typedef struct nsdsel_gtls_s nsdsel_gtls_t;
-typedef obj_t nsd_t;
-typedef obj_t nsdsel_t;
typedef struct msg msg_t;
typedef struct prop_s prop_t;
typedef struct interface_s interface_t;
@@ -98,6 +97,21 @@ typedef rsRetVal (*prsf_t)(struct vmstk_s*, int); /* pointer to a RainerScript f
typedef struct tcpLstnPortList_s tcpLstnPortList_t; // TODO: rename?
typedef struct strmLstnPortList_s strmLstnPortList_t; // TODO: rename?
+/* under Solaris (actually only SPARC), we need to redefine some types
+ * to be void, so that we get void* pointers. Otherwise, we will see
+ * alignment errors.
+ */
+#ifdef OS_SOLARIS
+ typedef void * obj_t_ptr;
+ typedef void nsd_t;
+ typedef void nsdsel_t;
+#else
+ typedef obj_t *obj_t_ptr;
+ typedef obj_t nsd_t;
+ typedef obj_t nsdsel_t;
+#endif
+
+
/* some universal 64 bit define... */
typedef long long int64;
typedef long long unsigned uint64;
@@ -360,6 +374,15 @@ enum rsRetVal_ /** return value. All methods return this if not specified oth
RS_RET_VAR_NOT_FOUND = -2142, /**< variable not found */
RS_RET_EMPTY_MSG = -2143, /**< provided (raw) MSG is empty */
RS_RET_PEER_CLOSED_CONN = -2144, /**< remote peer closed connection (information, no error) */
+ RS_RET_ERR_OPEN_KLOG = -2145, /**< error opening the kernel log socket (primarily solaris) */
+ RS_RET_ERR_AQ_CONLOG = -2146, /**< error aquiring console log (on solaris) */
+ RS_RET_ERR_DOOR = -2147, /**< some problems with handling the Solaris door functionality */
+ RS_RET_NO_SOCK_CONFIGURED = -2166, /**< no socket (name) was configured where one is required */
+ RS_RET_NO_LSTN_DEFINED = -2172, /**< no listener defined (e.g. inside an input module) */
+ RS_RET_EPOLL_CR_FAILED = -2173, /**< epoll_create() failed */
+ RS_RET_EPOLL_CTL_FAILED = -2174, /**< epoll_ctl() failed */
+ RS_RET_INTERNAL_ERROR = -2175, /**< rsyslogd internal error, unexpected code path reached */
+ RS_RET_OUTDATED_STMT = -2184, /**< some outdated statement/functionality is being used in conf file */
/* RainerScript error messages (range 1000.. 1999) */
RS_RET_SYSVAR_NOT_FOUND = 1001, /**< system variable could not be found (maybe misspelled) */
diff --git a/runtime/stream.c b/runtime/stream.c
index a12dda41..44b24ee2 100644
--- a/runtime/stream.c
+++ b/runtime/stream.c
@@ -222,7 +222,8 @@ doPhysOpen(strm_t *pThis)
}
pThis->fd = open((char*)pThis->pszCurrFName, iFlags | O_LARGEFILE, pThis->tOpenMode);
- DBGPRINTF("file '%s' opened as #%d with mode %d\n", pThis->pszCurrFName, pThis->fd, pThis->tOpenMode);
+ DBGPRINTF("file '%s' opened as #%d with mode %d\n", pThis->pszCurrFName,
+ pThis->fd, (int) pThis->tOpenMode);
if(pThis->fd == -1) {
char errStr[1024];
int err = errno;
diff --git a/runtime/unlimited_select.h b/runtime/unlimited_select.h
new file mode 100644
index 00000000..32dadc03
--- /dev/null
+++ b/runtime/unlimited_select.h
@@ -0,0 +1,45 @@
+/* unlimited_select.h
+ * Tweak the macros for accessing fd_set so that the select() syscall
+ * won't be limited to a particular number of file descriptors.
+ *
+ * Copyright 2009 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of rsyslog.
+ *
+ * Rsyslog is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Rsyslog 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Rsyslog. If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the GPL can be found in the file "COPYING" in this distribution.
+ */
+
+#ifndef UNLIMITED_SELECT_H_INCLUDED
+
+#include <string.h>
+#include <stdlib.h>
+#include <sys/select.h>
+#include "glbl.h"
+
+#ifdef USE_UNLIMITED_SELECT
+# undef FD_ZERO
+# define FD_ZERO(set) memset((set), 0, glbl.GetFdSetSize());
+#endif
+
+#ifdef USE_UNLIMITED_SELECT
+void freeFdSet(fd_set *p) {
+ free(p);
+}
+#else
+# define freeFdSet(x)
+#endif
+
+#endif /* #ifndef UNLIMITED_SELECT_H_INCLUDED */
diff --git a/runtime/vm.c b/runtime/vm.c
index aaf3c879..0ed174d1 100644
--- a/runtime/vm.c
+++ b/runtime/vm.c
@@ -34,6 +34,7 @@
#include "vm.h"
#include "sysvar.h"
#include "stringbuf.h"
+#include "unicode-helper.h"
/* static data */
DEFobjStaticHelpers
@@ -41,6 +42,8 @@ DEFobjCurrIf(vmstk)
DEFobjCurrIf(var)
DEFobjCurrIf(sysvar)
+static pthread_mutex_t mutGetenv; /* we need to make this global because otherwise we can not guarantee proper init! */
+
/* ------------------------------ function registry code and structures ------------------------------ */
/* we maintain a registry of known functions */
@@ -542,6 +545,42 @@ finalize_it:
}
+/* The getenv function. Note that we guard the OS call by a mutex, as that
+ * function is not guaranteed to be thread-safe. This implementation here is far from
+ * being optimal, at least we should cache the result. This is left TODO for
+ * a later revision.
+ * rgerhards, 2009-11-03
+ */
+static rsRetVal
+rsf_getenv(vmstk_t *pStk, int numOperands)
+{
+ DEFiRet;
+ var_t *operand1;
+ char *envResult;
+ cstr_t *pCstr;
+
+ if(numOperands != 1)
+ ABORT_FINALIZE(RS_RET_INVLD_NBR_ARGUMENTS);
+
+ /* pop args and do operaton (trivial case here...) */
+ vmstk.PopString(pStk, &operand1);
+ d_pthread_mutex_lock(&mutGetenv);
+ envResult = getenv((char*) rsCStrGetSzStr(operand1->val.pStr));
+ DBGPRINTF("rsf_getenv(): envvar '%s', return '%s'\n", rsCStrGetSzStr(operand1->val.pStr),
+ envResult == NULL ? "(NULL)" : envResult);
+ iRet = rsCStrConstructFromszStr(&pCstr, (envResult == NULL) ? UCHAR_CONSTANT("") : (uchar*)envResult);
+ d_pthread_mutex_unlock(&mutGetenv);
+ if(iRet != RS_RET_OK)
+ FINALIZE; /* need to do this after mutex is unlocked! */
+
+ /* Store result and cleanup */
+ var.SetString(operand1, pCstr);
+ 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
@@ -759,6 +798,8 @@ BEGINObjClassExit(vm, OBJ_IS_CORE_MODULE) /* class, version */
objRelease(sysvar, CORE_COMPONENT);
objRelease(var, CORE_COMPONENT);
objRelease(vmstk, CORE_COMPONENT);
+
+ pthread_mutex_destroy(&mutGetenv);
ENDObjClassExit(vm)
@@ -779,6 +820,9 @@ BEGINObjClassInit(vm, 1, OBJ_IS_CORE_MODULE) /* class, version */
/* register built-in functions // TODO: move to its own module */
CHKiRet(rsfrAddFunction((uchar*)"strlen", rsf_strlen));
CHKiRet(rsfrAddFunction((uchar*)"tolower", rsf_tolower));
+ CHKiRet(rsfrAddFunction((uchar*)"getenv", rsf_getenv));
+
+ pthread_mutex_init(&mutGetenv, NULL);
ENDObjClassInit(vm)
diff --git a/runtime/wti.c b/runtime/wti.c
index abdf4add..90bb14ed 100644
--- a/runtime/wti.c
+++ b/runtime/wti.c
@@ -146,7 +146,7 @@ wtiSetState(wti_t *pThis, qWrkCmd_t tCmd, int bActiveOnly, int bLockMutex)
break;
}
/* apply the new state */
- unsigned val = ATOMIC_CAS_VAL(pThis->tCurrCmd, tCurrCmd, tCmd);
+ unsigned val = ATOMIC_CAS_VAL((int*)&pThis->tCurrCmd, tCurrCmd, tCmd, &pThis->mutCurrCmd);
if(val != tCurrCmd) {
DBGPRINTF("wtiSetState PROBLEM, tCurrCmd %d overwritten with %d, wanted to set %d\n", tCurrCmd, val, tCmd);
}
@@ -178,7 +178,7 @@ wtiCancelThrd(wti_t *pThis)
dbgoprint((obj_t*) pThis, "canceling worker thread, curr stat %d\n", pThis->tCurrCmd);
pthread_cancel(pThis->thrdID);
wtiSetState(pThis, eWRKTHRD_TERMINATING, 0, MUTEX_ALREADY_LOCKED);
- ATOMIC_STORE_1_TO_INT(pThis->pWtp->bThrdStateChanged); /* indicate change, so harverster will be called */
+ wtpSetThrdStateChanged(pThis->pWtp, 1); /* indicate change, so harverster will be called */
}
d_pthread_mutex_unlock(&pThis->mut);
@@ -209,6 +209,7 @@ CODESTARTobjDestruct(wti)
/* actual destruction */
pthread_cond_destroy(&pThis->condExitDone);
pthread_mutex_destroy(&pThis->mut);
+ DESTROY_ATOMIC_HELPER_MUT(pThis->mutCurrCmd);
free(pThis->pszDbgHdr);
ENDobjDestruct(wti)
@@ -219,6 +220,7 @@ ENDobjDestruct(wti)
BEGINobjConstruct(wti) /* be sure to specify the object type also in END macro! */
pthread_cond_init(&pThis->condExitDone, NULL);
pthread_mutex_init(&pThis->mut, NULL);
+ INIT_ATOMIC_HELPER_MUT(pThis->mutCurrCmd);
ENDobjConstruct(wti)
@@ -326,8 +328,7 @@ wtiWorkerCancelCleanup(void *arg)
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
d_pthread_mutex_lock(&pWtp->mut);
wtiSetState(pThis, eWRKTHRD_TERMINATING, 0, MUTEX_ALREADY_LOCKED);
- /* TODO: sync access? I currently think it is NOT needed -- rgerhards, 2008-01-28 */
- ATOMIC_STORE_1_TO_INT(pWtp->bThrdStateChanged); /* indicate change, so harverster will be called */
+ wtpSetThrdStateChanged(pWtp, 1); /* indicate change, so harverster will be called */
d_pthread_mutex_unlock(&pWtp->mut);
pthread_setcancelstate(iCancelStateSave, NULL);
@@ -414,7 +415,7 @@ wtiWorker(wti_t *pThis)
pWtp->pfOnWorkerShutdown(pWtp->pUsr);
wtiSetState(pThis, eWRKTHRD_TERMINATING, 0, MUTEX_ALREADY_LOCKED);
- ATOMIC_STORE_1_TO_INT(pWtp->bThrdStateChanged); /* indicate change, so harverster will be called */
+ wtpSetThrdStateChanged(pWtp, 1); /* indicate change, so harverster will be called */
d_pthread_mutex_unlock(&pThis->mut);
pthread_setcancelstate(iCancelStateSave, NULL);
diff --git a/runtime/wti.h b/runtime/wti.h
index 72653b15..d81672f3 100644
--- a/runtime/wti.h
+++ b/runtime/wti.h
@@ -39,6 +39,7 @@ typedef struct wti_s {
pthread_mutex_t mut;
bool bShutdownRqtd; /* shutdown for this thread requested? 0 - no , 1 - yes */
uchar *pszDbgHdr; /* header string for debug messages */
+ DEF_ATOMIC_HELPER_MUT(mutCurrCmd);
} wti_t;
/* some symbolic constants for easier reference */
diff --git a/runtime/wtp.c b/runtime/wtp.c
index f71d5855..b4fd2e04 100644
--- a/runtime/wtp.c
+++ b/runtime/wtp.c
@@ -97,6 +97,7 @@ BEGINobjConstruct(wtp) /* be sure to specify the object type also in END macro!
pThis->pfOnWorkerCancel = NotImplementedDummy;
pThis->pfOnWorkerStartup = NotImplementedDummy;
pThis->pfOnWorkerShutdown = NotImplementedDummy;
+ INIT_ATOMIC_HELPER_MUT(pThis->mutThrdStateChanged);
ENDobjConstruct(wtp)
@@ -153,6 +154,7 @@ CODESTARTobjDestruct(wtp)
pthread_cond_destroy(&pThis->condThrdTrm);
pthread_mutex_destroy(&pThis->mut);
pthread_mutex_destroy(&pThis->mutThrdShutdwn);
+ DESTROY_ATOMIC_HELPER_MUT(pThis->mutThrdStateChanged);
free(pThis->pszDbgHdr);
ENDobjDestruct(wtp)
@@ -186,6 +188,20 @@ wtpWakeupAllWrkr(wtp_t *pThis)
}
+/* set the bThrdStateChanged in an atomic way. Note that
+ * val may only be 0 or 1.
+ */
+void
+wtpSetThrdStateChanged(wtp_t *pThis, int val)
+{
+ if(val == 0) {
+ ATOMIC_STORE_0_TO_INT(&pThis->bThrdStateChanged, pThis->mutThrdStateChanged);
+ } else {
+ ATOMIC_STORE_1_TO_INT(&pThis->bThrdStateChanged, pThis->mutThrdStateChanged);
+ }
+}
+
+
/* check if we had any worker thread changes and, if so, act
* on them. At a minimum, terminated threads are harvested (joined).
* This function MUST NEVER block on the queue mutex!
@@ -216,7 +232,7 @@ wtpProcessThrdChanges(wtp_t *pThis)
*/
do {
/* reset the change marker */
- ATOMIC_STORE_0_TO_INT(pThis->bThrdStateChanged);
+ wtpSetThrdStateChanged(pThis, 0);
/* go through all threads */
for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) {
wtiProcessThrdChanges(pThis->pWrkr[i], LOCK_MUTEX);
@@ -421,13 +437,15 @@ wtpWrkrExecCancelCleanup(void *arg)
static void *
wtpWorker(void *arg) /* the arg is actually a wti object, even though we are in wtp! */
{
- uchar *pszDbgHdr;
- uchar thrdName[32] = "rs:";
DEFiRet;
DEFVARS_mutexProtection;
wti_t *pWti = (wti_t*) arg;
wtp_t *pThis;
sigset_t sigSet;
+# if HAVE_PRCTL && defined PR_SET_NAME
+ uchar *pszDbgHdr;
+ uchar thrdName[32] = "rs:";
+# endif
ISOBJ_TYPE_assert(pWti, wti);
pThis = pWti->pWtp;
diff --git a/runtime/wtp.h b/runtime/wtp.h
index 1ce171cc..640c3320 100644
--- a/runtime/wtp.h
+++ b/runtime/wtp.h
@@ -26,6 +26,7 @@
#include <pthread.h>
#include "obj.h"
+#include "atomic.h"
/* commands and states for worker threads. */
typedef enum {
@@ -79,6 +80,7 @@ typedef struct wtp_s {
rsRetVal (*pfOnWorkerShutdown)(void *pUsr);
/* end user objects */
uchar *pszDbgHdr; /* header string for debug messages */
+ DEF_ATOMIC_HELPER_MUT(mutThrdStateChanged);
} wtp_t;
/* some symbolic constants for easier reference */
@@ -100,6 +102,7 @@ rsRetVal wtpSetDbgHdr(wtp_t *pThis, uchar *pszMsg, size_t lenMsg);
rsRetVal wtpSignalWrkrTermination(wtp_t *pWtp);
rsRetVal wtpShutdownAll(wtp_t *pThis, wtpState_t tShutdownCmd, struct timespec *ptTimeout);
int wtpGetCurNumWrkr(wtp_t *pThis, int bLockMutex);
+void wtpSetThrdStateChanged(wtp_t *pThis, int val);
PROTOTYPEObjClassInit(wtp);
PROTOTYPEpropSetMethFP(wtp, pfChkStopWrkr, rsRetVal(*pVal)(void*, int));
PROTOTYPEpropSetMethFP(wtp, pfRateLimiter, rsRetVal(*pVal)(void*));
generated by cgit (git 2.34.1) at 2024-04-28 16:56:37 +0000