initial creation of wti class (worker implementation missing)

2 files changed, 365 insertions, 0 deletions

diff --git a/wti.c b/wti.c
new file mode 100644
index 00000000..ac324bcc
--- /dev/null
+++ b/wti.c
@@ -0,0 +1,310 @@
+/* wti.c
+ *
+ * This file implements the worker thread instance (wti) class.
+ * 
+ * File begun on 2008-01-20 by RGerhards based on functions from the 
+ * previous queue object class (the wti functions have been extracted)
+ *
+ * There is some in-depth documentation available in doc/dev_queue.html
+ * (and in the web doc set on http://www.rsyslog.com/doc). Be sure to read it
+ * if you are getting aquainted to the object.
+ *
+ * Copyright 2008 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.
+ */
+#include "config.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+#include <signal.h>
+#include <pthread.h>
+#include <errno.h>
+
+#include "rsyslog.h"
+#include "syslogd.h"
+#include "stringbuf.h"
+#include "srUtils.h"
+#include "wtp.h"
+#include "wti.h"
+#include "obj.h"
+
+/* static data */
+DEFobjStaticHelpers
+
+/* forward-definitions */
+static void *wtiWorker(void *arg);
+
+/* methods */
+
+/* get the header for debug messages
+ * The caller must NOT free or otherwise modify the returned string!
+ */
+static inline uchar *
+wtiGetDbgHdr(wti_t *pThis)
+{
+	ISOBJ_TYPE_assert(pThis, wti);
+
+	if(pThis->pszDbgHdr == NULL)
+		return (uchar*) "wti"; /* should not normally happen */
+	else
+		return pThis->pszDbgHdr;
+}
+
+
+/* get the current worker state. For simplicity and speed, we have
+ * NOT used our regular calling interface this time. I hope that won't
+ * bite in the long term... -- rgerhards, 2008-01-17
+ */
+static inline qWrkCmd_t
+wtiGetState(wti_t *pThis)
+{
+	ISOBJ_TYPE_assert(pThis, wti);
+	// TODO: lock mutex?
+	return pThis->tCurrCmd;
+}
+
+
+
+/* send a command to a specific thread
+ * bActiveOnly specifies if the command should be sent only when the worker is
+ * in an active state. -- rgerhards, 2008-01-20
+ */
+rsRetVal
+wtiSetState(wti_t *pThis, qWrkCmd_t tCmd, bActiveOnly)
+{
+	DEFiRet;
+	DEFVARS_mutex_cancelsafeLock;
+	int iState;
+
+	ISOBJ_TYPE_assert(pThis, wti);
+	assert(tCmd <= eWRKTHRD_SHUTDOWN_IMMEDIATE);
+
+	mutex_cancelsafe_lock(&pThis->mut);
+
+	/* all worker states must be followed sequentially, only termination can be set in any state */
+	if(   (bActiveOnly && (pThis->tCurrCmd < eWRKTHRD_RUN_CREATED))
+	   || (pThis->tCurrCmd > tCmd && tCmd != eWRKTHRD_TERMINATING)) {
+		dbgprintf("%s: command %d can not be accepted in current %d processing state - ignored\n",
+			  wtiGetDbgHdr(pThis), tCmd, pThis->tCurrCmd);
+	} else {
+		dbgprintf("%s: receiving command %d\n", wtiGetDbgHdr(pThis), tCmd);
+		switch(tCmd) {
+			case eWRKTHRD_RUN_CREATED:
+				assert(pThis->tCurrCmd < eWRKTHRD_RUN_CREATED);
+				iState = pthread_create(&(pThis->thrdID), NULL, wtiWorker, (void*) pThis);
+				dbgprintf("wti: Worker thread %s, started with state %d.\n", wtiGetDbgHdr(pThis), iState);
+				break;
+			case eWRKTHRD_TERMINATING:
+				/* TODO: re-enable meaningful debug msg! (via function callback?)
+				dbgprintf("%s: thread terminating with %d entries left in queue, %d workers running.\n",
+					  wtiGetDbgHdr(pThis->pQueue), pThis->pQueue->iQueueSize,
+					  pThis->pQueue->iCurNumWrkThrd);
+				*/
+				dbgprintf("%s: worker terminating\n", wtiGetDbgHdr(pThis));
+				break;
+			case eWRKTHRD_RUNNING:
+				pthread_cond_signal(&pThis->condInitDone);
+				break;
+			/* these cases just to satisfy the compiler, we do (yet) not act an them: */
+			case eWRKTHRD_STOPPED:
+			case eWRKTHRD_RUN_INIT:
+			case eWRKTHRD_SHUTDOWN:
+			case eWRKTHRD_SHUTDOWN_IMMEDIATE:
+				/* DO NOTHING */
+				break;
+		}
+		pThis->tCurrCmd = tCmd; /* apply the new state */
+	}
+
+	mutex_cancelsafe_unlock(&pThis->mut);
+	return iRet;
+}
+
+
+/* Destructor */
+rsRetVal wtiDestruct(wti_t **ppThis)
+{
+	DEFiRet;
+	wti_t *pThis;
+	int iCancelStateSave;
+
+	assert(ppThis != NULL);
+	pThis = *ppThis;
+	ISOBJ_TYPE_assert(pThis, wti);
+
+	/* we can not be canceled, that would have a myriad of side-effects */
+	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave);
+
+	/* actual destruction */
+	pthread_cond_destroy(&pThis->condInitDone);
+	pthread_mutex_destroy(&pThis->mut);
+
+	if(pThis->pszDbgHdr != NULL)
+		free(pThis->pszDbgHdr);
+
+	/* and finally delete the queue objet itself */
+	free(pThis);
+	*ppThis = NULL;
+
+	/* back to normal */
+	pthread_setcancelstate(iCancelStateSave, NULL);
+
+	return iRet;
+}
+
+
+/* Standard-Constructor for the wti object
+ */
+BEGINobjConstruct(wti) /* be sure to specify the object type also in END macro! */
+	pthread_cond_init(&pThis->condInitDone, NULL);
+	pthread_mutex_init(&pThis->mut, NULL);
+ENDobjConstruct(wti)
+
+
+/* Construction finalizer
+ * rgerhards, 2008-01-17
+ */
+rsRetVal
+wtiConstructFinalize(wti_t *pThis)
+{
+	ISOBJ_TYPE_assert(pThis, wti);
+
+	dbgprintf("%s: finalizing construction of worker instance data\n", wtiGetDbgHdr(pThis));
+
+	/* initialize our thread instance descriptor */
+	pThis->pUsr = NULL;
+	pThis->tCurrCmd = eWRKTHRD_STOPPED;
+
+	return RS_RET_OK;
+}
+
+
+/* Waits until the specified worker thread 
+ * changed to full running state (aka has started up).
+ * rgerhards, 2008-01-17
+ */
+static inline rsRetVal
+wtiWaitStartup(wti_t *pThis)
+{
+	DEFVARS_mutex_cancelsafeLock;
+	ISOBJ_TYPE_assert(pThis, wti);
+
+	mutex_cancelsafe_lock(&pThis->mut);
+	if((pThis->tCurrCmd == eWRKTHRD_RUN_CREATED) || (pThis->tCurrCmd == eWRKTHRD_RUN_CREATED)) {
+		dbgprintf("wti: waiting on worker thread %s startup\n", wtiGetDbgHdr(pThis));
+		pthread_cond_wait(&pThis->condInitDone, &pThis->mut);
+dbgprintf("worker startup done!\n");
+	}
+	mutex_cancelsafe_unlock(&pThis->mut);
+
+	return RS_RET_OK;
+}
+
+
+/* join a specific worker thread
+ * we do not lock the mutex, because join will sync anyways...
+ */
+rsRetVal
+wtiJoinThrd(wti_t *pThis)
+{
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, wti);
+	dbgprintf("wti: waiting for worker %s termination, current state %d\n", wtiGetDbgHdr(pThis), pThis->tCurrCmd);
+	pthread_join(pThis->thrdID, NULL);
+	wtiSetState(pThis, eWRKTHRD_STOPPED); /* back to virgin... */
+	pThis->thrdID = 0; /* invalidate the thread ID so that we do not accidently find reused ones */
+	dbgprintf("wti: worker %s has stopped\n", wtiGetDbgHdr(pThis));
+
+	return iRet;
+}
+
+
+static void *
+wtiWorker(void *arg)
+{
+	wti_t *pThis = (wti_t*) arg;
+
+	ISOBJ_TYPE_assert(pThis, wti);
+
+	// TODO: add logic!
+	//
+	pthread_exit(0);
+}
+
+/* Starts a worker thread (on a specific index [i]!)
+ */
+rsRetVal
+wtiStart(wti_t *pThis)
+{
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, wti);
+	wtiSetState(pThis, eWRKTHRD_RUN_CREATED);
+
+	return iRet;
+}
+
+
+/* some simple object access methods */
+DEFpropSetMeth(wti, toShutdown, int);
+
+/* set the debug header message
+ * The passed-in string is duplicated. So if the caller does not need
+ * it any longer, it must free it. Must be called only before object is finalized.
+ * rgerhards, 2008-01-09
+ */
+rsRetVal
+wtiSetDbgHdr(wti_t *pThis, uchar *pszMsg, size_t lenMsg)
+{
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, wti);
+	assert(pszMsg != NULL);
+	
+	if(lenMsg < 1)
+		ABORT_FINALIZE(RS_RET_PARAM_ERROR);
+
+	if(pThis->pszDbgHdr != NULL) {
+		free(pThis->pszDbgHdr);
+		pThis->pszDbgHdr = NULL;
+	}
+
+	if((pThis->pszDbgHdr = malloc(sizeof(uchar) * lenMsg + 1)) == NULL)
+		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
+
+	memcpy(pThis->pszDbgHdr, pszMsg, lenMsg + 1); /* always think about the \0! */
+
+finalize_it:
+	return iRet;
+}
+
+
+/* Initialize the wti class. Must be called as the very first method
+ * before anything else is called inside this class.
+ * rgerhards, 2008-01-09
+ */
+BEGINObjClassInit(wti, 1) /* one is the object version (most important for persisting) */
+ENDObjClassInit(queue)
+
+/*
+ * vi:set ai:
+ */
diff --git a/wti.h b/wti.h
new file mode 100644
index 00000000..fbf1f150
--- /dev/null
+++ b/wti.h
@@ -0,0 +1,55 @@
+/* Definition of the worker thread instance (wti) class.
+ *
+ * Copyright 2008 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 WTI_H_INCLUDED
+#define WTI_H_INCLUDED
+
+#include <pthread.h>
+#include "wtp.h"
+#include "obj.h"
+
+/* the worker thread instance class */
+typedef struct wti_s {
+	BEGINobjInstance;
+	pthread_t thrdID;  /* thread ID */
+	qWrkCmd_t tCurrCmd; /* current command to be carried out by worker */
+	obj_t *pUsr;		/* current user object being processed (or NULL if none) */
+	wtp_t *pWtp; /* my worker thread pool (important if only the work thread instance is passed! */
+	int toShutdown;	/* shutdown timeout, used when idle */
+	pthread_cond_t condInitDone; /* signaled when the thread startup is done (once per thread existance) */
+	pthread_mutex_t mut;
+	uchar *pszDbgHdr;	/* header string for debug messages */
+} wti_t;
+
+/* some symbolic constants for easier reference */
+
+
+/* prototypes */
+rsRetVal wtiConstruct(wti_t **ppThis);
+rsRetVal wtiConstructFinalize(wti_t *pThis);
+rsRetVal wtiDestruct(wti_t **ppThis);
+PROTOTYPEObjClassInit(wti);
+PROTOTYPEpropSetMeth(wti, pszDbgHdr, uchar*);
+PROTOTYPEpropSetMeth(wti, toShutdown, int);
+#define wtiGetID(pThis) ((unsigned long) pThis)
+
+#endif /* #ifndef WTI_H_INCLUDED */