From 4d70c9b3e5e480d6dfa1c94506270f1f78e8ef32 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 13 Oct 2009 14:38:45 +0200 Subject: added some debug settings plus improved shutdown sequence ... non-working version! --- runtime/queue.h | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index 73c62b52..74bf2d31 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -59,10 +59,10 @@ typedef struct queue_s { BEGINobjInstance; queueType_t qType; int nLogDeq; /* number of elements currently logically dequeued */ + int bShutdownImmediate; /* should all workers cease processing messages? */ bool bEnqOnly; /* does queue run in enqueue-only mode (1) or not (0)? */ bool bSaveOnShutdown;/* persists everthing on shutdown (if DA!)? 1-yes, 0-no */ bool bQueueStarted; /* has queueStart() been called on this queue? 1-yes, 0-no */ - bool bQueueInDestruction;/* 1 if queue is in destruction process, 0 otherwise */ int iQueueSize; /* Current number of elements in the queue */ int iMaxQueueSize; /* how large can the queue grow? */ int iNumWorkerThreads;/* number of worker threads to use */ @@ -101,10 +101,11 @@ typedef struct queue_s { * the user really wanted...). -- rgerhards, 2008-04-02 */ /* end dequeue time window */ - rsRetVal (*pConsumer)(void *,batch_t*); /* user-supplied consumer function for dequeued messages */ + rsRetVal (*pConsumer)(void *,batch_t*,int*); /* user-supplied consumer function for dequeued messages */ /* calling interface for pConsumer: arg1 is the global user pointer from this structure, arg2 is the * user pointer array that was dequeued (actual sample: for actions, arg1 is the pAction and arg2 - * is pointer to an array of message message pointers) + * is pointer to an array of message message pointers), arg3 is a pointer to an interger which is zero + * during normal operations and one if the consumer must urgently shut down. */ /* type-specific handlers (set during construction) */ rsRetVal (*qConstruct)(struct queue_s *pThis); @@ -185,7 +186,7 @@ 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*,batch_t*)); + int iMaxQueueSize, rsRetVal (*pConsumer)(void*,batch_t*, int*)); PROTOTYPEObjClassInit(qqueue); PROTOTYPEpropSetMeth(qqueue, iPersistUpdCnt, int); PROTOTYPEpropSetMeth(qqueue, bSyncQueueFiles, int); -- cgit v1.2.3 From 90e8475260cf8ac54519b3d964d879489af879f6 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Mon, 19 Oct 2009 09:41:45 +0200 Subject: bugfix: message processing states were not set correctly in all cases however, this had no negative effect, as the message processing state was not evaluated when a batch was deleted, and that was the only case where the state could be wrong. --- runtime/queue.h | 6 ------ 1 file changed, 6 deletions(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index 74bf2d31..338f091b 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -164,12 +164,6 @@ typedef struct queue_s { } tVars; } qqueue_t; -/* some symbolic constants for easier reference */ -#define QUEUE_MODE_ENQDEQ 0 -#define QUEUE_MODE_ENQONLY 1 - -#define QUEUE_IDX_DA_WORKER 0 /* index for the DA worker (fixed) */ -#define QUEUE_PTR_DA_WORKER(x) (&((pThis)->pWrkThrds[0])) /* the define below is an "eternal" timeout for the timeout settings which require a value. * It is one day, which is not really eternal, but comes close to it if we think about -- cgit v1.2.3 From 553d1880d47b57b2f4e023c2017675f010afd9a0 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 27 Oct 2009 10:36:53 +0100 Subject: some cleanup --- runtime/queue.h | 1 - 1 file changed, 1 deletion(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index 338f091b..3b5d7038 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -113,7 +113,6 @@ typedef struct queue_s { rsRetVal (*qAdd)(struct queue_s *pThis, void *pUsr); rsRetVal (*qDeq)(struct queue_s *pThis, void **ppUsr); rsRetVal (*qDel)(struct queue_s *pThis); - rsRetVal (*qUnDeqAll)(struct queue_s *pThis); /* end type-specific handler */ /* synchronization variables */ pthread_mutex_t mutThrdMgmt; /* mutex for the queue's thread management */ -- cgit v1.2.3 From 386b7cd2f2ae6f9ac8e0b9c8b49934398c159ea4 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 27 Oct 2009 10:44:55 +0100 Subject: removed no longer needed flag variable --- runtime/queue.h | 1 - 1 file changed, 1 deletion(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index 3b5d7038..26c57a50 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -138,7 +138,6 @@ typedef struct queue_s { qDeqID deqIDAdd; /* next dequeue ID to use during add to queue store */ qDeqID deqIDDel; /* queue store delete position */ int bIsDA; /* is this queue disk assisted? */ - int bRunsDA; /* is this queue actually *running* disk assisted? */ struct queue_s *pqDA; /* queue for disk-assisted modes */ struct queue_s *pqParent;/* pointer to the parent (if this is a child queue) */ int bDAEnqOnly; /* EnqOnly setting for DA queue */ -- cgit v1.2.3 From 1ada506e2d90377c2475e103340d8986bf8847f9 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 27 Oct 2009 17:31:27 +0100 Subject: added the capability to have ruleset-specific main message queues This offers considerable additional flexibility AND superior performance (in cases where multiple inputs now can avoid lock contention) --- runtime/queue.h | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index 26c57a50..93573dae 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -55,7 +55,7 @@ typedef struct qLinkedList_S { /* the queue object */ -typedef struct queue_s { +struct queue_s { BEGINobjInstance; queueType_t qType; int nLogDeq; /* number of elements currently logically dequeued */ @@ -160,7 +160,7 @@ typedef struct queue_s { strm_t *pReadDel; /* current file for deleting */ } disk; } tVars; -} qqueue_t; +}; /* the define below is an "eternal" timeout for the timeout settings which require a value. -- cgit v1.2.3