diff options
Diffstat (limited to 'runtime/stream.c')
| -rw-r--r-- | runtime/stream.c | 394 |
1 files changed, 294 insertions, 100 deletions
diff --git a/runtime/stream.c b/runtime/stream.c index 3eb7708d..94fc0ca7 100644 --- a/runtime/stream.c +++ b/runtime/stream.c @@ -16,7 +16,7 @@ * it turns out to be problematic. Then, we need to quasi-refcount the number of accesses * to the object. * - * Copyright 2008-2012 Rainer Gerhards and Adiscon GmbH. + * Copyright 2008-2013 Rainer Gerhards and Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -45,6 +45,7 @@ #include <pthread.h> #include <fcntl.h> #include <unistd.h> +#include <sys/types.h> #include <sys/stat.h> /* required for HP UX */ #include <errno.h> #include <pthread.h> @@ -56,6 +57,7 @@ #include "stream.h" #include "unicode-helper.h" #include "module-template.h" +#include "cryprov.h" #if HAVE_SYS_PRCTL_H # include <sys/prctl.h> #endif @@ -65,7 +67,6 @@ # define O_LARGEFILE 0 #endif #ifndef HAVE_LSEEK64 - typedef off_t off64_t; # define lseek64(fd, offset, whence) lseek(fd, offset, whence) #endif @@ -74,12 +75,14 @@ DEFobjStaticHelpers DEFobjCurrIf(zlibw) /* forward definitions */ -static rsRetVal strmFlushInternal(strm_t *pThis); +static rsRetVal strmFlushInternal(strm_t *pThis, int bFlushZip); static rsRetVal strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf); static rsRetVal strmCloseFile(strm_t *pThis); static void *asyncWriterThread(void *pPtr); -static rsRetVal doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf); +static rsRetVal doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf, int bFlush); +static rsRetVal doZipFinish(strm_t *pThis); static rsRetVal strmPhysWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf); +static rsRetVal strmSeekCurrOffs(strm_t *pThis); /* methods */ @@ -196,6 +199,7 @@ static rsRetVal doPhysOpen(strm_t *pThis) { int iFlags = 0; + struct stat statOpen; DEFiRet; ISOBJ_TYPE_assert(pThis, strm); @@ -233,15 +237,75 @@ doPhysOpen(strm_t *pThis) ABORT_FINALIZE(RS_RET_FILE_NOT_FOUND); else ABORT_FINALIZE(RS_RET_IO_ERROR); + } + + if(pThis->tOperationsMode == STREAMMODE_READ) { + if(fstat(pThis->fd, &statOpen) == -1) { + DBGPRINTF("Error: cannot obtain inode# for file %s\n", pThis->pszCurrFName); + ABORT_FINALIZE(RS_RET_IO_ERROR); + } + pThis->inode = statOpen.st_ino; + } + + if(!ustrcmp(pThis->pszCurrFName, UCHAR_CONSTANT(_PATH_CONSOLE)) || isatty(pThis->fd)) { + DBGPRINTF("file %d is a tty-type file\n", pThis->fd); + pThis->bIsTTY = 1; + } else { + pThis->bIsTTY = 0; + } + + if(pThis->cryprov != NULL) { + CHKiRet(pThis->cryprov->OnFileOpen(pThis->cryprovData, + pThis->pszCurrFName, &pThis->cryprovFileData)); + } +finalize_it: + RETiRet; +} + + +static rsRetVal +strmSetCurrFName(strm_t *pThis) +{ + DEFiRet; + + if(pThis->sType == STREAMTYPE_FILE_CIRCULAR) { + CHKiRet(genFileName(&pThis->pszCurrFName, pThis->pszDir, pThis->lenDir, + pThis->pszFName, pThis->lenFName, pThis->iCurrFNum, pThis->iFileNumDigits)); } else { - if(!ustrcmp(pThis->pszCurrFName, UCHAR_CONSTANT(_PATH_CONSOLE)) || isatty(pThis->fd)) { - DBGPRINTF("file %d is a tty-type file\n", pThis->fd); - pThis->bIsTTY = 1; + if(pThis->pszDir == NULL) { + if((pThis->pszCurrFName = ustrdup(pThis->pszFName)) == NULL) + ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); } else { - pThis->bIsTTY = 0; + CHKiRet(genFileName(&pThis->pszCurrFName, pThis->pszDir, pThis->lenDir, + pThis->pszFName, pThis->lenFName, -1, 0)); } } +finalize_it: + RETiRet; +} + +/* This function checks if the actual file has changed and, if so, resets the + * offset. This is support for monitoring files. It should be called after + * deserializing the strm object and before doing any other operation on it + * (most importantly not an open or seek!). + */ +static rsRetVal +CheckFileChange(strm_t *pThis) +{ + struct stat statName; + DEFiRet; + CHKiRet(strmSetCurrFName(pThis)); + if(stat((char*) pThis->pszCurrFName, &statName) == -1) + ABORT_FINALIZE(RS_RET_IO_ERROR); + DBGPRINTF("stream/after deserialize checking for file change on '%s', " + "inode %u/%u, size/currOffs %llu/%llu\n", + pThis->pszCurrFName, (unsigned) pThis->inode, + (unsigned) statName.st_ino, statName.st_size, pThis->iCurrOffs); + if(pThis->inode != statName.st_ino || statName.st_size < pThis->iCurrOffs) { + DBGPRINTF("stream: file %s has changed\n", pThis->pszCurrFName); + pThis->iCurrOffs = 0; + } finalize_it: RETiRet; } @@ -264,19 +328,8 @@ static rsRetVal strmOpenFile(strm_t *pThis) if(pThis->pszFName == NULL) ABORT_FINALIZE(RS_RET_FILE_PREFIX_MISSING); - if(pThis->sType == STREAMTYPE_FILE_CIRCULAR) { - CHKiRet(genFileName(&pThis->pszCurrFName, pThis->pszDir, pThis->lenDir, - pThis->pszFName, pThis->lenFName, pThis->iCurrFNum, pThis->iFileNumDigits)); - } else { - if(pThis->pszDir == NULL) { - if((pThis->pszCurrFName = ustrdup(pThis->pszFName)) == NULL) - ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); - } else { - CHKiRet(genFileName(&pThis->pszCurrFName, pThis->pszDir, pThis->lenDir, - pThis->pszFName, pThis->lenFName, -1, 0)); - } - } - + CHKiRet(strmSetCurrFName(pThis)); + CHKiRet(doPhysOpen(pThis)); pThis->iCurrOffs = 0; @@ -334,6 +387,7 @@ strmWaitAsyncWriterDone(strm_t *pThis) */ static rsRetVal strmCloseFile(strm_t *pThis) { + off64_t currOffs; DEFiRet; ASSERT(pThis != NULL); @@ -341,7 +395,10 @@ static rsRetVal strmCloseFile(strm_t *pThis) (pThis->pszFName == NULL) ? "N/A" : (char*)pThis->pszFName); if(pThis->tOperationsMode != STREAMMODE_READ) { - strmFlushInternal(pThis); + strmFlushInternal(pThis, 0); + if(pThis->iZipLevel) { + doZipFinish(pThis); + } if(pThis->bAsyncWrite) { strmWaitAsyncWriterDone(pThis); } @@ -351,8 +408,14 @@ static rsRetVal strmCloseFile(strm_t *pThis) * against this. -- rgerhards, 2010-03-19 */ if(pThis->fd != -1) { + currOffs = lseek64(pThis->fd, 0, SEEK_CUR); close(pThis->fd); pThis->fd = -1; + pThis->inode = 0; + if(pThis->cryprov != NULL) { + pThis->cryprov->OnFileClose(pThis->cryprovFileData, currOffs); + pThis->cryprovFileData = NULL; + } } if(pThis->fdDir != -1) { @@ -361,7 +424,13 @@ static rsRetVal strmCloseFile(strm_t *pThis) pThis->fdDir = -1; } - if(pThis->bDeleteOnClose && pThis->pszCurrFName != NULL) { + if(pThis->bDeleteOnClose) { + if(pThis->pszCurrFName == NULL) { + CHKiRet(genFileName(&pThis->pszCurrFName, pThis->pszDir, pThis->lenDir, + pThis->pszFName, pThis->lenFName, pThis->iCurrFNum, + pThis->iFileNumDigits)); + } + DBGPRINTF("strmCloseFile: deleting '%s'\n", pThis->pszCurrFName); if(unlink((char*) pThis->pszCurrFName) == -1) { char errStr[1024]; int err = errno; @@ -369,12 +438,13 @@ static rsRetVal strmCloseFile(strm_t *pThis) DBGPRINTF("error %d unlinking '%s' - ignored: %s\n", errno, pThis->pszCurrFName, errStr); } - free(pThis->pszCurrFName); /* no longer needed in any case (just for open) */ + free(pThis->pszCurrFName); pThis->pszCurrFName = NULL; } pThis->iCurrOffs = 0; /* we are back at begin of file */ +finalize_it: RETiRet; } @@ -421,18 +491,15 @@ static rsRetVal strmHandleEOFMonitor(strm_t *pThis) { DEFiRet; - struct stat statOpen; struct stat statName; ISOBJ_TYPE_assert(pThis, strm); - if(fstat(pThis->fd, &statOpen) == -1) - ABORT_FINALIZE(RS_RET_IO_ERROR); if(stat((char*) pThis->pszCurrFName, &statName) == -1) ABORT_FINALIZE(RS_RET_IO_ERROR); - DBGPRINTF("stream checking for file change on '%s', inode %u/%u", - pThis->pszCurrFName, (unsigned) statOpen.st_ino, + DBGPRINTF("stream checking for file change on '%s', inode %u/%u\n", + pThis->pszCurrFName, (unsigned) pThis->inode, (unsigned) statName.st_ino); - if(statOpen.st_ino == statName.st_ino) { + if(pThis->inode == statName.st_ino) { ABORT_FINALIZE(RS_RET_EOF); } else { /* we had a file change! */ @@ -585,25 +652,33 @@ strmReadLine(strm_t *pThis, cstr_t **ppCStr, int mode) * mode = 2 LF <not whitespace> mode, a log line starts at the beginning of a line, but following lines that are indented are part of the same log entry * This modal interface is not nearly as flexible as being able to define a regex for when a new record starts, but it's also not nearly as hard (or as slow) to implement */ - DEFiRet; uchar c; uchar finished; + rsRetVal readCharRet; + DEFiRet; ASSERT(pThis != NULL); ASSERT(ppCStr != NULL); CHKiRet(cstrConstruct(ppCStr)); - - /* now read the line */ CHKiRet(strmReadChar(pThis, &c)); - if (mode == 0){ - while(c != '\n') { + + if(mode == 0) { + /* append previous message to current message if necessary */ + if(pThis->prevLineSegment != NULL) { + CHKiRet(cstrAppendCStr(*ppCStr, pThis->prevLineSegment)); + cstrDestruct(&pThis->prevLineSegment); + } + while(c != '\n') { CHKiRet(cstrAppendChar(*ppCStr, c)); - CHKiRet(strmReadChar(pThis, &c)); + readCharRet = strmReadChar(pThis, &c); + if(readCharRet == RS_RET_EOF) {/* end of file reached without \n? */ + CHKiRet(rsCStrConstructFromCStr(&pThis->prevLineSegment, *ppCStr)); + } + CHKiRet(readCharRet); } CHKiRet(cstrFinalize(*ppCStr)); - } - if (mode == 1){ + } else if(mode == 1) { finished=0; while(finished == 0){ if(c != '\n') { @@ -624,8 +699,7 @@ strmReadLine(strm_t *pThis, cstr_t **ppCStr, int mode) } } CHKiRet(cstrFinalize(*ppCStr)); - } - if (mode == 2){ + } else if(mode == 2) { /* indented follow-up lines */ finished=0; while(finished == 0){ @@ -675,9 +749,11 @@ BEGINobjConstruct(strm) /* be sure to specify the object type also in END macro! pThis->fd = -1; pThis->fdDir = -1; pThis->iUngetC = -1; + pThis->bVeryReliableZip = 0; pThis->sType = STREAMTYPE_FILE_SINGLE; pThis->sIOBufSize = glblGetIOBufSize(); pThis->tOpenMode = 0600; + pThis->prevLineSegment = NULL; ENDobjConstruct(strm) @@ -777,6 +853,7 @@ stopWriter(strm_t *pThis) BEGINobjDestruct(strm) /* be sure to specify the object type also in END and CODESTART macros! */ int i; CODESTARTobjDestruct(strm) + /* we need to stop the ZIP writer */ if(pThis->bAsyncWrite) /* Note: mutex will be unlocked in stopWriter! */ d_pthread_mutex_lock(&pThis->mut); @@ -919,14 +996,14 @@ finalize_it: /* write memory buffer to a stream object. */ static inline rsRetVal -doWriteInternal(strm_t *pThis, uchar *pBuf, size_t lenBuf) +doWriteInternal(strm_t *pThis, uchar *pBuf, size_t lenBuf, int bFlush) { DEFiRet; ASSERT(pThis != NULL); if(pThis->iZipLevel) { - CHKiRet(doZipWrite(pThis, pBuf, lenBuf)); + CHKiRet(doZipWrite(pThis, pBuf, lenBuf, bFlush)); } else { /* write without zipping */ CHKiRet(strmPhysWrite(pThis, pBuf, lenBuf)); @@ -971,7 +1048,7 @@ doAsyncWriteInternal(strm_t *pThis, size_t lenBuf) * the background thread. -- rgerhards, 2009-07-07 */ static rsRetVal -strmSchedWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) +strmSchedWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf, int bFlushZip) { DEFiRet; @@ -990,7 +1067,7 @@ strmSchedWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) if(pThis->bAsyncWrite) { CHKiRet(doAsyncWriteInternal(pThis, lenBuf)); } else { - CHKiRet(doWriteInternal(pThis, pBuf, lenBuf)); + CHKiRet(doWriteInternal(pThis, pBuf, lenBuf, bFlushZip)); } @@ -1011,17 +1088,20 @@ asyncWriterThread(void *pPtr) sbool bTimedOut = 0; strm_t *pThis = (strm_t*) pPtr; int err; + uchar thrdName[256] = "rs:"; ISOBJ_TYPE_assert(pThis, strm); BEGINfunc + ustrncpy(thrdName+3, pThis->pszFName, sizeof(thrdName)-4); + dbgOutputTID((char*)thrdName); # if HAVE_PRCTL && defined PR_SET_NAME - if(prctl(PR_SET_NAME, "rs:asyn strmwr", 0, 0, 0) != 0) { + if(prctl(PR_SET_NAME, (char*)thrdName, 0, 0, 0) != 0) { DBGPRINTF("prctl failed, not setting thread name for '%s'\n", "stream writer"); } # endif + d_pthread_mutex_lock(&pThis->mut); while(1) { /* loop broken inside */ - d_pthread_mutex_lock(&pThis->mut); while(pThis->iCnt == 0) { if(pThis->bStopWriter) { pthread_cond_broadcast(&pThis->isEmpty); @@ -1030,18 +1110,17 @@ asyncWriterThread(void *pPtr) } if(bTimedOut && pThis->iBufPtr > 0) { /* if we timed out, we need to flush pending data */ - strmFlushInternal(pThis); + strmFlushInternal(pThis, 0); bTimedOut = 0; - continue; /* now we should have data */ + d_pthread_mutex_unlock(&pThis->mut); + continue; } bTimedOut = 0; timeoutComp(&t, pThis->iFlushInterval * 1000); /* *1000 millisconds */ if(pThis->bDoTimedWait) { if((err = pthread_cond_timedwait(&pThis->notEmpty, &pThis->mut, &t)) != 0) { - if(err == ETIMEDOUT) { - bTimedOut = 1; - } else { - bTimedOut = 1; + bTimedOut = 1; /* simulate in any case */ + if(err != ETIMEDOUT) { char errStr[1024]; rs_strerror_r(err, errStr, sizeof(errStr)); DBGPRINTF("stream async writer timeout with error (%d): %s - ignoring\n", @@ -1056,8 +1135,12 @@ asyncWriterThread(void *pPtr) bTimedOut = 0; /* we may have timed out, but there *is* work to do... */ iDeq = pThis->iDeq++ % STREAM_ASYNC_NUMBUFS; - doWriteInternal(pThis, pThis->asyncBuf[iDeq].pBuf, pThis->asyncBuf[iDeq].lenBuf); + + /* now we can do the actual write in parallel */ + d_pthread_mutex_unlock(&pThis->mut); + doWriteInternal(pThis, pThis->asyncBuf[iDeq].pBuf, pThis->asyncBuf[iDeq].lenBuf, 0); // TODO: flush state // TODO: error check????? 2009-07-06 + d_pthread_mutex_lock(&pThis->mut); --pThis->iCnt; if(pThis->iCnt < STREAM_ASYNC_NUMBUFS) { @@ -1065,8 +1148,8 @@ asyncWriterThread(void *pPtr) if(pThis->iCnt == 0) pthread_cond_broadcast(&pThis->isEmpty); } - d_pthread_mutex_unlock(&pThis->mut); } + d_pthread_mutex_unlock(&pThis->mut); finalize_it: ENDfunc @@ -1128,9 +1211,16 @@ strmPhysWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) DEFiRet; ISOBJ_TYPE_assert(pThis, strm); + DBGPRINTF("strmPhysWrite, stream %p, len %u\n", pThis, (unsigned)lenBuf); if(pThis->fd == -1) CHKiRet(strmOpenFile(pThis)); + /* here we place our crypto interface */ + if(pThis->cryprov != NULL) { + pThis->cryprov->Encrypt(pThis->cryprovFileData, pBuf, &lenBuf); + } + /* end crypto */ + iWritten = lenBuf; CHKiRet(doWriteCall(pThis, pBuf, &iWritten)); @@ -1166,63 +1256,97 @@ finalize_it: * rgerhards, 2009-06-04 */ static rsRetVal -doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) +doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf, int bFlush) { - z_stream zstrm; int zRet; /* zlib return state */ - sbool bzInitDone = RSFALSE; DEFiRet; + unsigned outavail; assert(pThis != NULL); assert(pBuf != NULL); - /* allocate deflate state */ - zstrm.zalloc = Z_NULL; - zstrm.zfree = Z_NULL; - zstrm.opaque = Z_NULL; - zstrm.next_in = (Bytef*) pBuf; /* as of zlib doc, this must be set BEFORE DeflateInit2 */ - /* see note in file header for the params we use with deflateInit2() */ - zRet = zlibw.DeflateInit2(&zstrm, pThis->iZipLevel, Z_DEFLATED, 31, 9, Z_DEFAULT_STRATEGY); - if(zRet != Z_OK) { - DBGPRINTF("error %d returned from zlib/deflateInit2()\n", zRet); - ABORT_FINALIZE(RS_RET_ZLIB_ERR); + if(!pThis->bzInitDone) { + /* allocate deflate state */ + pThis->zstrm.zalloc = Z_NULL; + pThis->zstrm.zfree = Z_NULL; + pThis->zstrm.opaque = Z_NULL; + /* see note in file header for the params we use with deflateInit2() */ + zRet = zlibw.DeflateInit2(&pThis->zstrm, pThis->iZipLevel, Z_DEFLATED, 31, 9, Z_DEFAULT_STRATEGY); + if(zRet != Z_OK) { + DBGPRINTF("error %d returned from zlib/deflateInit2()\n", zRet); + ABORT_FINALIZE(RS_RET_ZLIB_ERR); + } + pThis->bzInitDone = RSTRUE; } - bzInitDone = RSTRUE; /* now doing the compression */ - zstrm.next_in = (Bytef*) pBuf; /* as of zlib doc, this must be set BEFORE DeflateInit2 */ - zstrm.avail_in = lenBuf; + pThis->zstrm.next_in = (Bytef*) pBuf; + pThis->zstrm.avail_in = lenBuf; /* run deflate() on buffer until everything has been compressed */ do { - DBGPRINTF("in deflate() loop, avail_in %d, total_in %ld\n", zstrm.avail_in, zstrm.total_in); - zstrm.avail_out = pThis->sIOBufSize; - zstrm.next_out = pThis->pZipBuf; - zRet = zlibw.Deflate(&zstrm, Z_FINISH); /* no bad return value */ - DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, zstrm.avail_out); - assert(zRet != Z_STREAM_ERROR); /* state not clobbered */ - if(zstrm.avail_out == pThis->sIOBufSize) - break; /* this is valid, indicates end of compression --> see zlib howto */ - CHKiRet(strmPhysWrite(pThis, (uchar*)pThis->pZipBuf, pThis->sIOBufSize - zstrm.avail_out)); - } while (zstrm.avail_out == 0); - assert(zstrm.avail_in == 0); /* all input will be used */ + DBGPRINTF("in deflate() loop, avail_in %d, total_in %ld\n", pThis->zstrm.avail_in, pThis->zstrm.total_in); + pThis->zstrm.avail_out = pThis->sIOBufSize; + pThis->zstrm.next_out = pThis->pZipBuf; + zRet = zlibw.Deflate(&pThis->zstrm, bFlush ? Z_SYNC_FLUSH : Z_NO_FLUSH); /* no bad return value */ + DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, pThis->zstrm.avail_out); + outavail =pThis->sIOBufSize - pThis->zstrm.avail_out; + if(outavail != 0) { + CHKiRet(strmPhysWrite(pThis, (uchar*)pThis->pZipBuf, outavail)); + } + } while (pThis->zstrm.avail_out == 0); finalize_it: - if(bzInitDone) { - zRet = zlibw.DeflateEnd(&zstrm); - if(zRet != Z_OK) { - DBGPRINTF("error %d returned from zlib/deflateEnd()\n", zRet); - } + if(pThis->bzInitDone && pThis->bVeryReliableZip) { + doZipFinish(pThis); } - RETiRet; } + +/* finish zlib buffer, to be called before closing the ZIP file (if + * running in stream mode). + */ +static rsRetVal +doZipFinish(strm_t *pThis) +{ + int zRet; /* zlib return state */ + DEFiRet; + unsigned outavail; + assert(pThis != NULL); + + if(!pThis->bzInitDone) + goto done; + + pThis->zstrm.avail_in = 0; + /* run deflate() on buffer until everything has been compressed */ + do { + DBGPRINTF("in deflate() loop, avail_in %d, total_in %ld\n", pThis->zstrm.avail_in, pThis->zstrm.total_in); + pThis->zstrm.avail_out = pThis->sIOBufSize; + pThis->zstrm.next_out = pThis->pZipBuf; + zRet = zlibw.Deflate(&pThis->zstrm, Z_FINISH); /* no bad return value */ + DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, pThis->zstrm.avail_out); + outavail = pThis->sIOBufSize - pThis->zstrm.avail_out; + if(outavail != 0) { + CHKiRet(strmPhysWrite(pThis, (uchar*)pThis->pZipBuf, outavail)); + } + } while (pThis->zstrm.avail_out == 0); + +finalize_it: + zRet = zlibw.DeflateEnd(&pThis->zstrm); + if(zRet != Z_OK) { + DBGPRINTF("error %d returned from zlib/deflateEnd()\n", zRet); + } + + pThis->bzInitDone = 0; +done: RETiRet; +} + /* flush stream output buffer to persistent storage. This can be called at any time * and is automatically called when the output buffer is full. * rgerhards, 2008-01-10 */ static rsRetVal -strmFlushInternal(strm_t *pThis) +strmFlushInternal(strm_t *pThis, int bFlushZip) { DEFiRet; @@ -1232,7 +1356,7 @@ strmFlushInternal(strm_t *pThis) (long) pThis->iBufPtr, (pThis->iBufPtr == 0) ? " (no need to flush)" : ""); if(pThis->tOperationsMode != STREAMMODE_READ && pThis->iBufPtr > 0) { - iRet = strmSchedWrite(pThis, pThis->pIOBuf, pThis->iBufPtr); + iRet = strmSchedWrite(pThis, pThis->pIOBuf, pThis->iBufPtr, bFlushZip); } RETiRet; @@ -1254,7 +1378,7 @@ strmFlush(strm_t *pThis) if(pThis->bAsyncWrite) d_pthread_mutex_lock(&pThis->mut); - CHKiRet(strmFlushInternal(pThis)); + CHKiRet(strmFlushInternal(pThis, 1)); finalize_it: if(pThis->bAsyncWrite) @@ -1277,11 +1401,15 @@ static rsRetVal strmSeek(strm_t *pThis, off64_t offs) if(pThis->fd == -1) { CHKiRet(strmOpenFile(pThis)); } else { - CHKiRet(strmFlushInternal(pThis)); + CHKiRet(strmFlushInternal(pThis, 0)); } long long i; DBGOPRINT((obj_t*) pThis, "file %d seek, pos %llu\n", pThis->fd, (long long unsigned) offs); - i = lseek64(pThis->fd, offs, SEEK_SET); // TODO: check error! + i = lseek64(pThis->fd, offs, SEEK_SET); + if(i != offs) { + DBGPRINTF("strmSeek: error %lld seeking to offset %lld\n", i, offs); + ABORT_FINALIZE(RS_RET_IO_ERROR); + } pThis->iCurrOffs = offs; /* we are now at *this* offset */ pThis->iBufPtr = 0; /* buffer invalidated */ @@ -1289,6 +1417,56 @@ finalize_it: RETiRet; } +/* multi-file seek, seeks to file number & offset within file. This + * is a support function for the queue, in circular mode. DO NOT USE + * IT FOR OTHER NEEDS - it may not work as expected. It will + * seek to the new position and delete interim files, as it skips them. + * Note: this code can be removed when the queue gets a new disk store + * handler (if and when it does ;)). + * The output parameter bytesDel receives the number of bytes that have + * been deleted (if a file is deleted) or 0 if nothing was deleted. + * rgerhards, 2012-11-07 + */ +rsRetVal +strmMultiFileSeek(strm_t *pThis, int FNum, off64_t offs, off64_t *bytesDel) +{ + struct stat statBuf; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, strm); + + if(FNum == 0 && offs == 0) { /* happens during queue init */ + *bytesDel = 0; + FINALIZE; + } + + if(pThis->iCurrFNum != FNum) { + /* Note: we assume that no more than one file is skipped - an + * assumption that is being used also by the whole rest of the + * code and most notably the queue subsystem. + */ + CHKiRet(genFileName(&pThis->pszCurrFName, pThis->pszDir, pThis->lenDir, + pThis->pszFName, pThis->lenFName, pThis->iCurrFNum, + pThis->iFileNumDigits)); + stat((char*)pThis->pszCurrFName, &statBuf); + *bytesDel = statBuf.st_size; + DBGPRINTF("strmMultiFileSeek: detected new filenum, was %d, new %d, " + "deleting '%s' (%lld bytes)\n", pThis->iCurrFNum, FNum, + pThis->pszCurrFName, (long long) *bytesDel); + unlink((char*)pThis->pszCurrFName); + free(pThis->pszCurrFName); + pThis->pszCurrFName = NULL; + pThis->iCurrFNum = FNum; + } else { + *bytesDel = 0; + } + pThis->iCurrOffs = offs; + +finalize_it: + RETiRet; +} + + /* seek to current offset. This is primarily a helper to readjust the OS file * pointer after a strm object has been deserialized. @@ -1320,7 +1498,7 @@ static rsRetVal strmWriteChar(strm_t *pThis, uchar c) /* if the buffer is full, we need to flush before we can write */ if(pThis->iBufPtr == pThis->sIOBufSize) { - CHKiRet(strmFlushInternal(pThis)); + CHKiRet(strmFlushInternal(pThis, 0)); } /* we now always have space for one character, so we simply copy it */ *(pThis->pIOBuf + pThis->iBufPtr) = c; @@ -1380,17 +1558,17 @@ strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) ASSERT(pThis != NULL); ASSERT(pBuf != NULL); -//DBGPRINTF("strmWrite(%p, '%65.65s', %ld);, disabled %d, sizelim %ld, size %lld\n", pThis, pBuf,lenBuf, pThis->bDisabled, pThis->iSizeLimit, pThis->iCurrOffs); - if(pThis->bAsyncWrite) - d_pthread_mutex_lock(&pThis->mut); - + /* DEV DEBUG ONLY DBGPRINTF("strmWrite(%p[%s], '%65.65s', %ld);, disabled %d, sizelim %ld, size %lld\n", pThis, pThis->pszCurrFName, pBuf,(long) lenBuf, pThis->bDisabled, (long) pThis->iSizeLimit, (long long) pThis->iCurrOffs); */ if(pThis->bDisabled) ABORT_FINALIZE(RS_RET_STREAM_DISABLED); + if(pThis->bAsyncWrite) + d_pthread_mutex_lock(&pThis->mut); + iOffset = 0; do { if(pThis->iBufPtr == pThis->sIOBufSize) { - CHKiRet(strmFlushInternal(pThis)); /* get a new buffer for rest of data */ + CHKiRet(strmFlushInternal(pThis, 0)); /* get a new buffer for rest of data */ } iWrite = pThis->sIOBufSize - pThis->iBufPtr; /* this fits in current buf */ if(iWrite > lenBuf) @@ -1405,7 +1583,7 @@ strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) * write it. This seems more natural than waiting (hours?) for the next message... */ if(pThis->iBufPtr == pThis->sIOBufSize) { - CHKiRet(strmFlushInternal(pThis)); /* get a new buffer for rest of data */ + CHKiRet(strmFlushInternal(pThis, 0)); /* get a new buffer for rest of data */ } finalize_it: @@ -1433,11 +1611,14 @@ DEFpropSetMeth(strm, tOperationsMode, int) DEFpropSetMeth(strm, tOpenMode, mode_t) DEFpropSetMeth(strm, sType, strmType_t) DEFpropSetMeth(strm, iZipLevel, int) +DEFpropSetMeth(strm, bVeryReliableZip, int) DEFpropSetMeth(strm, bSync, int) DEFpropSetMeth(strm, sIOBufSize, size_t) DEFpropSetMeth(strm, iSizeLimit, off_t) DEFpropSetMeth(strm, iFlushInterval, int) DEFpropSetMeth(strm, pszSizeLimitCmd, uchar*) +DEFpropSetMeth(strm, cryprov, cryprov_if_t*) +DEFpropSetMeth(strm, cryprovData, void*) static rsRetVal strmSetiMaxFiles(strm_t *pThis, int iNewVal) { @@ -1564,7 +1745,7 @@ static rsRetVal strmSerialize(strm_t *pThis, strm_t *pStrm) ISOBJ_TYPE_assert(pThis, strm); ISOBJ_TYPE_assert(pStrm, strm); - strmFlushInternal(pThis); + strmFlushInternal(pThis, 0); CHKiRet(obj.BeginSerialize(pStrm, (obj_t*) pThis)); objSerializeSCALAR(pStrm, iCurrFNum, INT); @@ -1584,6 +1765,11 @@ static rsRetVal strmSerialize(strm_t *pThis, strm_t *pStrm) l = pThis->iCurrOffs; objSerializeSCALAR_VAR(pStrm, iCurrOffs, INT64, l); + l = pThis->inode; + objSerializeSCALAR_VAR(pStrm, inode, INT64, l); + + objSerializePTR(pStrm, prevLineSegment, PSZ); + CHKiRet(obj.EndSerialize(pStrm)); finalize_it: @@ -1681,6 +1867,8 @@ static rsRetVal strmSetProperty(strm_t *pThis, var_t *pProp) CHKiRet(strmSettOpenMode(pThis, pProp->val.num)); } else if(isProp("iCurrOffs")) { pThis->iCurrOffs = pProp->val.num; + } else if(isProp("inode")) { + pThis->inode = (ino_t) pProp->val.num; } else if(isProp("iMaxFileSize")) { CHKiRet(strmSetiMaxFileSize(pThis, pProp->val.num)); } else if(isProp("iMaxFiles")) { @@ -1689,6 +1877,8 @@ static rsRetVal strmSetProperty(strm_t *pThis, var_t *pProp) CHKiRet(strmSetiFileNumDigits(pThis, pProp->val.num)); } else if(isProp("bDeleteOnClose")) { CHKiRet(strmSetbDeleteOnClose(pThis, pProp->val.num)); + } else if(isProp("prevLineSegment")) { + CHKiRet(rsCStrConstructFromCStr(&pThis->prevLineSegment, pProp->val.pStr)); } finalize_it: @@ -1748,6 +1938,7 @@ CODESTARTobjQueryInterface(strm) pIf->GetCurrOffset = strmGetCurrOffset; pIf->Dup = strmDup; pIf->SetWCntr = strmSetWCntr; + pIf->CheckFileChange = CheckFileChange; /* set methods */ pIf->SetbDeleteOnClose = strmSetbDeleteOnClose; pIf->SetiMaxFileSize = strmSetiMaxFileSize; @@ -1757,11 +1948,14 @@ CODESTARTobjQueryInterface(strm) pIf->SettOpenMode = strmSettOpenMode; pIf->SetsType = strmSetsType; pIf->SetiZipLevel = strmSetiZipLevel; + pIf->SetbVeryReliableZip = strmSetbVeryReliableZip; pIf->SetbSync = strmSetbSync; pIf->SetsIOBufSize = strmSetsIOBufSize; pIf->SetiSizeLimit = strmSetiSizeLimit; pIf->SetiFlushInterval = strmSetiFlushInterval; pIf->SetpszSizeLimitCmd = strmSetpszSizeLimitCmd; + pIf->Setcryprov = strmSetcryprov; + pIf->SetcryprovData = strmSetcryprovData; finalize_it: ENDobjQueryInterface(strm) |