diff options
Diffstat (limited to 'runtime')
| -rw-r--r-- | runtime/queue.c | 38 |
1 files changed, 30 insertions, 8 deletions
diff --git a/runtime/queue.c b/runtime/queue.c index c594af7c..a6928910 100644 --- a/runtime/queue.c +++ b/runtime/queue.c @@ -87,6 +87,7 @@ static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis); static rsRetVal qConstructDirect(qqueue_t __attribute__((unused)) *pThis); static rsRetVal qDelDirect(qqueue_t __attribute__((unused)) *pThis); static rsRetVal qDestructDisk(qqueue_t *pThis); +rsRetVal qqueueSetSpoolDir(qqueue_t *pThis, uchar *pszSpoolDir, int lenSpoolDir); /* some constants for queuePersist () */ #define QUEUE_CHECKPOINT 1 @@ -418,6 +419,7 @@ StartDA(qqueue_t *pThis) CHKiRet(qqueueSetiDeqSlowdown(pThis->pqDA, pThis->iDeqSlowdown)); CHKiRet(qqueueSetMaxFileSize(pThis->pqDA, pThis->iMaxFileSize)); CHKiRet(qqueueSetFilePrefix(pThis->pqDA, pThis->pszFilePrefix, pThis->lenFilePrefix)); + CHKiRet(qqueueSetSpoolDir(pThis->pqDA, pThis->pszSpoolDir, pThis->lenSpoolDir)); CHKiRet(qqueueSetiPersistUpdCnt(pThis->pqDA, pThis->iPersistUpdCnt)); CHKiRet(qqueueSetbSyncQueueFiles(pThis->pqDA, pThis->bSyncQueueFiles)); CHKiRet(qqueueSettoActShutdown(pThis->pqDA, pThis->toActShutdown)); @@ -730,7 +732,7 @@ qqueueLoadPersStrmInfoFixup(strm_t *pStrm, qqueue_t __attribute__((unused)) *pTh DEFiRet; ISOBJ_TYPE_assert(pStrm, strm); ISOBJ_TYPE_assert(pThis, qqueue); - CHKiRet(strm.SetDir(pStrm, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir()))); + CHKiRet(strm.SetDir(pStrm, pThis->pszSpoolDir, pThis->lenSpoolDir)); finalize_it: RETiRet; } @@ -837,7 +839,7 @@ static rsRetVal qConstructDisk(qqueue_t *pThis) } else { CHKiRet(strm.Construct(&pThis->tVars.disk.pWrite)); CHKiRet(strm.SetbSync(pThis->tVars.disk.pWrite, pThis->bSyncQueueFiles)); - CHKiRet(strm.SetDir(pThis->tVars.disk.pWrite, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir()))); + CHKiRet(strm.SetDir(pThis->tVars.disk.pWrite, pThis->pszSpoolDir, pThis->lenSpoolDir)); CHKiRet(strm.SetiMaxFiles(pThis->tVars.disk.pWrite, 10000000)); CHKiRet(strm.SettOperationsMode(pThis->tVars.disk.pWrite, STREAMMODE_WRITE)); CHKiRet(strm.SetsType(pThis->tVars.disk.pWrite, STREAMTYPE_FILE_CIRCULAR)); @@ -849,7 +851,7 @@ static rsRetVal qConstructDisk(qqueue_t *pThis) CHKiRet(strm.Construct(&pThis->tVars.disk.pReadDeq)); CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pReadDeq, 0)); - CHKiRet(strm.SetDir(pThis->tVars.disk.pReadDeq, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir()))); + CHKiRet(strm.SetDir(pThis->tVars.disk.pReadDeq, pThis->pszSpoolDir, pThis->lenSpoolDir)); CHKiRet(strm.SetiMaxFiles(pThis->tVars.disk.pReadDeq, 10000000)); CHKiRet(strm.SettOperationsMode(pThis->tVars.disk.pReadDeq, STREAMMODE_READ)); CHKiRet(strm.SetsType(pThis->tVars.disk.pReadDeq, STREAMTYPE_FILE_CIRCULAR)); @@ -862,7 +864,7 @@ static rsRetVal qConstructDisk(qqueue_t *pThis) CHKiRet(strm.Construct(&pThis->tVars.disk.pReadDel)); CHKiRet(strm.SetbSync(pThis->tVars.disk.pReadDel, pThis->bSyncQueueFiles)); CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pReadDel, 1)); - CHKiRet(strm.SetDir(pThis->tVars.disk.pReadDel, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir()))); + CHKiRet(strm.SetDir(pThis->tVars.disk.pReadDel, pThis->pszSpoolDir, pThis->lenSpoolDir)); CHKiRet(strm.SetiMaxFiles(pThis->tVars.disk.pReadDel, 10000000)); CHKiRet(strm.SettOperationsMode(pThis->tVars.disk.pReadDel, STREAMMODE_READ)); CHKiRet(strm.SetsType(pThis->tVars.disk.pReadDel, STREAMTYPE_FILE_CIRCULAR)); @@ -2091,7 +2093,7 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */ pThis->iNumWorkerThreads = 1; /* we need exactly one worker */ /* pre-construct file name for .qi file */ pThis->lenQIFNam = snprintf((char*)pszQIFNam, sizeof(pszQIFNam) / sizeof(uchar), - "%s/%s.qi", (char*) glbl.GetWorkDir(), (char*)pThis->pszFilePrefix); + "%s/%s.qi", (char*) pThis->pszSpoolDir, (char*)pThis->pszFilePrefix); pThis->pszQIFNam = ustrdup(pszQIFNam); DBGOPRINT((obj_t*) pThis, ".qi file name is '%s', len %d\n", pThis->pszQIFNam, (int) pThis->lenQIFNam); @@ -2183,10 +2185,12 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */ pThis->iFullDlyMrk = wrk; } - DBGOPRINT((obj_t*) pThis, "type %d, enq-only %d, disk assisted %d, maxFileSz %lld, maxQSize %d, lqsize %d, pqsize %d, child %d, " - "full delay %d, light delay %d, deq batch size %d starting, high wtrmrk %d, low wtrmrk %d, " + DBGOPRINT((obj_t*) pThis, "type %d, enq-only %d, disk assisted %d, spoolDir '%s', maxFileSz %lld, " + "maxQSize %d, lqsize %d, pqsize %d, child %d, full delay %d, " + "light delay %d, deq batch size %d starting, high wtrmrk %d, low wtrmrk %d, " "max wrkr %d, min msgs f. wrkr %d\n", - pThis->qType, pThis->bEnqOnly, pThis->bIsDA, pThis->iMaxFileSize, pThis->iMaxQueueSize, + pThis->qType, pThis->bEnqOnly, pThis->bIsDA, pThis->pszSpoolDir, + pThis->iMaxFileSize, pThis->iMaxQueueSize, getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis), pThis->pqParent == NULL ? 0 : 1, pThis->iFullDlyMrk, pThis->iLightDlyMrk, pThis->iDeqBatchSize, pThis->iHighWtrMrk, pThis->iLowWtrMrk, @@ -2497,6 +2501,24 @@ CODESTARTobjDestruct(qqueue) ENDobjDestruct(qqueue) +/* set the queue's spool directory. The directory MUST NOT be NULL. + * The passed-in string is duplicated. So if the caller does not need + * it any longer, it must free it. + */ +rsRetVal +qqueueSetSpoolDir(qqueue_t *pThis, uchar *pszSpoolDir, int lenSpoolDir) +{ + DEFiRet; + + free(pThis->pszSpoolDir); + CHKmalloc(pThis->pszSpoolDir = ustrdup(pszSpoolDir)); + pThis->lenSpoolDir = lenSpoolDir; + +finalize_it: + RETiRet; +} + + /* set the queue's file prefix * The passed-in string is duplicated. So if the caller does not need * it any longer, it must free it. |