diff options
Diffstat (limited to 'runtime/queue.h')
| -rw-r--r-- | runtime/queue.h | 18 |
1 files changed, 12 insertions, 6 deletions
diff --git a/runtime/queue.h b/runtime/queue.h index 93573dae..1c758134 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -60,9 +60,9 @@ struct queue_s { 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 */ + sbool bEnqOnly; /* does queue run in enqueue-only mode (1) or not (0)? */ + sbool bSaveOnShutdown;/* persists everthing on shutdown (if DA!)? 1-yes, 0-no */ + sbool bQueueStarted; /* has queueStart() been called on this queue? 1-yes, 0-no */ 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 */ @@ -73,14 +73,14 @@ struct queue_s { void *pUsr; /* a global, user-supplied pointer. Is passed back to consumer. */ 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 */ - bool bSyncQueueFiles;/* if working with files, sync them after each write? */ + sbool bSyncQueueFiles;/* if working with files, sync them after each write? */ int iHighWtrMrk; /* high water mark for disk-assisted memory queues */ int iLowWtrMrk; /* low water mark for disk-assisted memory queues */ int iDiscardMrk; /* if the queue is above this mark, low-severity messages are discarded */ int iFullDlyMrk; /* if the queue is above this mark, FULL_DELAYable message are put on hold */ int iLightDlyMrk; /* if the queue is above this mark, LIGHT_DELAYable message are put on hold */ int iDiscardSeverity;/* messages of this severity above are discarded on too-full queue */ - bool bNeedDelQIF; /* does the QIF file need to be deleted when queue becomes empty? */ + sbool bNeedDelQIF; /* does the QIF file need to be deleted when queue becomes empty? */ 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) */ @@ -114,6 +114,9 @@ struct queue_s { rsRetVal (*qDeq)(struct queue_s *pThis, void **ppUsr); rsRetVal (*qDel)(struct queue_s *pThis); /* end type-specific handler */ + /* public entry points (set during construction, permit to set best algorithm for params selected) */ + rsRetVal (*MultiEnq)(qqueue_t *pThis, multi_submit_t *pMultiSub); + /* end public entry points */ /* synchronization variables */ pthread_mutex_t mutThrdMgmt; /* mutex for the queue's thread management */ pthread_mutex_t *mut; /* mutex for enqueing and dequeueing messages */ @@ -160,6 +163,8 @@ struct queue_s { strm_t *pReadDel; /* current file for deleting */ } disk; } tVars; + DEF_ATOMIC_HELPER_MUT(mutQueueSize); + DEF_ATOMIC_HELPER_MUT(mutLogDeq); }; @@ -172,13 +177,14 @@ struct queue_s { /* prototypes */ rsRetVal qqueueDestruct(qqueue_t **ppThis); -rsRetVal qqueueMultiEnqObj(qqueue_t *pThis, multi_submit_t *pMultiSub); +rsRetVal qqueueEnqObjDirect(qqueue_t *pThis, void *pUsr); rsRetVal qqueueEnqObj(qqueue_t *pThis, flowControl_t flwCtlType, void *pUsr); 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*)); +rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch); PROTOTYPEObjClassInit(qqueue); PROTOTYPEpropSetMeth(qqueue, iPersistUpdCnt, int); PROTOTYPEpropSetMeth(qqueue, bSyncQueueFiles, int); |