summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--linux/Makefile1
-rw-r--r--master.make2
-rw-r--r--syslogd.c154
3 files changed, 122 insertions, 35 deletions
diff --git a/linux/Makefile b/linux/Makefile
index 7d0498bf..0d341184 100644
--- a/linux/Makefile
+++ b/linux/Makefile
@@ -73,6 +73,7 @@ endif
ifeq ($(strip $(FEATURE_PTHREADS)), 1)
F_PTHREADS=-DUSE_PTHREADS
+ LPTHREAD=-lpthread
endif
ifeq ($(strip $(FEATURE_RFC3195)), 1)
diff --git a/master.make b/master.make
index 3c8c1d13..4df8c5b5 100644
--- a/master.make
+++ b/master.make
@@ -42,7 +42,7 @@ test: syslog_tst tsyslogd
install: install_man install_exec
syslogd: syslogd.o pidfile.o template.o stringbuf.o srUtils.o outchannel.o parse.o
- ${CC} ${LDFLAGS} -o syslogd syslogd.o pidfile.o template.o outchannel.o stringbuf.o srUtils.o parse.o ${LIBS}
+ ${CC} ${LDFLAGS} $(LPTHREAD) -o syslogd syslogd.o pidfile.o template.o outchannel.o stringbuf.o srUtils.o parse.o ${LIBS}
rfc3195d: rfc3195d.o
${CC} ${LDFLAGS} -o rfc3195d rfc3195d.o ${LIBLOGGING_BIN}
diff --git a/syslogd.c b/syslogd.c
index 02504a7f..7ff2b09d 100644
--- a/syslogd.c
+++ b/syslogd.c
@@ -346,8 +346,7 @@ int funix[MAXFUNIX] = { -1, }; /* read-only after startup */
/*
* Flags to logmsg().
*/
-
-#define IGN_CONS 0x001 /* don't print on console */
+/* NO LONGER NEEDED: #define IGN_CONS 0x001 * don't print on console */
#define SYNC_FILE 0x002 /* do fsync on file after printing */
#define ADDDATE 0x004 /* add a date to the message */
#define MARK 0x008 /* this message is a mark */
@@ -456,6 +455,7 @@ struct msg {
char *pszTIMESTAMP3164; /* TIMESTAMP as RFC3164 formatted string (always 15 charcters) */
char *pszTIMESTAMP3339; /* TIMESTAMP as RFC3339 formatted string (32 charcters at most) */
char *pszTIMESTAMP_MySQL;/* TIMESTAMP as MySQL formatted string (always 14 charcters) */
+ int msgFlags; /* flags associated with this message */
};
@@ -577,6 +577,17 @@ static EHostnameCmpMode eDfltHostnameCmpMode;
static rsCStrObj *pDfltHostnameCmp;
static rsCStrObj *pDfltProgNameCmp;
+/* supporting structures for multithreading */
+#ifdef USE_PTHREADS
+static pthread_cond_t cndRunWorker = PTHREAD_COND_INITIALIZER;
+static pthread_mutex_t mtxRunWorker = PTHREAD_MUTEX_INITIALIZER;
+static pthread_t thrdWorker;
+static int bGlblDone = 0;
+static struct msg *bufpMsg = NULL;
+#endif
+/* END supporting structures for multithreading */
+
+
/*
* Intervals at which we flush out "message repeated" messages,
* in seconds after previous message is logged. After each flush,
@@ -770,6 +781,8 @@ static char template_StdUsrMsgFmt[] = "\" %syslogtag%%msg%\n\r\"";
static char template_StdDBFmt[] = "\"insert into SystemEvents (Message, Facility, FromHost, Priority, DeviceReportedTime, ReceivedAt, InfoUnitID, SysLogTag) values ('%msg%', %syslogfacility%, '%HOSTNAME%', %syslogpriority%, '%timereported:::date-mysql%', '%timegenerated:::date-mysql%', %iut%, '%syslogtag%')\",SQL";
/* end template */
+
+static void *singleWorker(void *vParam); /* REMOVEME later 2005-10-24 */
/* Function prototypes. */
int main(int argc, char **argv);
static char **crunch_list(char *list);
@@ -778,7 +791,7 @@ static void untty(void);
static void printchopped(char *hname, char *msg, int len, int fd, int iSourceType);
static void printline(char *hname, char *msg, int iSource);
static void logmsg(int pri, struct msg*, int flags);
-static void fprintlog(register struct filed *f, int flags);
+static void fprintlog(register struct filed *f);
static void endtty();
static void wallmsg(register struct filed *f);
static void reapchild();
@@ -821,7 +834,7 @@ static int create_udp_socket();
* rgerhards 2005-10-24
*/
-/* END Acess functions for the struct filed */
+/* END Access functions for the struct filed */
/* Code for handling allowed/disallowed senders
*/
@@ -1266,12 +1279,10 @@ static void TCPSendSetStatus(struct filed *f, enum TCPSendStatus iNewState)
/* there can potentially be a race condition, so guard by mutex */
# ifdef USE_PTHREADS
pthread_mutex_lock(&f->f_un.f_forw.mtxTCPSend);
-dprintf("SetStats aquired mutex\n");
# endif
f->f_un.f_forw.status = iNewState;
# ifdef USE_PTHREADS
pthread_mutex_unlock(&f->f_un.f_forw.mtxTCPSend);
-dprintf("SetStats freed mutex\n");
# endif
}
@@ -3008,6 +3019,19 @@ int main(int argc, char **argv)
if (myPid != ppid)
kill (ppid, SIGTERM);
#endif
+ /* END OF INTIALIZATION
+ * ... but keep in mind that we might do a restart and thus init() might
+ * be called again. If that happens, we must shut down all active threads,
+ * do the init() and then restart things.
+ * rgerhards, 2005-10-24
+ */
+#ifdef USE_PTHREADS
+ /* start up worker thread */
+ { int i;
+ i = pthread_create(&thrdWorker, NULL, singleWorker, NULL);
+ printf("worker thread started with state %d\n", i);
+ }
+#endif
/* Main loop begins here. */
for (;;) {
@@ -3862,9 +3886,9 @@ int shouldProcessThisMessage(struct filed *f, struct msg *pMsg)
* See comment dated 2005-10-13 in logmsg() on multithreading.
* rgerhards, 2005-10-13
*/
-static void processMsg(struct msg *pMsg, int flags)
+static void processMsg(struct msg *pMsg)
{
- register struct filed *f;
+ struct filed *f;
assert(pMsg != NULL);
@@ -3883,7 +3907,7 @@ static void processMsg(struct msg *pMsg, int flags)
if (f->f_file >= 0) {
untty();
f->f_pMsg = MsgAddRef(pMsg); /* is expected here... */
- fprintlog(f, flags);
+ fprintlog(f);
MsgDestruct(pMsg);
(void) close(f->f_file);
f->f_file = -1;
@@ -3902,7 +3926,7 @@ static void processMsg(struct msg *pMsg, int flags)
if (f->f_file >= 0) {
untty();
f->f_pMsg = MsgAddRef(pMsg); /* is expected here... */
- fprintlog(f, flags);
+ fprintlog(f);
MsgDestruct(pMsg);
(void) close(f->f_file);
f->f_file = -1;
@@ -3944,31 +3968,26 @@ static void processMsg(struct msg *pMsg, int flags)
break; /* that's it for this message ;) */
}
- if (f->f_type == F_CONSOLE && (flags & IGN_CONS))
- continue;
-
/* don't output marks to recently written files */
- if ((flags & MARK) && (now - f->f_time) < MarkInterval / 2)
+ if ((pMsg->msgFlags & MARK) && (now - f->f_time) < MarkInterval / 2)
continue;
/*
* suppress duplicate lines to this file
*/
- if ((flags & MARK) == 0 && getMSGLen(pMsg) == getMSGLen(f->f_pMsg) &&
+ if ((pMsg->msgFlags & MARK) == 0 && getMSGLen(pMsg) == getMSGLen(f->f_pMsg) &&
!strcmp(getMSG(pMsg), getMSG(f->f_pMsg)) &&
!strcmp(getHOSTNAME(pMsg), getHOSTNAME(f->f_pMsg))) {
f->f_prevcount++;
dprintf("msg repeated %d times, %ld sec of %d.\n",
f->f_prevcount, now - f->f_time,
repeatinterval[f->f_repeatcount]);
- /*
- * If domark would have logged this by now,
+ /* If domark would have logged this by now,
* flush it now (so we don't hold isolated messages),
- * but back off so we'll flush less often
- * in the future.
+ * but back off so we'll flush less often in the future.
*/
if (now > REPEATTIME(f)) {
- fprintlog(f, flags);
+ fprintlog(f);
BACKOFF(f);
}
} else {
@@ -3980,12 +3999,73 @@ static void processMsg(struct msg *pMsg, int flags)
MsgDestruct(f->f_pMsg);
f->f_pMsg = MsgAddRef(pMsg);
/* call the output driver */
- fprintlog(f, flags);
+ fprintlog(f);
}
}
}
+#ifdef USE_PTHREADS
+/* This block contains code that is only present when USE_PTHREADS is
+ * enabled. I plan to move it to some other file, but for the time
+ * being, I include it here because that saves me from the need to
+ * do so many external definitons.
+ * rgerhards, 2005-10-24
+ */
+
+/* The worker thread (so far, we have dual-threading, so only one
+ * worker thread. Having more than one worker requires considerable
+ * additional code review in regard to thread-safety.
+ */
+static void *singleWorker(void *vParam)
+{
+ struct msg *pMsg;
+ while(!bGlblDone) {
+ pthread_mutex_lock(&mtxRunWorker);
+ printf("worker waits for next message\n");
+ pthread_cond_wait(&cndRunWorker, &mtxRunWorker);
+ /* dequeue element (still protected from mutex) */
+ printf("Worker dequeues...\n");
+ pMsg = bufpMsg;
+ pthread_mutex_unlock(&mtxRunWorker);
+ /* do actual processing (the lengthy part, runs in parallel) */
+ dprintf("worker is running\n");
+ processMsg(pMsg);
+ }
+
+ pthread_exit(0);
+}
+
+/* END threads-related code */
+#endif /* #ifdef USE_PTHREADS */
+
+
+/* This method enqueues a message into the the message buffer. It also
+ * the worker thread, so that the message will be processed. If we are
+ * compiled without PTHREADS support, we simply use this method as
+ * an alias for processMsg().
+ * See comment dated 2005-10-13 in logmsg() on multithreading.
+ * rgerhards, 2005-10-24
+ */
+#ifndef USE_PTHREADS
+#define enqueueMsg(x) processMsg((x))
+#else
+static void enqueueMsg(struct msg *pMsg)
+{
+ int iRet;
+ assert(pMsg != NULL);
+
+ iRet = pthread_mutex_lock(&mtxRunWorker);
+ printf("EnqueueMsg waiting on mutex (%d)\n", iRet);
+ bufpMsg = pMsg;
+ /* now activate the worker thread */
+ pthread_mutex_unlock(&mtxRunWorker);
+ iRet = pthread_cond_signal(&cndRunWorker);
+ printf("EnqueueMsg signaled condition (%d)\n", iRet);
+}
+#endif /* #ifndef USE_PTHREADS */
+
+
/*
* Log a message to the appropriate log files, users, etc. based on
* the priority.
@@ -4146,7 +4226,8 @@ void logmsg(int pri, struct msg *pMsg, int flags)
* consumer) to its own method, now called "processMsg()".
*/
- processMsg(pMsg, flags);
+ pMsg->msgFlags = flags;
+ enqueueMsg(pMsg);
#ifndef SYSV
(void) sigsetmask(omask);
@@ -4545,7 +4626,7 @@ again:
* This whole function is probably about to change once we have the
* message abstraction.
*/
-void fprintlog(register struct filed *f, int flags)
+void fprintlog(register struct filed *f)
{
char *msg;
char *psz; /* for shell support */
@@ -4683,14 +4764,6 @@ void fprintlog(register struct filed *f, int flags)
case F_CONSOLE:
f->f_time = now;
-#ifdef UNIXPC
- if (1) {
-#else
- if (flags & IGN_CONS) {
-#endif
- dprintf(" (ignored).\n");
- break;
- }
/* FALLTHROUGH */
case F_TTY:
@@ -5138,7 +5211,7 @@ static void die(int sig)
* I am not sure it is correct as done.
* TODO: verify later!
*/
- fprintlog(f, 0);
+ fprintlog(f);
}
Initialized = was_initialized; /* we restore this so that the logmsgInternal()
@@ -5154,6 +5227,20 @@ static void die(int sig)
logmsgInternal(LOG_SYSLOG|LOG_INFO, buf, LocalHostName, ADDDATE);
}
+#ifdef USE_PTHREADS
+ /* We are now done with all messages, so we need to wake up the
+ * worker thread and then wait for it to finish.
+ */
+ /* TODO: this code is not really working. It's just good as a test
+ * harness!! It dumps *at least* because we have no qeuue!
+ */
+ bGlblDone = 1;
+ pthread_cond_signal(&cndRunWorker);
+ pthread_join(thrdWorker, NULL);
+ pthread_cond_destroy(&cndRunWorker);
+#endif
+
+
/* Free ressources and close MySQL connections */
for (f = Files; f != NULL ; f = f->f_next) {
/* free iovec if it was allocated */
@@ -5449,7 +5536,7 @@ static void init()
* I am not sure it is correct as done.
* TODO: verify later!
*/
- fprintlog(f, 0);
+ fprintlog(f);
/* free iovec if it was allocated */
if(f->f_iov != NULL) {
@@ -6379,7 +6466,6 @@ static rsRetVal cfline(char *line, register struct filed *f)
/* in this case, we also need a mutex... */
# ifdef USE_PTHREADS
pthread_mutex_init(&f->f_un.f_forw.mtxTCPSend, 0);
-dprintf("initializing mutex!\n");
# endif
} else {
f->f_un.f_forw.protocol = FORW_UDP;