diff options
Diffstat (limited to 'runtime/queue.c')
| -rw-r--r-- | runtime/queue.c | 17 |
1 files changed, 8 insertions, 9 deletions
diff --git a/runtime/queue.c b/runtime/queue.c index 1ae386e7..5102b0df 100644 --- a/runtime/queue.c +++ b/runtime/queue.c @@ -1649,6 +1649,10 @@ DequeueConsumableElements(qqueue_t *pThis, wti_t *pWti, int *piRemainingQueueSiz nDeleted = pWti->batch.nElemDeq; DeleteProcessedBatch(pThis, &pWti->batch); +//int iii = pthread_mutex_trylock(pThis->mut); +//char errStr[1024]; +//rs_strerror_r(iii, errStr, sizeof(errStr)); +//dbgprintf("DequeueConsumableElemnts mutex locked: %d (16 is EBUSY = OK): %s\n", iii, errStr); nDequeued = nDiscarded = 0; while((iQueueSize = getLogicalQueueSize(pThis)) > 0 && nDequeued < pThis->iDeqBatchSize) { //dbgprintf("DequeueConsumableElements, index %d\n", nDequeued); @@ -2473,15 +2477,6 @@ qqueueEnqObj(qqueue_t *pThis, flowControl_t flowCtlType, void *pUsr) ISOBJ_TYPE_assert(pThis, qqueue); - /* first check if we need to discard this message (which will cause CHKiRet() to exit) - * rgerhards, 2008-10-07: It is OK to do this outside of mutex protection. The queue size - * and bRunsDA parameters may not reflect the correct settings here, but they are - * "good enough" in the sense that they can be used to drive the decision. Valgrind's - * threading tools may point this access to be an error, but this is done - * intentional. I do not see this causes problems to us. - */ - CHKiRet(qqueueChkDiscardMsg(pThis, getPhysicalQueueSize(pThis), pThis->bRunsDA, pUsr)); - /* Please note that this function is not cancel-safe and consequently * sets the calling thread's cancelibility state to PTHREAD_CANCEL_DISABLE * during its execution. If that is not done, race conditions occur if the @@ -2493,6 +2488,10 @@ qqueueEnqObj(qqueue_t *pThis, flowControl_t flowCtlType, void *pUsr) d_pthread_mutex_lock(pThis->mut); } + /* first check if we need to discard this message (which will cause CHKiRet() to exit) + */ + CHKiRet(qqueueChkDiscardMsg(pThis, getPhysicalQueueSize(pThis), pThis->bRunsDA, pUsr)); + /* then check if we need to add an assistance disk queue */ if(pThis->bIsDA) CHKiRet(ChkStrtDA(pThis)); |