From e3040285dbf0854443bc2443e0de5ac59f6f839e Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Mon, 6 Jul 2009 16:38:09 +0200 Subject: first shot at asynchronous stream writer with timeout capability ... seems to work on quick testing, but needs a far more testing and improvement. Good milestone commit. --- runtime/apc.c | 2 + runtime/debug.h | 4 +- runtime/stream.c | 216 +++++++++++++++++++++++++++++++++++++------------------ runtime/stream.h | 17 ++++- runtime/wtp.c | 6 +- 5 files changed, 171 insertions(+), 74 deletions(-) (limited to 'runtime') diff --git a/runtime/apc.c b/runtime/apc.c index 5919191d..bc330e39 100644 --- a/runtime/apc.c +++ b/runtime/apc.c @@ -335,9 +335,11 @@ CancelApc(apc_id_t id) { DEFVARS_mutexProtection_uncond; + BEGINfunc BEGIN_MTX_PROTECTED_OPERATIONS_UNCOND(&listMutex); deleteApc(id); END_MTX_PROTECTED_OPERATIONS_UNCOND(&listMutex); + ENDfunc return RS_RET_OK; } diff --git a/runtime/debug.h b/runtime/debug.h index 1375493d..8b66d784 100644 --- a/runtime/debug.h +++ b/runtime/debug.h @@ -134,8 +134,8 @@ void dbgPrintAllDebugInfo(void); /* debug aides */ -//#ifdef RTINST -#if 0 // temporarily removed for helgrind +#ifdef RTINST +//#if 0 // temporarily removed for helgrind #define d_pthread_mutex_lock(x) dbgMutexLock(x, pdbgFuncDB, __LINE__, dbgCALLStaCK_POP_POINT ) #define d_pthread_mutex_trylock(x) dbgMutexTryLock(x, pdbgFuncDB, __LINE__, dbgCALLStaCK_POP_POINT ) #define d_pthread_mutex_unlock(x) dbgMutexUnlock(x, pdbgFuncDB, __LINE__, dbgCALLStaCK_POP_POINT ) diff --git a/runtime/stream.c b/runtime/stream.c index 00c726d9..a9f4803f 100644 --- a/runtime/stream.c +++ b/runtime/stream.c @@ -48,37 +48,21 @@ #include "stream.h" #include "unicode-helper.h" #include "module-template.h" -#include "apc.h" +#include /* static data */ DEFobjStaticHelpers DEFobjCurrIf(zlibw) -DEFobjCurrIf(apc) /* forward definitions */ static rsRetVal strmFlush(strm_t *pThis); static rsRetVal strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf); static rsRetVal strmCloseFile(strm_t *pThis); +static void *asyncWriterThread(void *pPtr); /* methods */ -/* async flush apc handler - */ -static void -flushApc(void *param1, void __attribute__((unused)) *param2) -{ - DEFVARS_mutexProtection_uncond; - strm_t *pThis = (strm_t*) param1; - ISOBJ_TYPE_assert(pThis, strm); - - BEGIN_MTX_PROTECTED_OPERATIONS_UNCOND(&pThis->mut); - strmFlush(pThis); - pThis->apcRequested = 0; - END_MTX_PROTECTED_OPERATIONS_UNCOND(&pThis->mut); -} - - /* Try to resolve a size limit situation. This is used to support custom-file size handlers * for omfile. It first runs the command, and then checks if we are still above the size * treshold. Note that this works only with single file names, NOT with circular names. @@ -569,11 +553,11 @@ ENDobjConstruct(strm) static rsRetVal strmConstructFinalize(strm_t *pThis) { rsRetVal localRet; + int i; DEFiRet; ASSERT(pThis != NULL); - CHKmalloc(pThis->pIOBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize)); pThis->iBufPtrMax = 0; /* results in immediate read request */ if(pThis->iZipLevel) { /* do we need a zip buf? */ localRet = objUse(zlibw, LM_ZLIBW_FILENAME); @@ -601,9 +585,33 @@ static rsRetVal strmConstructFinalize(strm_t *pThis) } } - /* if we should call flush apc's, we need a mutex */ +dbgprintf("TTT: before checks: iFlushInterval %d, bAsyncWrite %d\n", pThis->iFlushInterval, pThis->bAsyncWrite); + /* if we have a flush interval, we need to do async writes in any case */ if(pThis->iFlushInterval != 0) { + pThis->bAsyncWrite = 1; + } +dbgprintf("TTT: after checks: iFlushInterval %d, bAsyncWrite %d\n", pThis->iFlushInterval, pThis->bAsyncWrite); + + /* if we work asynchronously, we need a couple of synchronization objects */ + if(pThis->bAsyncWrite) { pthread_mutex_init(&pThis->mut, 0); + pthread_cond_init(&pThis->notFull, 0); + pthread_cond_init(&pThis->notEmpty, 0); + pthread_cond_init(&pThis->isEmpty, 0); + pThis->iCnt = pThis->iEnq = pThis->iDeq = 0; + for(i = 0 ; i < STREAM_ASYNC_NUMBUFS ; ++i) { + CHKmalloc(pThis->asyncBuf[i].pBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize)); + } + //pThis->pIOBuf = pThis->ioBuf[0]; + pThis->bStopWriter = 0; + // TODO: detached thread? + if(pthread_create(&pThis->writerThreadID, NULL, asyncWriterThread, pThis) != 0) + dbgprintf("ERROR: stream %p cold not create writer thread\n", pThis); + // TODO: remove that below later! + CHKmalloc(pThis->pIOBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize)); + } else { + /* we work synchronously, so we need to alloc a fixed pIOBuf */ + CHKmalloc(pThis->pIOBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize)); } finalize_it: @@ -611,8 +619,26 @@ finalize_it: } +/* stop the writer thread (we MUST be runnnig asynchronously when this method + * is called!) -- rgerhards, 2009-07-06 + */ +static inline void +stopWriter(strm_t *pThis) +{ + BEGINfunc + d_pthread_mutex_lock(&pThis->mut); + pThis->bStopWriter = 1; + pthread_cond_signal(&pThis->notEmpty); + d_pthread_cond_wait(&pThis->isEmpty, &pThis->mut); + d_pthread_mutex_unlock(&pThis->mut); + pthread_join(pThis->writerThreadID, NULL); + ENDfunc +} + + /* destructor for the strm object */ BEGINobjDestruct(strm) /* be sure to specify the object type also in END and CODESTART macros! */ + int i; CODESTARTobjDestruct(strm) if(pThis->tOperationsMode != STREAMMODE_READ) strmFlush(pThis); @@ -625,16 +651,23 @@ CODESTARTobjDestruct(strm) objRelease(zlibw, LM_ZLIBW_FILENAME); } - if(pThis->iFlushInterval != 0) { - // TODO: check if there is an apc and remove it! - pthread_mutex_destroy(&pThis->mut); - } - free(pThis->pszDir); - free(pThis->pIOBuf); free(pThis->pZipBuf); free(pThis->pszCurrFName); free(pThis->pszFName); + + if(pThis->bAsyncWrite) { + stopWriter(pThis); + pthread_mutex_destroy(&pThis->mut); + pthread_cond_destroy(&pThis->notFull); + pthread_cond_destroy(&pThis->notEmpty); + pthread_cond_destroy(&pThis->isEmpty); + for(i = 0 ; i < STREAM_ASYNC_NUMBUFS ; ++i) { + free(pThis->asyncBuf[i].pBuf); + } + } else { + free(pThis->pIOBuf); + } ENDobjDestruct(strm) @@ -730,11 +763,93 @@ doWriteCall(strm_t *pThis, uchar *pBuf, size_t *pLenBuf) pWriteBuf += iWritten; } while(lenBuf > 0); /* Warning: do..while()! */ + dbgoprint((obj_t*) pThis, "file %d write wrote %d bytes\n", pThis->fd, (int) iWritten); + finalize_it: *pLenBuf = iTotalWritten; RETiRet; } +#include + +/* This is the writer thread for asynchronous mode. + * -- rgerhards, 2009-07-06 + */ +static void* +asyncWriterThread(void *pPtr) +{ + int iDeq; + size_t iWritten; + strm_t *pThis = (strm_t*) pPtr; + ISOBJ_TYPE_assert(pThis, strm); + + BEGINfunc + if(prctl(PR_SET_NAME, "rs:asyn strmwr", 0, 0, 0) != 0) { + DBGPRINTF("prctl failed, not setting thread name for '%s'\n", "stream writer"); + } + +fprintf(stderr, "async stream writer thread started\n");fflush(stderr); +dbgprintf("TTT: writer thread startup\n"); + while(1) { /* loop broken inside */ + d_pthread_mutex_lock(&pThis->mut); + while(pThis->iCnt == 0) { +dbgprintf("TTT: writer thread empty queue, stopWriter=%d\n", pThis->bStopWriter); + if(pThis->bStopWriter) { + pthread_cond_signal(&pThis->isEmpty); + d_pthread_mutex_unlock(&pThis->mut); + goto finalize_it; /* break main loop */ + } + d_pthread_cond_wait(&pThis->notEmpty, &pThis->mut); + } + + iDeq = pThis->iDeq++ % STREAM_ASYNC_NUMBUFS; + iWritten = pThis->asyncBuf[iDeq].lenBuf; + doWriteCall(pThis, pThis->asyncBuf[iDeq].pBuf, &iWritten); + // TODO: error check!!!!! 2009-07-06 + + --pThis->iCnt; + if(pThis->iCnt < STREAM_ASYNC_NUMBUFS) { + pthread_cond_signal(&pThis->notFull); + } + d_pthread_mutex_unlock(&pThis->mut); + } + +finalize_it: +dbgprintf("TTT: writer thread shutdown\n"); + ENDfunc + return NULL; /* to keep pthreads happy */ +} + + +/* This function is called to "do" an async write call, what primarily means that + * the data is handed over to the writer thread (which will then do the actual write + * in parallel. -- rgerhards, 2009-07-06 + */ +static inline rsRetVal +doAsyncWriteCall(strm_t *pThis, uchar *pBuf, size_t *pLenBuf) +{ + int iEnq; + DEFiRet; + ISOBJ_TYPE_assert(pThis, strm); + + d_pthread_mutex_lock(&pThis->mut); + while(pThis->iCnt >= STREAM_ASYNC_NUMBUFS) + d_pthread_cond_wait(&pThis->notFull, &pThis->mut); + + iEnq = pThis->iEnq++ % STREAM_ASYNC_NUMBUFS; + // TODO: optimize, memcopy only for getting it initially going! + //pThis->asyncBuf[iEnq].pBuf = pBuf; + memcpy(pThis->asyncBuf[iEnq].pBuf, pBuf, *pLenBuf); + pThis->asyncBuf[iEnq].lenBuf = *pLenBuf; + + if(++pThis->iCnt == 1) + pthread_cond_signal(&pThis->notEmpty); + d_pthread_mutex_unlock(&pThis->mut); + +finalize_it: + RETiRet; +} + /* sync the file to disk, so that any unwritten data is persisted. This * also syncs the directory and thus makes sure that the file survives @@ -788,8 +903,11 @@ strmPhysWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) CHKiRet(strmOpenFile(pThis)); iWritten = lenBuf; - CHKiRet(doWriteCall(pThis, pBuf, &iWritten)); - dbgoprint((obj_t*) pThis, "file %d write wrote %d bytes\n", pThis->fd, (int) iWritten); + if(pThis->bAsyncWrite) { + CHKiRet(doAsyncWriteCall(pThis, pBuf, &iWritten)); + } else { + CHKiRet(doWriteCall(pThis, pBuf, &iWritten)); + } pThis->iBufPtr = 0; pThis->iCurrOffs += iWritten; @@ -993,37 +1111,11 @@ finalize_it: } -/* schedule an Apc flush request. - * rgerhards, 2009-06-15 - */ -static inline rsRetVal -scheduleFlushRequest(strm_t *pThis) -{ - apc_t *pApc; - DEFiRet; - - if(!pThis->apcRequested) { - /* we do an request only if none is yet pending */ - pThis->apcRequested = 1; - // TODO: find similar thing later CHKiRet(apc.CancelApc(pThis->apcID)); -dbgprintf("XXX: requesting to add apc!\n"); - CHKiRet(apc.Construct(&pApc)); - CHKiRet(apc.SetProcedure(pApc, (void (*)(void*, void*))flushApc)); - CHKiRet(apc.SetParam1(pApc, pThis)); - CHKiRet(apc.ConstructFinalize(pApc, &pThis->apcID)); - } - -finalize_it: - RETiRet; -} - - /* write memory buffer to a stream object */ static rsRetVal strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) { - DEFVARS_mutexProtection_uncond; DEFiRet; size_t iPartial; @@ -1034,11 +1126,6 @@ dbgprintf("strmWrite(%p, '%65.65s', %ld);, disabled %d, sizelim %ld, size %lld\n if(pThis->bDisabled) ABORT_FINALIZE(RS_RET_STREAM_DISABLED); -RUNLOG_VAR("%d", pThis->iFlushInterval); - if(pThis->iFlushInterval != 0) { - BEGIN_MTX_PROTECTED_OPERATIONS_UNCOND(&pThis->mut); - } - /* check if the to-be-written data is larger than our buffer size */ if(lenBuf >= pThis->sIOBufSize) { /* it is - so we do a direct write, that is most efficient. @@ -1067,17 +1154,7 @@ RUNLOG_VAR("%d", pThis->iFlushInterval); } } - /* we ignore the outcome of scheduleFlushRequest(), as we will write the data always at - * termination. For Zip mode, it could be fatal if we write after each record. - */ - if(pThis->iFlushInterval != 0) - scheduleFlushRequest(pThis); - finalize_it: - if(pThis->iFlushInterval != 0) { - END_MTX_PROTECTED_OPERATIONS_UNCOND(&pThis->mut); - } - RETiRet; } @@ -1390,7 +1467,6 @@ ENDobjQueryInterface(strm) */ BEGINObjClassInit(strm, 1, OBJ_IS_CORE_MODULE) /* request objects we use */ - CHKiRet(objUse(apc, CORE_COMPONENT)); OBJSetMethodHandler(objMethod_SERIALIZE, strmSerialize); OBJSetMethodHandler(objMethod_SETPROPERTY, strmSetProperty); diff --git a/runtime/stream.h b/runtime/stream.h index ac003c7b..2c1ac255 100644 --- a/runtime/stream.h +++ b/runtime/stream.h @@ -87,6 +87,7 @@ typedef enum { /* when extending, do NOT change existing modes! */ STREAMMODE_WRITE_APPEND = 4 } strmMode_t; +#define STREAM_ASYNC_NUMBUFS 2 /* must be a power of 2 -- TODO: make configurable */ /* The strm_t data structure */ typedef struct strm_s { BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */ @@ -112,7 +113,7 @@ typedef struct strm_s { int fd; /* the file descriptor, -1 if closed */ int fdDir; /* the directory's descriptor, in case bSync is requested (-1 if closed) */ uchar *pszCurrFName; /* name of current file (if open) */ - uchar *pIOBuf; /* io Buffer */ + uchar *pIOBuf; /* the iobuffer currently in use to gather data */ size_t iBufPtrMax; /* current max Ptr in Buffer (if partial read!) */ size_t iBufPtr; /* pointer into current buffer */ int iUngetC; /* char set via UngetChar() call or -1 if none set */ @@ -120,9 +121,22 @@ typedef struct strm_s { int iZipLevel; /* zip level (0..9). If 0, zip is completely disabled */ Bytef *pZipBuf; /* support for async flush procesing */ + bool bAsyncWrite; /* do asynchronous writes (always if a flush interval is given) */ + bool bStopWriter; /* shall writer thread terminate? */ int iFlushInterval; /* flush in which interval - 0, no flushing */ apc_id_t apcID; /* id of current Apc request (used for cancelling) */ pthread_mutex_t mut;/* mutex for flush in async mode */ + pthread_cond_t notFull; + pthread_cond_t notEmpty; + pthread_cond_t isEmpty; + short iEnq; + short iDeq; + short iCnt; /* current nbr of elements in buffer */ + struct { + uchar *pBuf; + size_t lenBuf; + } asyncBuf[STREAM_ASYNC_NUMBUFS]; + pthread_t writerThreadID; int apcRequested; /* is an apc Requested? */ /* support for omfile size-limiting commands, special counters, NOT persisted! */ off_t iSizeLimit; /* file size limit, 0 = no limit */ @@ -130,6 +144,7 @@ typedef struct strm_s { bool bIsTTY; /* is this a tty file? */ } strm_t; + /* interfaces */ BEGINinterface(strm) /* name must also be changed in ENDinterface macro! */ rsRetVal (*Construct)(strm_t **ppThis); diff --git a/runtime/wtp.c b/runtime/wtp.c index 02662cde..e37ebddf 100644 --- a/runtime/wtp.c +++ b/runtime/wtp.c @@ -421,6 +421,8 @@ wtpWrkrExecCancelCleanup(void *arg) static void * wtpWorker(void *arg) /* the arg is actually a wti object, even though we are in wtp! */ { + uchar *pszDbgHdr; + uchar thrdName[32] = "rs:"; DEFiRet; DEFVARS_mutexProtection; wti_t *pWti = (wti_t*) arg; @@ -435,7 +437,9 @@ wtpWorker(void *arg) /* the arg is actually a wti object, even though we are in pthread_sigmask(SIG_BLOCK, &sigSet, NULL); /* set thread name - we ignore if the call fails, has no harsh consequences... */ - if(prctl(PR_SET_NAME, wtpGetDbgHdr(pThis), 0, 0, 0) != 0) { + pszDbgHdr = wtpGetDbgHdr(pThis); + strncpy(thrdName+3, pszDbgHdr, 20); + if(prctl(PR_SET_NAME, thrdName, 0, 0, 0) != 0) { DBGPRINTF("prctl failed, not setting thread name for '%s'\n", wtpGetDbgHdr(pThis)); } -- cgit