From 7667845bd72b6f92eabc975318a4f288a77f2630 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Wed, 22 Apr 2009 15:06:45 +0200 Subject: first attempt at dequeueing multiple batches inside the queue ... but this code has serious problems when terminating the queue, also it is far from being optimal. I will commit a series of patches (hopefully) as I am on the path to the final implementation. --- runtime/queue.h | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index a267862d..7ecb9294 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -54,6 +54,7 @@ typedef struct qWrkThrd_s { pthread_mutex_t mut; } qWrkThrd_t; /* type for queue worker threads */ + /* the queue object */ typedef struct queue_s { BEGINobjInstance; @@ -84,6 +85,7 @@ typedef struct queue_s { int toActShutdown; /* timeout for long-running action shutdown in ms */ int toWrkShutdown; /* timeout for idle workers in ms, -1 means indefinite (0 is immediate) */ int toEnq; /* enqueue timeout */ + int iDeqMaxAtOnce; /* max number of elements that shall be dequeued at once */ /* rate limiting settings (will be expanded) */ int iDeqSlowdown; /* slow down dequeue by specified nbr of microseconds */ /* end rate limiting */ @@ -97,7 +99,7 @@ typedef struct queue_s { * applied to detect user configuration errors (and tell me how should we detect what * the user really wanted...). -- rgerhards, 2008-04-02 */ - /* ane dequeue time window */ + /* end dequeue time window */ rsRetVal (*pConsumer)(void *,void*); /* 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 that was dequeued (actual sample: for actions, arg1 is the pAction and arg2 is pointer -- cgit v1.2.3 From e4b3f6d287d74b34d27b4e296c33cb3f1294a58c Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Wed, 22 Apr 2009 16:39:58 +0200 Subject: now batches are handed down to the actual consumer ... but the action consumer does not do anything really intelligent with them. But the DA consumer is already done, as is the main message queue consumer. --- runtime/queue.h | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index 7ecb9294..4fb57d07 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -100,11 +100,10 @@ typedef struct queue_s { * the user really wanted...). -- rgerhards, 2008-04-02 */ /* end dequeue time window */ - rsRetVal (*pConsumer)(void *,void*); /* user-supplied consumer function for dequeued messages */ + rsRetVal (*pConsumer)(void *,aUsrp_t*); /* 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 that was dequeued (actual sample: for actions, arg1 is the pAction and arg2 is pointer - * to message) - * rgerhards, 2008-01-28 + * 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) */ /* type-specific handlers (set during construction) */ rsRetVal (*qConstruct)(struct queue_s *pThis); @@ -185,7 +184,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*,void*)); + int iMaxQueueSize, rsRetVal (*pConsumer)(void*,aUsrp_t*)); PROTOTYPEObjClassInit(qqueue); PROTOTYPEpropSetMeth(qqueue, iPersistUpdCnt, int); PROTOTYPEpropSetMeth(qqueue, iDeqtWinFromHr, int); -- cgit v1.2.3 From 5c0aeae8ab1f344a022d586dc26c5d78203f7e0b Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Thu, 23 Apr 2009 12:50:07 +0200 Subject: added $MainMsgQueueDequeueBatchSize and $ActionQueueDequeueBatchSize configuration directives --- runtime/queue.h | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index 4fb57d07..8a60254b 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -85,7 +85,7 @@ typedef struct queue_s { int toActShutdown; /* timeout for long-running action shutdown in ms */ int toWrkShutdown; /* timeout for idle workers in ms, -1 means indefinite (0 is immediate) */ int toEnq; /* enqueue timeout */ - int iDeqMaxAtOnce; /* max number of elements that shall be dequeued at once */ + int iDeqBatchSize; /* max number of elements that shall be dequeued at once */ /* rate limiting settings (will be expanded) */ int iDeqSlowdown; /* slow down dequeue by specified nbr of microseconds */ /* end rate limiting */ @@ -202,6 +202,7 @@ PROTOTYPEpropSetMeth(qqueue, bSaveOnShutdown, int); PROTOTYPEpropSetMeth(qqueue, pUsr, void*); PROTOTYPEpropSetMeth(qqueue, iDeqSlowdown, int); PROTOTYPEpropSetMeth(qqueue, sizeOnDiskMax, int64); +PROTOTYPEpropSetMeth(qqueue, iDeqBatchSize, int); #define qqueueGetID(pThis) ((unsigned long) pThis) #endif /* #ifndef QUEUE_H_INCLUDED */ -- cgit v1.2.3 From bb79e96dc300fa5a2182e7c047afb3b15c5dc870 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 12 May 2009 15:27:40 +0200 Subject: moving to a cleaner implementation of batches ... now that we know what we need from a theoretical POV. --- runtime/queue.h | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index 8a60254b..4a5f16a1 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -27,6 +27,7 @@ #include #include "obj.h" #include "wtp.h" +#include "batch.h" #include "stream.h" /* queue types */ @@ -100,7 +101,7 @@ typedef struct queue_s { * the user really wanted...). -- rgerhards, 2008-04-02 */ /* end dequeue time window */ - rsRetVal (*pConsumer)(void *,aUsrp_t*); /* user-supplied consumer function for dequeued messages */ + rsRetVal (*pConsumer)(void *,batch_t*); /* 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) @@ -184,7 +185,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*,aUsrp_t*)); + int iMaxQueueSize, rsRetVal (*pConsumer)(void*,batch_t*)); PROTOTYPEObjClassInit(qqueue); PROTOTYPEpropSetMeth(qqueue, iPersistUpdCnt, int); PROTOTYPEpropSetMeth(qqueue, iDeqtWinFromHr, int); -- cgit v1.2.3 From 93f873277bfe5ebb309ff5e92f5dc7244ebd9f1a Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Mon, 18 May 2009 17:28:34 +0200 Subject: t-delete list implemented, queue store drivers updated... ... on the way to the ultra-reliable queue modes (redesign doc). This version does not really work, but is a good commit point. Next comes queue size calculation. DA mode does not yet work. --- runtime/queue.h | 25 ++++++++++++++++++++----- 1 file changed, 20 insertions(+), 5 deletions(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index 4a5f16a1..00cee419 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -30,6 +30,15 @@ #include "batch.h" #include "stream.h" +/* support for the toDelete list */ +typedef struct toDeleteLst_s toDeleteLst_t; +struct toDeleteLst_s { + qDeqID deqID; + int nElem; + struct toDeleteLst_s *pNext; +}; + + /* queue types */ typedef enum { QUEUETYPE_FIXED_ARRAY = 0,/* a simple queue made out of a fixed (initially malloced) array fast but memoryhog */ @@ -85,6 +94,7 @@ typedef struct queue_s { int toQShutdown; /* timeout for regular queue shutdown in ms */ int toActShutdown; /* timeout for long-running action shutdown in ms */ int toWrkShutdown; /* timeout for idle workers in ms, -1 means indefinite (0 is immediate) */ + toDeleteLst_t *toDeleteLst;/* this queue's to-delete list */ int toEnq; /* enqueue timeout */ int iDeqBatchSize; /* max number of elements that shall be dequeued at once */ /* rate limiting settings (will be expanded) */ @@ -110,7 +120,8 @@ typedef struct queue_s { rsRetVal (*qConstruct)(struct queue_s *pThis); rsRetVal (*qDestruct)(struct queue_s *pThis); rsRetVal (*qAdd)(struct queue_s *pThis, void *pUsr); - rsRetVal (*qDel)(struct queue_s *pThis, void **ppUsr); + rsRetVal (*qDeq)(struct queue_s *pThis, void **ppUsr); + rsRetVal (*qDel)(struct queue_s *pThis); /* end type-specific handler */ /* synchronization variables */ pthread_mutex_t mutThrdMgmt; /* mutex for the queue's thread management */ @@ -134,6 +145,8 @@ typedef struct queue_s { int iNumberFiles; /* how many files make up the queue? */ int64 iMaxFileSize; /* max size for a single queue file */ int64 sizeOnDiskMax; /* maximum size on disk allowed */ + 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 */ @@ -148,18 +161,20 @@ typedef struct queue_s { /* now follow queueing mode specific data elements */ union { /* different data elements based on queue type (qType) */ struct { - long head, tail; + long deqhead, head, tail; void** pBuf; /* the queued user data structure */ } farray; struct { - qLinkedList_t *pRoot; + qLinkedList_t *pDeqRoot; + qLinkedList_t *pDelRoot; qLinkedList_t *pLast; } linklist; struct { int64 sizeOnDisk; /* current amount of disk space used */ int64 bytesRead; /* number of bytes read from current (undeleted!) file */ - strm_t *pWrite; /* current file to be written */ - strm_t *pRead; /* current file to be read */ + strm_t *pWrite; /* current file to be written */ + strm_t *pReadDeq; /* current file for dequeueing */ + strm_t *pReadDel; /* current file for deleting */ } disk; } tVars; } qqueue_t; -- cgit v1.2.3 From fe5bea77ac2533faab3b7b73bc253c4dc7d702bc Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Mon, 18 May 2009 18:39:52 +0200 Subject: removed queue's UngetObj() call ... which is no longer needed thanks to the new queue design. --- 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 00cee419..e47b8762 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -152,12 +152,6 @@ typedef struct queue_s { 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 */ - /* some data elements for the queueUngetObj() functionality. This list should always be short - * and is always kept in memory - */ - qLinkedList_t *pUngetRoot; - qLinkedList_t *pUngetLast; - int iUngottenObjs; /* number of objects currently in the "ungotten" list */ /* now follow queueing mode specific data elements */ union { /* different data elements based on queue type (qType) */ struct { -- cgit v1.2.3 From a4dad2009992d436ba23c2d0a4a43b483aac40fc Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 19 May 2009 11:03:09 +0200 Subject: queue size calculation now based on logical/physical dequeue ... needed to split the old single counter into two. I wouldn't bet that I made some mistakes while doing so, but at least some ad-hoc tests plus the testbench do no longer indicate errors. --- runtime/queue.h | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index e47b8762..92bf8ae5 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -74,7 +74,8 @@ typedef struct queue_s { 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 */ int bQueueInDestruction;/* 1 if queue is in destruction process, 0 otherwise */ - int iQueueSize; /* Current number of elements in the queue */ + int iQueueSize; /* Current number of elements in queue store (some are already logically dequeued!) */ + int nLogDeq; /* number of elements currently logically dequeued */ int iMaxQueueSize; /* how large can the queue grow? */ int iNumWorkerThreads;/* number of worker threads to use */ int iCurNumWrkThrd;/* current number of active worker threads */ -- cgit v1.2.3 From 0cf8e88a348dc574244e4f5c2be26f47e8bfff08 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 19 May 2009 18:58:33 +0200 Subject: solved the intended-discard-during-dequeue issue --- runtime/queue.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index 92bf8ae5..954a7fd4 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -34,7 +34,7 @@ typedef struct toDeleteLst_s toDeleteLst_t; struct toDeleteLst_s { qDeqID deqID; - int nElem; + int nElemDeq; /* numbe of elements that were dequeued and as such must now be discarded */ struct toDeleteLst_s *pNext; }; -- cgit v1.2.3 From aa9426f683fa6af9280bc63050ee0187ba4c57e1 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 26 May 2009 12:43:43 +0200 Subject: solved design issue with queue termination ... and also improved the test suite. There is a design issue in the v3 queue engine that manifested to some serious problems with the new processing mode. However, in v3 shutdown may take eternally if a queue runs in DA mode, is configured to preserve data AND the action fails and retries immediately. There is no cure available for v3, it would require doing much of the work we have done on the new engine. The window of exposure, as one might guess from the description, is very small. That is probably the reason why we have not seen it in practice. --- runtime/queue.h | 1 + 1 file changed, 1 insertion(+) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index 954a7fd4..c1fe597d 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -123,6 +123,7 @@ 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