From 40fffde2b6a36ba12388b89d422104c258a667f7 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Fri, 2 Nov 2012 18:55:53 +0100 Subject: generate disk .qi file once at queue construction ... instead of each time a file write happens. In some situations (very frequent sync), this can probably be a big performane win. --- runtime/queue.h | 2 ++ 1 file changed, 2 insertions(+) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index edb770c6..f9d6067f 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -135,6 +135,8 @@ struct queue_s { size_t lenSpoolDir; uchar *pszFilePrefix; size_t lenFilePrefix; + uchar *pszQIFNam; /* full .qi file name, based on parts above */ + size_t lenQIFNam; 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 */ -- cgit v1.2.3 From feddb2cc8fa213725a14e556d0366aef8d3a4339 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Sat, 3 Nov 2012 11:40:13 +0100 Subject: queue: change gerenic queue pUsr ptr to be action_t this was always action_t, but the initial design was more generic. We are making it specific now in order to gain better performance (after all, we did not need the generic engine in the past 8 years...) --- 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 f9d6067f..a5a020ae 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -71,7 +71,7 @@ struct queue_s { int iMinMsgsPerWrkr;/* minimum nbr of msgs per worker thread, if more, a new worker is started until max wrkrs */ wtp_t *pWtpDA; wtp_t *pWtpReg; - void *pUsr; /* a global, user-supplied pointer. Is passed back to consumer. */ + action_t *pAction; /* for action queues, ptr to action object; for main queues unused */ int iUpdsSincePersist;/* nbr of queue updates since the last persist call */ int iPersistUpdCnt; /* persits queue info after this nbr of updates - 0 -> persist only on shutdown */ sbool bSyncQueueFiles;/* if working with files, sync them after each write? */ @@ -215,7 +215,7 @@ PROTOTYPEpropSetMeth(qqueue, iDiscardMrk, int); PROTOTYPEpropSetMeth(qqueue, iDiscardSeverity, int); PROTOTYPEpropSetMeth(qqueue, iMinMsgsPerWrkr, int); PROTOTYPEpropSetMeth(qqueue, bSaveOnShutdown, int); -PROTOTYPEpropSetMeth(qqueue, pUsr, void*); +PROTOTYPEpropSetMeth(qqueue, pAction, action_t*); PROTOTYPEpropSetMeth(qqueue, iDeqSlowdown, int); PROTOTYPEpropSetMeth(qqueue, sizeOnDiskMax, int64); PROTOTYPEpropSetMeth(qqueue, iDeqBatchSize, int); -- cgit v1.2.3 From c55e0a5a06e69106bc346057dd61dcb98688a4aa Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Sat, 3 Nov 2012 12:32:50 +0100 Subject: queue: change generic msg ptr (pUsr) to be of msg_t type --- runtime/queue.h | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index a5a020ae..e6ccdcdb 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -51,7 +51,7 @@ typedef enum { /* list member definition for linked list types of queues: */ typedef struct qLinkedList_S { struct qLinkedList_S *pNext; - void *pUsr; + msg_t *pMsg; } qLinkedList_t; @@ -111,8 +111,8 @@ struct queue_s { /* type-specific handlers (set during construction) */ rsRetVal (*qConstruct)(struct queue_s *pThis); rsRetVal (*qDestruct)(struct queue_s *pThis); - rsRetVal (*qAdd)(struct queue_s *pThis, void *pUsr); - rsRetVal (*qDeq)(struct queue_s *pThis, void **ppUsr); + rsRetVal (*qAdd)(struct queue_s *pThis, msg_t *pMsg); + rsRetVal (*qDeq)(struct queue_s *pThis, msg_t **ppMsg); rsRetVal (*qDel)(struct queue_s *pThis); /* end type-specific handler */ /* public entry points (set during construction, permit to set best algorithm for params selected) */ @@ -186,8 +186,8 @@ struct queue_s { /* prototypes */ rsRetVal qqueueDestruct(qqueue_t **ppThis); -rsRetVal qqueueEnqObjDirect(qqueue_t *pThis, void *pUsr); -rsRetVal qqueueEnqObj(qqueue_t *pThis, flowControl_t flwCtlType, void *pUsr); +rsRetVal qqueueEnqMsgDirect(qqueue_t *pThis, msg_t *pMsg); +rsRetVal qqueueEnqMsg(qqueue_t *pThis, flowControl_t flwCtlType, msg_t *pMsg); rsRetVal qqueueStart(qqueue_t *pThis); rsRetVal qqueueSetMaxFileSize(qqueue_t *pThis, size_t iMaxFileSize); rsRetVal qqueueSetFilePrefix(qqueue_t *pThis, uchar *pszPrefix, size_t iLenPrefix); -- cgit v1.2.3 From c563914d6f96efc1c4da02a7f49409297b20f656 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Mon, 19 Nov 2012 16:46:41 +0100 Subject: queue: file delete stream does no longer do real io This stream is primarily used for state tracking, and has been modified to do just that. This results in considerable less io being done and the respective speedup. --- runtime/queue.h | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) (limited to 'runtime/queue.h') diff --git a/runtime/queue.h b/runtime/queue.h index e6ccdcdb..7db2d90d 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -147,7 +147,8 @@ struct queue_s { struct queue_s *pqParent;/* pointer to the parent (if this is a child queue) */ int bDAEnqOnly; /* EnqOnly setting for DA queue */ /* now follow queueing mode specific data elements */ - union { /* different data elements based on queue type (qType) */ + //union { /* different data elements based on queue type (qType) */ + struct { /* different data elements based on queue type (qType) */ struct { long deqhead, head, tail; void** pBuf; /* the queued user data structure */ @@ -159,7 +160,9 @@ struct queue_s { } linklist; struct { int64 sizeOnDisk; /* current amount of disk space used */ - int64 bytesRead; /* number of bytes read from current (undeleted!) file */ + int64 deqOffs; /* offset after dequeue batch - used for file deleter */ + int deqFileNumIn; /* same for the circular file numbers, mainly for */ + int deqFileNumOut;/* deleting finished files */ strm_t *pWrite; /* current file to be written */ strm_t *pReadDeq; /* current file for dequeueing */ strm_t *pReadDel; /* current file for deleting */ -- cgit v1.2.3