summaryrefslogtreecommitdiffstatshomepage
path: root/3rdparty/lzma/C/MtCoder.c
diff options
context:
space:
mode:
Diffstat (limited to '3rdparty/lzma/C/MtCoder.c')
-rw-r--r--3rdparty/lzma/C/MtCoder.c710
1 files changed, 477 insertions, 233 deletions
diff --git a/3rdparty/lzma/C/MtCoder.c b/3rdparty/lzma/C/MtCoder.c
index 8d9731d0a28..6f58abb2a39 100644
--- a/3rdparty/lzma/C/MtCoder.c
+++ b/3rdparty/lzma/C/MtCoder.c
@@ -1,327 +1,571 @@
/* MtCoder.c -- Multi-thread Coder
-2015-10-13 : Igor Pavlov : Public domain */
+2023-04-13 : Igor Pavlov : Public domain */
#include "Precomp.h"
#include "MtCoder.h"
-void LoopThread_Construct(CLoopThread *p)
-{
- Thread_Construct(&p->thread);
- Event_Construct(&p->startEvent);
- Event_Construct(&p->finishedEvent);
-}
-
-void LoopThread_Close(CLoopThread *p)
-{
- Thread_Close(&p->thread);
- Event_Close(&p->startEvent);
- Event_Close(&p->finishedEvent);
-}
+#ifndef Z7_ST
-static THREAD_FUNC_RET_TYPE THREAD_FUNC_CALL_TYPE LoopThreadFunc(void *pp)
+static SRes MtProgressThunk_Progress(ICompressProgressPtr pp, UInt64 inSize, UInt64 outSize)
{
- CLoopThread *p = (CLoopThread *)pp;
- for (;;)
+ Z7_CONTAINER_FROM_VTBL_TO_DECL_VAR_pp_vt_p(CMtProgressThunk)
+ UInt64 inSize2 = 0;
+ UInt64 outSize2 = 0;
+ if (inSize != (UInt64)(Int64)-1)
{
- if (Event_Wait(&p->startEvent) != 0)
- return SZ_ERROR_THREAD;
- if (p->stop)
- return 0;
- p->res = p->func(p->param);
- if (Event_Set(&p->finishedEvent) != 0)
- return SZ_ERROR_THREAD;
+ inSize2 = inSize - p->inSize;
+ p->inSize = inSize;
}
+ if (outSize != (UInt64)(Int64)-1)
+ {
+ outSize2 = outSize - p->outSize;
+ p->outSize = outSize;
+ }
+ return MtProgress_ProgressAdd(p->mtProgress, inSize2, outSize2);
}
-WRes LoopThread_Create(CLoopThread *p)
-{
- p->stop = 0;
- RINOK(AutoResetEvent_CreateNotSignaled(&p->startEvent));
- RINOK(AutoResetEvent_CreateNotSignaled(&p->finishedEvent));
- return Thread_Create(&p->thread, LoopThreadFunc, p);
-}
-WRes LoopThread_StopAndWait(CLoopThread *p)
+void MtProgressThunk_CreateVTable(CMtProgressThunk *p)
{
- p->stop = 1;
- if (Event_Set(&p->startEvent) != 0)
- return SZ_ERROR_THREAD;
- return Thread_Wait(&p->thread);
+ p->vt.Progress = MtProgressThunk_Progress;
}
-WRes LoopThread_StartSubThread(CLoopThread *p) { return Event_Set(&p->startEvent); }
-WRes LoopThread_WaitSubThread(CLoopThread *p) { return Event_Wait(&p->finishedEvent); }
-static SRes Progress(ICompressProgress *p, UInt64 inSize, UInt64 outSize)
-{
- return (p && p->Progress(p, inSize, outSize) != SZ_OK) ? SZ_ERROR_PROGRESS : SZ_OK;
-}
-static void MtProgress_Init(CMtProgress *p, ICompressProgress *progress)
-{
- unsigned i;
- for (i = 0; i < NUM_MT_CODER_THREADS_MAX; i++)
- p->inSizes[i] = p->outSizes[i] = 0;
- p->totalInSize = p->totalOutSize = 0;
- p->progress = progress;
- p->res = SZ_OK;
-}
+#define RINOK_THREAD(x) { if ((x) != 0) return SZ_ERROR_THREAD; }
-static void MtProgress_Reinit(CMtProgress *p, unsigned index)
-{
- p->inSizes[index] = 0;
- p->outSizes[index] = 0;
-}
-#define UPDATE_PROGRESS(size, prev, total) \
- if (size != (UInt64)(Int64)-1) { total += size - prev; prev = size; }
+static THREAD_FUNC_DECL ThreadFunc(void *pp);
-SRes MtProgress_Set(CMtProgress *p, unsigned index, UInt64 inSize, UInt64 outSize)
-{
- SRes res;
- CriticalSection_Enter(&p->cs);
- UPDATE_PROGRESS(inSize, p->inSizes[index], p->totalInSize)
- UPDATE_PROGRESS(outSize, p->outSizes[index], p->totalOutSize)
- if (p->res == SZ_OK)
- p->res = Progress(p->progress, p->totalInSize, p->totalOutSize);
- res = p->res;
- CriticalSection_Leave(&p->cs);
- return res;
-}
-static void MtProgress_SetError(CMtProgress *p, SRes res)
+static SRes MtCoderThread_CreateAndStart(CMtCoderThread *t)
{
- CriticalSection_Enter(&p->cs);
- if (p->res == SZ_OK)
- p->res = res;
- CriticalSection_Leave(&p->cs);
+ WRes wres = AutoResetEvent_OptCreate_And_Reset(&t->startEvent);
+ if (wres == 0)
+ {
+ t->stop = False;
+ if (!Thread_WasCreated(&t->thread))
+ wres = Thread_Create(&t->thread, ThreadFunc, t);
+ if (wres == 0)
+ wres = Event_Set(&t->startEvent);
+ }
+ if (wres == 0)
+ return SZ_OK;
+ return MY_SRes_HRESULT_FROM_WRes(wres);
}
-static void MtCoder_SetError(CMtCoder* p, SRes res)
+
+static void MtCoderThread_Destruct(CMtCoderThread *t)
{
- CriticalSection_Enter(&p->cs);
- if (p->res == SZ_OK)
- p->res = res;
- CriticalSection_Leave(&p->cs);
-}
+ if (Thread_WasCreated(&t->thread))
+ {
+ t->stop = 1;
+ Event_Set(&t->startEvent);
+ Thread_Wait_Close(&t->thread);
+ }
-/* ---------- MtThread ---------- */
+ Event_Close(&t->startEvent);
-void CMtThread_Construct(CMtThread *p, CMtCoder *mtCoder)
-{
- p->mtCoder = mtCoder;
- p->outBuf = 0;
- p->inBuf = 0;
- Event_Construct(&p->canRead);
- Event_Construct(&p->canWrite);
- LoopThread_Construct(&p->thread);
+ if (t->inBuf)
+ {
+ ISzAlloc_Free(t->mtCoder->allocBig, t->inBuf);
+ t->inBuf = NULL;
+ }
}
-#define RINOK_THREAD(x) { if ((x) != 0) return SZ_ERROR_THREAD; }
-static void CMtThread_CloseEvents(CMtThread *p)
-{
- Event_Close(&p->canRead);
- Event_Close(&p->canWrite);
-}
-static void CMtThread_Destruct(CMtThread *p)
+
+/*
+ ThreadFunc2() returns:
+ SZ_OK - in all normal cases (even for stream error or memory allocation error)
+ SZ_ERROR_THREAD - in case of failure in system synch function
+*/
+
+static SRes ThreadFunc2(CMtCoderThread *t)
{
- CMtThread_CloseEvents(p);
+ CMtCoder *mtc = t->mtCoder;
- if (Thread_WasCreated(&p->thread.thread))
+ for (;;)
{
- LoopThread_StopAndWait(&p->thread);
- LoopThread_Close(&p->thread);
- }
+ unsigned bi;
+ SRes res;
+ SRes res2;
+ BoolInt finished;
+ unsigned bufIndex;
+ size_t size;
+ const Byte *inData;
+ UInt64 readProcessed = 0;
+
+ RINOK_THREAD(Event_Wait(&mtc->readEvent))
+
+ /* after Event_Wait(&mtc->readEvent) we must call Event_Set(&mtc->readEvent) in any case to unlock another threads */
+
+ if (mtc->stopReading)
+ {
+ return Event_Set(&mtc->readEvent) == 0 ? SZ_OK : SZ_ERROR_THREAD;
+ }
- if (p->mtCoder->alloc)
- IAlloc_Free(p->mtCoder->alloc, p->outBuf);
- p->outBuf = 0;
+ res = MtProgress_GetError(&mtc->mtProgress);
+
+ size = 0;
+ inData = NULL;
+ finished = True;
- if (p->mtCoder->alloc)
- IAlloc_Free(p->mtCoder->alloc, p->inBuf);
- p->inBuf = 0;
-}
+ if (res == SZ_OK)
+ {
+ size = mtc->blockSize;
+ if (mtc->inStream)
+ {
+ if (!t->inBuf)
+ {
+ t->inBuf = (Byte *)ISzAlloc_Alloc(mtc->allocBig, mtc->blockSize);
+ if (!t->inBuf)
+ res = SZ_ERROR_MEM;
+ }
+ if (res == SZ_OK)
+ {
+ res = SeqInStream_ReadMax(mtc->inStream, t->inBuf, &size);
+ readProcessed = mtc->readProcessed + size;
+ mtc->readProcessed = readProcessed;
+ }
+ if (res != SZ_OK)
+ {
+ mtc->readRes = res;
+ /* after reading error - we can stop encoding of previous blocks */
+ MtProgress_SetError(&mtc->mtProgress, res);
+ }
+ else
+ finished = (size != mtc->blockSize);
+ }
+ else
+ {
+ size_t rem;
+ readProcessed = mtc->readProcessed;
+ rem = mtc->inDataSize - (size_t)readProcessed;
+ if (size > rem)
+ size = rem;
+ inData = mtc->inData + (size_t)readProcessed;
+ readProcessed += size;
+ mtc->readProcessed = readProcessed;
+ finished = (mtc->inDataSize == (size_t)readProcessed);
+ }
+ }
-#define MY_BUF_ALLOC(buf, size, newSize) \
- if (buf == 0 || size != newSize) \
- { IAlloc_Free(p->mtCoder->alloc, buf); \
- size = newSize; buf = (Byte *)IAlloc_Alloc(p->mtCoder->alloc, size); \
- if (buf == 0) return SZ_ERROR_MEM; }
+ /* we must get some block from blocksSemaphore before Event_Set(&mtc->readEvent) */
-static SRes CMtThread_Prepare(CMtThread *p)
-{
- MY_BUF_ALLOC(p->inBuf, p->inBufSize, p->mtCoder->blockSize)
- MY_BUF_ALLOC(p->outBuf, p->outBufSize, p->mtCoder->destBlockSize)
+ res2 = SZ_OK;
- p->stopReading = False;
- p->stopWriting = False;
- RINOK_THREAD(AutoResetEvent_CreateNotSignaled(&p->canRead));
- RINOK_THREAD(AutoResetEvent_CreateNotSignaled(&p->canWrite));
+ if (Semaphore_Wait(&mtc->blocksSemaphore) != 0)
+ {
+ res2 = SZ_ERROR_THREAD;
+ if (res == SZ_OK)
+ {
+ res = res2;
+ // MtProgress_SetError(&mtc->mtProgress, res);
+ }
+ }
- return SZ_OK;
-}
+ bi = mtc->blockIndex;
-static SRes FullRead(ISeqInStream *stream, Byte *data, size_t *processedSize)
-{
- size_t size = *processedSize;
- *processedSize = 0;
- while (size != 0)
- {
- size_t curSize = size;
- SRes res = stream->Read(stream, data, &curSize);
- *processedSize += curSize;
- data += curSize;
- size -= curSize;
- RINOK(res);
- if (curSize == 0)
- return SZ_OK;
- }
- return SZ_OK;
-}
+ if (++mtc->blockIndex >= mtc->numBlocksMax)
+ mtc->blockIndex = 0;
-#define GET_NEXT_THREAD(p) &p->mtCoder->threads[p->index == p->mtCoder->numThreads - 1 ? 0 : p->index + 1]
+ bufIndex = (unsigned)(int)-1;
-static SRes MtThread_Process(CMtThread *p, Bool *stop)
-{
- CMtThread *next;
- *stop = True;
- if (Event_Wait(&p->canRead) != 0)
- return SZ_ERROR_THREAD;
-
- next = GET_NEXT_THREAD(p);
-
- if (p->stopReading)
- {
- next->stopReading = True;
- return Event_Set(&next->canRead) == 0 ? SZ_OK : SZ_ERROR_THREAD;
- }
+ if (res == SZ_OK)
+ res = MtProgress_GetError(&mtc->mtProgress);
- {
- size_t size = p->mtCoder->blockSize;
- size_t destSize = p->outBufSize;
-
- RINOK(FullRead(p->mtCoder->inStream, p->inBuf, &size));
- next->stopReading = *stop = (size != p->mtCoder->blockSize);
- if (Event_Set(&next->canRead) != 0)
- return SZ_ERROR_THREAD;
-
- RINOK(p->mtCoder->mtCallback->Code(p->mtCoder->mtCallback, p->index,
- p->outBuf, &destSize, p->inBuf, size, *stop));
-
- MtProgress_Reinit(&p->mtCoder->mtProgress, p->index);
-
- if (Event_Wait(&p->canWrite) != 0)
- return SZ_ERROR_THREAD;
- if (p->stopWriting)
- return SZ_ERROR_FAIL;
- if (p->mtCoder->outStream->Write(p->mtCoder->outStream, p->outBuf, destSize) != destSize)
- return SZ_ERROR_WRITE;
- return Event_Set(&next->canWrite) == 0 ? SZ_OK : SZ_ERROR_THREAD;
+ if (res != SZ_OK)
+ finished = True;
+
+ if (!finished)
+ {
+ if (mtc->numStartedThreads < mtc->numStartedThreadsLimit
+ && mtc->expectedDataSize != readProcessed)
+ {
+ res = MtCoderThread_CreateAndStart(&mtc->threads[mtc->numStartedThreads]);
+ if (res == SZ_OK)
+ mtc->numStartedThreads++;
+ else
+ {
+ MtProgress_SetError(&mtc->mtProgress, res);
+ finished = True;
+ }
+ }
+ }
+
+ if (finished)
+ mtc->stopReading = True;
+
+ RINOK_THREAD(Event_Set(&mtc->readEvent))
+
+ if (res2 != SZ_OK)
+ return res2;
+
+ if (res == SZ_OK)
+ {
+ CriticalSection_Enter(&mtc->cs);
+ bufIndex = mtc->freeBlockHead;
+ mtc->freeBlockHead = mtc->freeBlockList[bufIndex];
+ CriticalSection_Leave(&mtc->cs);
+
+ res = mtc->mtCallback->Code(mtc->mtCallbackObject, t->index, bufIndex,
+ mtc->inStream ? t->inBuf : inData, size, finished);
+
+ // MtProgress_Reinit(&mtc->mtProgress, t->index);
+
+ if (res != SZ_OK)
+ MtProgress_SetError(&mtc->mtProgress, res);
+ }
+
+ {
+ CMtCoderBlock *block = &mtc->blocks[bi];
+ block->res = res;
+ block->bufIndex = bufIndex;
+ block->finished = finished;
+ }
+
+ #ifdef MTCODER_USE_WRITE_THREAD
+ RINOK_THREAD(Event_Set(&mtc->writeEvents[bi]))
+ #else
+ {
+ unsigned wi;
+ {
+ CriticalSection_Enter(&mtc->cs);
+ wi = mtc->writeIndex;
+ if (wi == bi)
+ mtc->writeIndex = (unsigned)(int)-1;
+ else
+ mtc->ReadyBlocks[bi] = True;
+ CriticalSection_Leave(&mtc->cs);
+ }
+
+ if (wi != bi)
+ {
+ if (res != SZ_OK || finished)
+ return 0;
+ continue;
+ }
+
+ if (mtc->writeRes != SZ_OK)
+ res = mtc->writeRes;
+
+ for (;;)
+ {
+ if (res == SZ_OK && bufIndex != (unsigned)(int)-1)
+ {
+ res = mtc->mtCallback->Write(mtc->mtCallbackObject, bufIndex);
+ if (res != SZ_OK)
+ {
+ mtc->writeRes = res;
+ MtProgress_SetError(&mtc->mtProgress, res);
+ }
+ }
+
+ if (++wi >= mtc->numBlocksMax)
+ wi = 0;
+ {
+ BoolInt isReady;
+
+ CriticalSection_Enter(&mtc->cs);
+
+ if (bufIndex != (unsigned)(int)-1)
+ {
+ mtc->freeBlockList[bufIndex] = mtc->freeBlockHead;
+ mtc->freeBlockHead = bufIndex;
+ }
+
+ isReady = mtc->ReadyBlocks[wi];
+
+ if (isReady)
+ mtc->ReadyBlocks[wi] = False;
+ else
+ mtc->writeIndex = wi;
+
+ CriticalSection_Leave(&mtc->cs);
+
+ RINOK_THREAD(Semaphore_Release1(&mtc->blocksSemaphore))
+
+ if (!isReady)
+ break;
+ }
+
+ {
+ CMtCoderBlock *block = &mtc->blocks[wi];
+ if (res == SZ_OK && block->res != SZ_OK)
+ res = block->res;
+ bufIndex = block->bufIndex;
+ finished = block->finished;
+ }
+ }
+ }
+ #endif
+
+ if (finished || res != SZ_OK)
+ return 0;
}
}
-static THREAD_FUNC_RET_TYPE THREAD_FUNC_CALL_TYPE ThreadFunc(void *pp)
+
+static THREAD_FUNC_DECL ThreadFunc(void *pp)
{
- CMtThread *p = (CMtThread *)pp;
+ CMtCoderThread *t = (CMtCoderThread *)pp;
for (;;)
{
- Bool stop;
- CMtThread *next = GET_NEXT_THREAD(p);
- SRes res = MtThread_Process(p, &stop);
- if (res != SZ_OK)
+ if (Event_Wait(&t->startEvent) != 0)
+ return (THREAD_FUNC_RET_TYPE)SZ_ERROR_THREAD;
+ if (t->stop)
+ return 0;
{
- MtCoder_SetError(p->mtCoder, res);
- MtProgress_SetError(&p->mtCoder->mtProgress, res);
- next->stopReading = True;
- next->stopWriting = True;
- Event_Set(&next->canRead);
- Event_Set(&next->canWrite);
- return res;
+ SRes res = ThreadFunc2(t);
+ CMtCoder *mtc = t->mtCoder;
+ if (res != SZ_OK)
+ {
+ MtProgress_SetError(&mtc->mtProgress, res);
+ }
+
+ #ifndef MTCODER_USE_WRITE_THREAD
+ {
+ unsigned numFinished = (unsigned)InterlockedIncrement(&mtc->numFinishedThreads);
+ if (numFinished == mtc->numStartedThreads)
+ if (Event_Set(&mtc->finishedEvent) != 0)
+ return (THREAD_FUNC_RET_TYPE)SZ_ERROR_THREAD;
+ }
+ #endif
}
- if (stop)
- return 0;
}
}
-void MtCoder_Construct(CMtCoder* p)
+
+
+void MtCoder_Construct(CMtCoder *p)
{
unsigned i;
- p->alloc = 0;
- for (i = 0; i < NUM_MT_CODER_THREADS_MAX; i++)
+
+ p->blockSize = 0;
+ p->numThreadsMax = 0;
+ p->expectedDataSize = (UInt64)(Int64)-1;
+
+ p->inStream = NULL;
+ p->inData = NULL;
+ p->inDataSize = 0;
+
+ p->progress = NULL;
+ p->allocBig = NULL;
+
+ p->mtCallback = NULL;
+ p->mtCallbackObject = NULL;
+
+ p->allocatedBufsSize = 0;
+
+ Event_Construct(&p->readEvent);
+ Semaphore_Construct(&p->blocksSemaphore);
+
+ for (i = 0; i < MTCODER_THREADS_MAX; i++)
{
- CMtThread *t = &p->threads[i];
+ CMtCoderThread *t = &p->threads[i];
+ t->mtCoder = p;
t->index = i;
- CMtThread_Construct(t, p);
+ t->inBuf = NULL;
+ t->stop = False;
+ Event_Construct(&t->startEvent);
+ Thread_CONSTRUCT(&t->thread)
}
+
+ #ifdef MTCODER_USE_WRITE_THREAD
+ for (i = 0; i < MTCODER_BLOCKS_MAX; i++)
+ Event_Construct(&p->writeEvents[i]);
+ #else
+ Event_Construct(&p->finishedEvent);
+ #endif
+
CriticalSection_Init(&p->cs);
CriticalSection_Init(&p->mtProgress.cs);
}
-void MtCoder_Destruct(CMtCoder* p)
+
+
+
+static void MtCoder_Free(CMtCoder *p)
{
unsigned i;
- for (i = 0; i < NUM_MT_CODER_THREADS_MAX; i++)
- CMtThread_Destruct(&p->threads[i]);
+
+ /*
+ p->stopReading = True;
+ if (Event_IsCreated(&p->readEvent))
+ Event_Set(&p->readEvent);
+ */
+
+ for (i = 0; i < MTCODER_THREADS_MAX; i++)
+ MtCoderThread_Destruct(&p->threads[i]);
+
+ Event_Close(&p->readEvent);
+ Semaphore_Close(&p->blocksSemaphore);
+
+ #ifdef MTCODER_USE_WRITE_THREAD
+ for (i = 0; i < MTCODER_BLOCKS_MAX; i++)
+ Event_Close(&p->writeEvents[i]);
+ #else
+ Event_Close(&p->finishedEvent);
+ #endif
+}
+
+
+void MtCoder_Destruct(CMtCoder *p)
+{
+ MtCoder_Free(p);
+
CriticalSection_Delete(&p->cs);
CriticalSection_Delete(&p->mtProgress.cs);
}
+
SRes MtCoder_Code(CMtCoder *p)
{
- unsigned i, numThreads = p->numThreads;
+ unsigned numThreads = p->numThreadsMax;
+ unsigned numBlocksMax;
+ unsigned i;
SRes res = SZ_OK;
- p->res = SZ_OK;
+
+ if (numThreads > MTCODER_THREADS_MAX)
+ numThreads = MTCODER_THREADS_MAX;
+ numBlocksMax = MTCODER_GET_NUM_BLOCKS_FROM_THREADS(numThreads);
+
+ if (p->blockSize < ((UInt32)1 << 26)) numBlocksMax++;
+ if (p->blockSize < ((UInt32)1 << 24)) numBlocksMax++;
+ if (p->blockSize < ((UInt32)1 << 22)) numBlocksMax++;
+
+ if (numBlocksMax > MTCODER_BLOCKS_MAX)
+ numBlocksMax = MTCODER_BLOCKS_MAX;
+
+ if (p->blockSize != p->allocatedBufsSize)
+ {
+ for (i = 0; i < MTCODER_THREADS_MAX; i++)
+ {
+ CMtCoderThread *t = &p->threads[i];
+ if (t->inBuf)
+ {
+ ISzAlloc_Free(p->allocBig, t->inBuf);
+ t->inBuf = NULL;
+ }
+ }
+ p->allocatedBufsSize = p->blockSize;
+ }
+
+ p->readRes = SZ_OK;
MtProgress_Init(&p->mtProgress, p->progress);
- for (i = 0; i < numThreads; i++)
+ #ifdef MTCODER_USE_WRITE_THREAD
+ for (i = 0; i < numBlocksMax; i++)
+ {
+ RINOK_THREAD(AutoResetEvent_OptCreate_And_Reset(&p->writeEvents[i]))
+ }
+ #else
+ RINOK_THREAD(AutoResetEvent_OptCreate_And_Reset(&p->finishedEvent))
+ #endif
+
+ {
+ RINOK_THREAD(AutoResetEvent_OptCreate_And_Reset(&p->readEvent))
+ RINOK_THREAD(Semaphore_OptCreateInit(&p->blocksSemaphore, numBlocksMax, numBlocksMax))
+ }
+
+ for (i = 0; i < MTCODER_BLOCKS_MAX - 1; i++)
+ p->freeBlockList[i] = i + 1;
+ p->freeBlockList[MTCODER_BLOCKS_MAX - 1] = (unsigned)(int)-1;
+ p->freeBlockHead = 0;
+
+ p->readProcessed = 0;
+ p->blockIndex = 0;
+ p->numBlocksMax = numBlocksMax;
+ p->stopReading = False;
+
+ #ifndef MTCODER_USE_WRITE_THREAD
+ p->writeIndex = 0;
+ p->writeRes = SZ_OK;
+ for (i = 0; i < MTCODER_BLOCKS_MAX; i++)
+ p->ReadyBlocks[i] = False;
+ p->numFinishedThreads = 0;
+ #endif
+
+ p->numStartedThreadsLimit = numThreads;
+ p->numStartedThreads = 0;
+
+ // for (i = 0; i < numThreads; i++)
{
- RINOK(CMtThread_Prepare(&p->threads[i]));
+ CMtCoderThread *nextThread = &p->threads[p->numStartedThreads++];
+ RINOK(MtCoderThread_CreateAndStart(nextThread))
}
- for (i = 0; i < numThreads; i++)
+ RINOK_THREAD(Event_Set(&p->readEvent))
+
+ #ifdef MTCODER_USE_WRITE_THREAD
{
- CMtThread *t = &p->threads[i];
- CLoopThread *lt = &t->thread;
+ unsigned bi = 0;
- if (!Thread_WasCreated(&lt->thread))
+ for (;; bi++)
{
- lt->func = ThreadFunc;
- lt->param = t;
+ if (bi >= numBlocksMax)
+ bi = 0;
+
+ RINOK_THREAD(Event_Wait(&p->writeEvents[bi]))
- if (LoopThread_Create(lt) != SZ_OK)
{
- res = SZ_ERROR_THREAD;
- break;
+ const CMtCoderBlock *block = &p->blocks[bi];
+ unsigned bufIndex = block->bufIndex;
+ BoolInt finished = block->finished;
+ if (res == SZ_OK && block->res != SZ_OK)
+ res = block->res;
+
+ if (bufIndex != (unsigned)(int)-1)
+ {
+ if (res == SZ_OK)
+ {
+ res = p->mtCallback->Write(p->mtCallbackObject, bufIndex);
+ if (res != SZ_OK)
+ MtProgress_SetError(&p->mtProgress, res);
+ }
+
+ CriticalSection_Enter(&p->cs);
+ {
+ p->freeBlockList[bufIndex] = p->freeBlockHead;
+ p->freeBlockHead = bufIndex;
+ }
+ CriticalSection_Leave(&p->cs);
+ }
+
+ RINOK_THREAD(Semaphore_Release1(&p->blocksSemaphore))
+
+ if (finished)
+ break;
}
}
}
+ #else
+ {
+ WRes wres = Event_Wait(&p->finishedEvent);
+ res = MY_SRes_HRESULT_FROM_WRes(wres);
+ }
+ #endif
if (res == SZ_OK)
- {
- unsigned j;
- for (i = 0; i < numThreads; i++)
- {
- CMtThread *t = &p->threads[i];
- if (LoopThread_StartSubThread(&t->thread) != SZ_OK)
- {
- res = SZ_ERROR_THREAD;
- p->threads[0].stopReading = True;
- break;
- }
- }
+ res = p->readRes;
- Event_Set(&p->threads[0].canWrite);
- Event_Set(&p->threads[0].canRead);
+ if (res == SZ_OK)
+ res = p->mtProgress.res;
- for (j = 0; j < i; j++)
- LoopThread_WaitSubThread(&p->threads[j].thread);
- }
+ #ifndef MTCODER_USE_WRITE_THREAD
+ if (res == SZ_OK)
+ res = p->writeRes;
+ #endif
- for (i = 0; i < numThreads; i++)
- CMtThread_CloseEvents(&p->threads[i]);
- return (res == SZ_OK) ? p->res : res;
+ if (res != SZ_OK)
+ MtCoder_Free(p);
+ return res;
}
+
+#endif
+
+#undef RINOK_THREAD