[pcsx_rearmed.git] / deps / libchdr / deps / lzma-22.01 / src / MtCoder.c

/* MtCoder.c -- Multi-thread Coder\r
2021-12-21 : Igor Pavlov : Public domain */\r
\r
#include "Precomp.h"\r
\r
#include "MtCoder.h"\r
\r
#ifndef _7ZIP_ST\r
\r
static SRes MtProgressThunk_Progress(const ICompressProgress *pp, UInt64 inSize, UInt64 outSize)\r
{\r
  CMtProgressThunk *thunk = CONTAINER_FROM_VTBL(pp, CMtProgressThunk, vt);\r
  UInt64 inSize2 = 0;\r
  UInt64 outSize2 = 0;\r
  if (inSize != (UInt64)(Int64)-1)\r
  {\r
    inSize2 = inSize - thunk->inSize;\r
    thunk->inSize = inSize;\r
  }\r
  if (outSize != (UInt64)(Int64)-1)\r
  {\r
    outSize2 = outSize - thunk->outSize;\r
    thunk->outSize = outSize;\r
  }\r
  return MtProgress_ProgressAdd(thunk->mtProgress, inSize2, outSize2);\r
}\r
\r
\r
void MtProgressThunk_CreateVTable(CMtProgressThunk *p)\r
{\r
  p->vt.Progress = MtProgressThunk_Progress;\r
}\r
\r
\r
\r
#define RINOK_THREAD(x) { if ((x) != 0) return SZ_ERROR_THREAD; }\r
\r
\r
static WRes ArEvent_OptCreate_And_Reset(CEvent *p)\r
{\r
  if (Event_IsCreated(p))\r
    return Event_Reset(p);\r
  return AutoResetEvent_CreateNotSignaled(p);\r
}\r
\r
\r
static THREAD_FUNC_DECL ThreadFunc(void *pp);\r
\r
\r
static SRes MtCoderThread_CreateAndStart(CMtCoderThread *t)\r
{\r
  WRes wres = ArEvent_OptCreate_And_Reset(&t->startEvent);\r
  if (wres == 0)\r
  {\r
    t->stop = False;\r
    if (!Thread_WasCreated(&t->thread))\r
      wres = Thread_Create(&t->thread, ThreadFunc, t);\r
    if (wres == 0)\r
      wres = Event_Set(&t->startEvent);\r
  }\r
  if (wres == 0)\r
    return SZ_OK;\r
  return MY_SRes_HRESULT_FROM_WRes(wres);\r
}\r
\r
\r
static void MtCoderThread_Destruct(CMtCoderThread *t)\r
{\r
  if (Thread_WasCreated(&t->thread))\r
  {\r
    t->stop = 1;\r
    Event_Set(&t->startEvent);\r
    Thread_Wait_Close(&t->thread);\r
  }\r
\r
  Event_Close(&t->startEvent);\r
\r
  if (t->inBuf)\r
  {\r
    ISzAlloc_Free(t->mtCoder->allocBig, t->inBuf);\r
    t->inBuf = NULL;\r
  }\r
}\r
\r
\r
\r
static SRes FullRead(ISeqInStream *stream, Byte *data, size_t *processedSize)\r
{\r
  size_t size = *processedSize;\r
  *processedSize = 0;\r
  while (size != 0)\r
  {\r
    size_t cur = size;\r
    SRes res = ISeqInStream_Read(stream, data, &cur);\r
    *processedSize += cur;\r
    data += cur;\r
    size -= cur;\r
    RINOK(res);\r
    if (cur == 0)\r
      return SZ_OK;\r
  }\r
  return SZ_OK;\r
}\r
\r
\r
/*\r
  ThreadFunc2() returns:\r
  SZ_OK           - in all normal cases (even for stream error or memory allocation error)\r
  SZ_ERROR_THREAD - in case of failure in system synch function\r
*/\r
\r
static SRes ThreadFunc2(CMtCoderThread *t)\r
{\r
  CMtCoder *mtc = t->mtCoder;\r
\r
  for (;;)\r
  {\r
    unsigned bi;\r
    SRes res;\r
    SRes res2;\r
    BoolInt finished;\r
    unsigned bufIndex;\r
    size_t size;\r
    const Byte *inData;\r
    UInt64 readProcessed = 0;\r
    \r
    RINOK_THREAD(Event_Wait(&mtc->readEvent))\r
\r
    /* after Event_Wait(&mtc->readEvent) we must call Event_Set(&mtc->readEvent) in any case to unlock another threads */\r
\r
    if (mtc->stopReading)\r
    {\r
      return Event_Set(&mtc->readEvent) == 0 ? SZ_OK : SZ_ERROR_THREAD;\r
    }\r
\r
    res = MtProgress_GetError(&mtc->mtProgress);\r
    \r
    size = 0;\r
    inData = NULL;\r
    finished = True;\r
\r
    if (res == SZ_OK)\r
    {\r
      size = mtc->blockSize;\r
      if (mtc->inStream)\r
      {\r
        if (!t->inBuf)\r
        {\r
          t->inBuf = (Byte *)ISzAlloc_Alloc(mtc->allocBig, mtc->blockSize);\r
          if (!t->inBuf)\r
            res = SZ_ERROR_MEM;\r
        }\r
        if (res == SZ_OK)\r
        {\r
          res = FullRead(mtc->inStream, t->inBuf, &size);\r
          readProcessed = mtc->readProcessed + size;\r
          mtc->readProcessed = readProcessed;\r
        }\r
        if (res != SZ_OK)\r
        {\r
          mtc->readRes = res;\r
          /* after reading error - we can stop encoding of previous blocks */\r
          MtProgress_SetError(&mtc->mtProgress, res);\r
        }\r
        else\r
          finished = (size != mtc->blockSize);\r
      }\r
      else\r
      {\r
        size_t rem;\r
        readProcessed = mtc->readProcessed;\r
        rem = mtc->inDataSize - (size_t)readProcessed;\r
        if (size > rem)\r
          size = rem;\r
        inData = mtc->inData + (size_t)readProcessed;\r
        readProcessed += size;\r
        mtc->readProcessed = readProcessed;\r
        finished = (mtc->inDataSize == (size_t)readProcessed);\r
      }\r
    }\r
\r
    /* we must get some block from blocksSemaphore before Event_Set(&mtc->readEvent) */\r
\r
    res2 = SZ_OK;\r
\r
    if (Semaphore_Wait(&mtc->blocksSemaphore) != 0)\r
    {\r
      res2 = SZ_ERROR_THREAD;\r
      if (res == SZ_OK)\r
      {\r
        res = res2;\r
        // MtProgress_SetError(&mtc->mtProgress, res);\r
      }\r
    }\r
\r
    bi = mtc->blockIndex;\r
\r
    if (++mtc->blockIndex >= mtc->numBlocksMax)\r
      mtc->blockIndex = 0;\r
\r
    bufIndex = (unsigned)(int)-1;\r
\r
    if (res == SZ_OK)\r
      res = MtProgress_GetError(&mtc->mtProgress);\r
\r
    if (res != SZ_OK)\r
      finished = True;\r
\r
    if (!finished)\r
    {\r
      if (mtc->numStartedThreads < mtc->numStartedThreadsLimit\r
          && mtc->expectedDataSize != readProcessed)\r
      {\r
        res = MtCoderThread_CreateAndStart(&mtc->threads[mtc->numStartedThreads]);\r
        if (res == SZ_OK)\r
          mtc->numStartedThreads++;\r
        else\r
        {\r
          MtProgress_SetError(&mtc->mtProgress, res);\r
          finished = True;\r
        }\r
      }\r
    }\r
\r
    if (finished)\r
      mtc->stopReading = True;\r
\r
    RINOK_THREAD(Event_Set(&mtc->readEvent))\r
\r
    if (res2 != SZ_OK)\r
      return res2;\r
\r
    if (res == SZ_OK)\r
    {\r
      CriticalSection_Enter(&mtc->cs);\r
      bufIndex = mtc->freeBlockHead;\r
      mtc->freeBlockHead = mtc->freeBlockList[bufIndex];\r
      CriticalSection_Leave(&mtc->cs);\r
      \r
      res = mtc->mtCallback->Code(mtc->mtCallbackObject, t->index, bufIndex,\r
          mtc->inStream ? t->inBuf : inData, size, finished);\r
      \r
      // MtProgress_Reinit(&mtc->mtProgress, t->index);\r
\r
      if (res != SZ_OK)\r
        MtProgress_SetError(&mtc->mtProgress, res);\r
    }\r
\r
    {\r
      CMtCoderBlock *block = &mtc->blocks[bi];\r
      block->res = res;\r
      block->bufIndex = bufIndex;\r
      block->finished = finished;\r
    }\r
    \r
    #ifdef MTCODER__USE_WRITE_THREAD\r
      RINOK_THREAD(Event_Set(&mtc->writeEvents[bi]))\r
    #else\r
    {\r
      unsigned wi;\r
      {\r
        CriticalSection_Enter(&mtc->cs);\r
        wi = mtc->writeIndex;\r
        if (wi == bi)\r
          mtc->writeIndex = (unsigned)(int)-1;\r
        else\r
          mtc->ReadyBlocks[bi] = True;\r
        CriticalSection_Leave(&mtc->cs);\r
      }\r
\r
      if (wi != bi)\r
      {\r
        if (res != SZ_OK || finished)\r
          return 0;\r
        continue;\r
      }\r
\r
      if (mtc->writeRes != SZ_OK)\r
        res = mtc->writeRes;\r
\r
      for (;;)\r
      {\r
        if (res == SZ_OK && bufIndex != (unsigned)(int)-1)\r
        {\r
          res = mtc->mtCallback->Write(mtc->mtCallbackObject, bufIndex);\r
          if (res != SZ_OK)\r
          {\r
            mtc->writeRes = res;\r
            MtProgress_SetError(&mtc->mtProgress, res);\r
          }\r
        }\r
\r
        if (++wi >= mtc->numBlocksMax)\r
          wi = 0;\r
        {\r
          BoolInt isReady;\r
\r
          CriticalSection_Enter(&mtc->cs);\r
          \r
          if (bufIndex != (unsigned)(int)-1)\r
          {\r
            mtc->freeBlockList[bufIndex] = mtc->freeBlockHead;\r
            mtc->freeBlockHead = bufIndex;\r
          }\r
          \r
          isReady = mtc->ReadyBlocks[wi];\r
          \r
          if (isReady)\r
            mtc->ReadyBlocks[wi] = False;\r
          else\r
            mtc->writeIndex = wi;\r
          \r
          CriticalSection_Leave(&mtc->cs);\r
\r
          RINOK_THREAD(Semaphore_Release1(&mtc->blocksSemaphore))\r
\r
          if (!isReady)\r
            break;\r
        }\r
\r
        {\r
          CMtCoderBlock *block = &mtc->blocks[wi];\r
          if (res == SZ_OK && block->res != SZ_OK)\r
            res = block->res;\r
          bufIndex = block->bufIndex;\r
          finished = block->finished;\r
        }\r
      }\r
    }\r
    #endif\r
      \r
    if (finished || res != SZ_OK)\r
      return 0;\r
  }\r
}\r
\r
\r
static THREAD_FUNC_DECL ThreadFunc(void *pp)\r
{\r
  CMtCoderThread *t = (CMtCoderThread *)pp;\r
  for (;;)\r
  {\r
    if (Event_Wait(&t->startEvent) != 0)\r
      return (THREAD_FUNC_RET_TYPE)SZ_ERROR_THREAD;\r
    if (t->stop)\r
      return 0;\r
    {\r
      SRes res = ThreadFunc2(t);\r
      CMtCoder *mtc = t->mtCoder;\r
      if (res != SZ_OK)\r
      {\r
        MtProgress_SetError(&mtc->mtProgress, res);\r
      }\r
      \r
      #ifndef MTCODER__USE_WRITE_THREAD\r
      {\r
        unsigned numFinished = (unsigned)InterlockedIncrement(&mtc->numFinishedThreads);\r
        if (numFinished == mtc->numStartedThreads)\r
          if (Event_Set(&mtc->finishedEvent) != 0)\r
            return (THREAD_FUNC_RET_TYPE)SZ_ERROR_THREAD;\r
      }\r
      #endif\r
    }\r
  }\r
}\r
\r
\r
\r
void MtCoder_Construct(CMtCoder *p)\r
{\r
  unsigned i;\r
  \r
  p->blockSize = 0;\r
  p->numThreadsMax = 0;\r
  p->expectedDataSize = (UInt64)(Int64)-1;\r
\r
  p->inStream = NULL;\r
  p->inData = NULL;\r
  p->inDataSize = 0;\r
\r
  p->progress = NULL;\r
  p->allocBig = NULL;\r
\r
  p->mtCallback = NULL;\r
  p->mtCallbackObject = NULL;\r
\r
  p->allocatedBufsSize = 0;\r
\r
  Event_Construct(&p->readEvent);\r
  Semaphore_Construct(&p->blocksSemaphore);\r
\r
  for (i = 0; i < MTCODER__THREADS_MAX; i++)\r
  {\r
    CMtCoderThread *t = &p->threads[i];\r
    t->mtCoder = p;\r
    t->index = i;\r
    t->inBuf = NULL;\r
    t->stop = False;\r
    Event_Construct(&t->startEvent);\r
    Thread_Construct(&t->thread);\r
  }\r
\r
  #ifdef MTCODER__USE_WRITE_THREAD\r
    for (i = 0; i < MTCODER__BLOCKS_MAX; i++)\r
      Event_Construct(&p->writeEvents[i]);\r
  #else\r
    Event_Construct(&p->finishedEvent);\r
  #endif\r
\r
  CriticalSection_Init(&p->cs);\r
  CriticalSection_Init(&p->mtProgress.cs);\r
}\r
\r
\r
\r
\r
static void MtCoder_Free(CMtCoder *p)\r
{\r
  unsigned i;\r
\r
  /*\r
  p->stopReading = True;\r
  if (Event_IsCreated(&p->readEvent))\r
    Event_Set(&p->readEvent);\r
  */\r
\r
  for (i = 0; i < MTCODER__THREADS_MAX; i++)\r
    MtCoderThread_Destruct(&p->threads[i]);\r
\r
  Event_Close(&p->readEvent);\r
  Semaphore_Close(&p->blocksSemaphore);\r
\r
  #ifdef MTCODER__USE_WRITE_THREAD\r
    for (i = 0; i < MTCODER__BLOCKS_MAX; i++)\r
      Event_Close(&p->writeEvents[i]);\r
  #else\r
    Event_Close(&p->finishedEvent);\r
  #endif\r
}\r
\r
\r
void MtCoder_Destruct(CMtCoder *p)\r
{\r
  MtCoder_Free(p);\r
\r
  CriticalSection_Delete(&p->cs);\r
  CriticalSection_Delete(&p->mtProgress.cs);\r
}\r
\r
\r
SRes MtCoder_Code(CMtCoder *p)\r
{\r
  unsigned numThreads = p->numThreadsMax;\r
  unsigned numBlocksMax;\r
  unsigned i;\r
  SRes res = SZ_OK;\r
\r
  if (numThreads > MTCODER__THREADS_MAX)\r
    numThreads = MTCODER__THREADS_MAX;\r
  numBlocksMax = MTCODER__GET_NUM_BLOCKS_FROM_THREADS(numThreads);\r
  \r
  if (p->blockSize < ((UInt32)1 << 26)) numBlocksMax++;\r
  if (p->blockSize < ((UInt32)1 << 24)) numBlocksMax++;\r
  if (p->blockSize < ((UInt32)1 << 22)) numBlocksMax++;\r
\r
  if (numBlocksMax > MTCODER__BLOCKS_MAX)\r
    numBlocksMax = MTCODER__BLOCKS_MAX;\r
\r
  if (p->blockSize != p->allocatedBufsSize)\r
  {\r
    for (i = 0; i < MTCODER__THREADS_MAX; i++)\r
    {\r
      CMtCoderThread *t = &p->threads[i];\r
      if (t->inBuf)\r
      {\r
        ISzAlloc_Free(p->allocBig, t->inBuf);\r
        t->inBuf = NULL;\r
      }\r
    }\r
    p->allocatedBufsSize = p->blockSize;\r
  }\r
\r
  p->readRes = SZ_OK;\r
\r
  MtProgress_Init(&p->mtProgress, p->progress);\r
\r
  #ifdef MTCODER__USE_WRITE_THREAD\r
    for (i = 0; i < numBlocksMax; i++)\r
    {\r
      RINOK_THREAD(ArEvent_OptCreate_And_Reset(&p->writeEvents[i]));\r
    }\r
  #else\r
    RINOK_THREAD(ArEvent_OptCreate_And_Reset(&p->finishedEvent));\r
  #endif\r
\r
  {\r
    RINOK_THREAD(ArEvent_OptCreate_And_Reset(&p->readEvent));\r
    RINOK_THREAD(Semaphore_OptCreateInit(&p->blocksSemaphore, numBlocksMax, numBlocksMax));\r
  }\r
\r
  for (i = 0; i < MTCODER__BLOCKS_MAX - 1; i++)\r
    p->freeBlockList[i] = i + 1;\r
  p->freeBlockList[MTCODER__BLOCKS_MAX - 1] = (unsigned)(int)-1;\r
  p->freeBlockHead = 0;\r
\r
  p->readProcessed = 0;\r
  p->blockIndex = 0;\r
  p->numBlocksMax = numBlocksMax;\r
  p->stopReading = False;\r
\r
  #ifndef MTCODER__USE_WRITE_THREAD\r
    p->writeIndex = 0;\r
    p->writeRes = SZ_OK;\r
    for (i = 0; i < MTCODER__BLOCKS_MAX; i++)\r
      p->ReadyBlocks[i] = False;\r
    p->numFinishedThreads = 0;\r
  #endif\r
\r
  p->numStartedThreadsLimit = numThreads;\r
  p->numStartedThreads = 0;\r
\r
  // for (i = 0; i < numThreads; i++)\r
  {\r
    CMtCoderThread *nextThread = &p->threads[p->numStartedThreads++];\r
    RINOK(MtCoderThread_CreateAndStart(nextThread));\r
  }\r
\r
  RINOK_THREAD(Event_Set(&p->readEvent))\r
\r
  #ifdef MTCODER__USE_WRITE_THREAD\r
  {\r
    unsigned bi = 0;\r
\r
    for (;; bi++)\r
    {\r
      if (bi >= numBlocksMax)\r
        bi = 0;\r
\r
      RINOK_THREAD(Event_Wait(&p->writeEvents[bi]))\r
\r
      {\r
        const CMtCoderBlock *block = &p->blocks[bi];\r
        unsigned bufIndex = block->bufIndex;\r
        BoolInt finished = block->finished;\r
        if (res == SZ_OK && block->res != SZ_OK)\r
          res = block->res;\r
\r
        if (bufIndex != (unsigned)(int)-1)\r
        {\r
          if (res == SZ_OK)\r
          {\r
            res = p->mtCallback->Write(p->mtCallbackObject, bufIndex);\r
            if (res != SZ_OK)\r
              MtProgress_SetError(&p->mtProgress, res);\r
          }\r
          \r
          CriticalSection_Enter(&p->cs);\r
          {\r
            p->freeBlockList[bufIndex] = p->freeBlockHead;\r
            p->freeBlockHead = bufIndex;\r
          }\r
          CriticalSection_Leave(&p->cs);\r
        }\r
        \r
        RINOK_THREAD(Semaphore_Release1(&p->blocksSemaphore))\r
\r
        if (finished)\r
          break;\r
      }\r
    }\r
  }\r
  #else\r
  {\r
    WRes wres = Event_Wait(&p->finishedEvent);\r
    res = MY_SRes_HRESULT_FROM_WRes(wres);\r
  }\r
  #endif\r
\r
  if (res == SZ_OK)\r
    res = p->readRes;\r
\r
  if (res == SZ_OK)\r
    res = p->mtProgress.res;\r
\r
  #ifndef MTCODER__USE_WRITE_THREAD\r
    if (res == SZ_OK)\r
      res = p->writeRes;\r
  #endif\r
\r
  if (res != SZ_OK)\r
    MtCoder_Free(p);\r
  return res;\r
}\r
\r
#endif\r
Commit	Line	Data
9e052883	1	/* MtCoder.c -- Multi-thread Coder\r
	2	2021-12-21 : Igor Pavlov : Public domain */\r
	3	\r
	4	#include "Precomp.h"\r
	5	\r
	6	#include "MtCoder.h"\r
	7	\r
	8	#ifndef _7ZIP_ST\r
	9	\r
	10	static SRes MtProgressThunk_Progress(const ICompressProgress *pp, UInt64 inSize, UInt64 outSize)\r
	11	{\r
	12	CMtProgressThunk *thunk = CONTAINER_FROM_VTBL(pp, CMtProgressThunk, vt);\r
	13	UInt64 inSize2 = 0;\r
	14	UInt64 outSize2 = 0;\r
	15	if (inSize != (UInt64)(Int64)-1)\r
	16	{\r
	17	inSize2 = inSize - thunk->inSize;\r
	18	thunk->inSize = inSize;\r
	19	}\r
	20	if (outSize != (UInt64)(Int64)-1)\r
	21	{\r
	22	outSize2 = outSize - thunk->outSize;\r
	23	thunk->outSize = outSize;\r
	24	}\r
	25	return MtProgress_ProgressAdd(thunk->mtProgress, inSize2, outSize2);\r
	26	}\r
	27	\r
	28	\r
	29	void MtProgressThunk_CreateVTable(CMtProgressThunk *p)\r
	30	{\r
	31	p->vt.Progress = MtProgressThunk_Progress;\r
	32	}\r
	33	\r
	34	\r
	35	\r
	36	#define RINOK_THREAD(x) { if ((x) != 0) return SZ_ERROR_THREAD; }\r
	37	\r
	38	\r
	39	static WRes ArEvent_OptCreate_And_Reset(CEvent *p)\r
	40	{\r
	41	if (Event_IsCreated(p))\r
	42	return Event_Reset(p);\r
	43	return AutoResetEvent_CreateNotSignaled(p);\r
	44	}\r
	45	\r
	46	\r
	47	static THREAD_FUNC_DECL ThreadFunc(void *pp);\r
	48	\r
	49	\r
	50	static SRes MtCoderThread_CreateAndStart(CMtCoderThread *t)\r
	51	{\r
	52	WRes wres = ArEvent_OptCreate_And_Reset(&t->startEvent);\r
	53	if (wres == 0)\r
	54	{\r
	55	t->stop = False;\r
	56	if (!Thread_WasCreated(&t->thread))\r
	57	wres = Thread_Create(&t->thread, ThreadFunc, t);\r
	58	if (wres == 0)\r
	59	wres = Event_Set(&t->startEvent);\r
	60	}\r
	61	if (wres == 0)\r
	62	return SZ_OK;\r
	63	return MY_SRes_HRESULT_FROM_WRes(wres);\r
	64	}\r
65	\r
66	\r
67	static void MtCoderThread_Destruct(CMtCoderThread *t)\r
68	{\r
69	if (Thread_WasCreated(&t->thread))\r
70	{\r
71	t->stop = 1;\r
72	Event_Set(&t->startEvent);\r
73	Thread_Wait_Close(&t->thread);\r
74	}\r
75	\r
76	Event_Close(&t->startEvent);\r
77	\r
78	if (t->inBuf)\r
79	{\r
80	ISzAlloc_Free(t->mtCoder->allocBig, t->inBuf);\r
81	t->inBuf = NULL;\r
82	}\r
83	}\r
84	\r
85	\r
86	\r
87	static SRes FullRead(ISeqInStream stream, Byte data, size_t *processedSize)\r
88	{\r
89	size_t size = *processedSize;\r
90	*processedSize = 0;\r
91	while (size != 0)\r
92	{\r
93	size_t cur = size;\r
94	SRes res = ISeqInStream_Read(stream, data, &cur);\r
95	*processedSize += cur;\r
96	data += cur;\r
97	size -= cur;\r
98	RINOK(res);\r
99	if (cur == 0)\r
100	return SZ_OK;\r
101	}\r
102	return SZ_OK;\r
103	}\r
104	\r
105	\r
106	/*\r
107	ThreadFunc2() returns:\r
108	SZ_OK - in all normal cases (even for stream error or memory allocation error)\r
109	SZ_ERROR_THREAD - in case of failure in system synch function\r
110	*/\r
111	\r
112	static SRes ThreadFunc2(CMtCoderThread *t)\r
113	{\r
114	CMtCoder *mtc = t->mtCoder;\r
115	\r
116	for (;;)\r
117	{\r
118	unsigned bi;\r
119	SRes res;\r
120	SRes res2;\r
121	BoolInt finished;\r
122	unsigned bufIndex;\r
123	size_t size;\r
124	const Byte *inData;\r
125	UInt64 readProcessed = 0;\r
126	\r
127	RINOK_THREAD(Event_Wait(&mtc->readEvent))\r
128	\r
129	/* after Event_Wait(&mtc->readEvent) we must call Event_Set(&mtc->readEvent) in any case to unlock another threads */\r
130	\r
131	if (mtc->stopReading)\r
132	{\r
133	return Event_Set(&mtc->readEvent) == 0 ? SZ_OK : SZ_ERROR_THREAD;\r
134	}\r
135	\r
136	res = MtProgress_GetError(&mtc->mtProgress);\r
137	\r
138	size = 0;\r
139	inData = NULL;\r
140	finished = True;\r
141	\r
142	if (res == SZ_OK)\r
143	{\r
144	size = mtc->blockSize;\r
145	if (mtc->inStream)\r
146	{\r
147	if (!t->inBuf)\r
148	{\r
149	t->inBuf = (Byte *)ISzAlloc_Alloc(mtc->allocBig, mtc->blockSize);\r
150	if (!t->inBuf)\r
151	res = SZ_ERROR_MEM;\r
152	}\r
153	if (res == SZ_OK)\r
154	{\r
155	res = FullRead(mtc->inStream, t->inBuf, &size);\r
156	readProcessed = mtc->readProcessed + size;\r
157	mtc->readProcessed = readProcessed;\r
158	}\r
159	if (res != SZ_OK)\r
160	{\r
161	mtc->readRes = res;\r
162	/* after reading error - we can stop encoding of previous blocks */\r
163	MtProgress_SetError(&mtc->mtProgress, res);\r
164	}\r
165	else\r
166	finished = (size != mtc->blockSize);\r
167	}\r
168	else\r
169	{\r
170	size_t rem;\r
171	readProcessed = mtc->readProcessed;\r
172	rem = mtc->inDataSize - (size_t)readProcessed;\r
173	if (size > rem)\r
174	size = rem;\r
175	inData = mtc->inData + (size_t)readProcessed;\r
176	readProcessed += size;\r
177	mtc->readProcessed = readProcessed;\r
178	finished = (mtc->inDataSize == (size_t)readProcessed);\r
179	}\r
180	}\r
181	\r
182	/* we must get some block from blocksSemaphore before Event_Set(&mtc->readEvent) */\r
183	\r
184	res2 = SZ_OK;\r
185	\r
186	if (Semaphore_Wait(&mtc->blocksSemaphore) != 0)\r
187	{\r
188	res2 = SZ_ERROR_THREAD;\r
189	if (res == SZ_OK)\r
190	{\r
191	res = res2;\r
192	// MtProgress_SetError(&mtc->mtProgress, res);\r
193	}\r
194	}\r
195	\r
196	bi = mtc->blockIndex;\r
197	\r
198	if (++mtc->blockIndex >= mtc->numBlocksMax)\r
199	mtc->blockIndex = 0;\r
200	\r
201	bufIndex = (unsigned)(int)-1;\r
202	\r
203	if (res == SZ_OK)\r
204	res = MtProgress_GetError(&mtc->mtProgress);\r
205	\r
206	if (res != SZ_OK)\r
207	finished = True;\r
208	\r
209	if (!finished)\r
210	{\r
211	if (mtc->numStartedThreads < mtc->numStartedThreadsLimit\r
212	&& mtc->expectedDataSize != readProcessed)\r
213	{\r
214	res = MtCoderThread_CreateAndStart(&mtc->threads[mtc->numStartedThreads]);\r
215	if (res == SZ_OK)\r
216	mtc->numStartedThreads++;\r
217	else\r
218	{\r
219	MtProgress_SetError(&mtc->mtProgress, res);\r
220	finished = True;\r
221	}\r
222	}\r
223	}\r
224	\r
225	if (finished)\r
226	mtc->stopReading = True;\r
227	\r
228	RINOK_THREAD(Event_Set(&mtc->readEvent))\r
229	\r
230	if (res2 != SZ_OK)\r
231	return res2;\r
232	\r
233	if (res == SZ_OK)\r
234	{\r
235	CriticalSection_Enter(&mtc->cs);\r
236	bufIndex = mtc->freeBlockHead;\r
237	mtc->freeBlockHead = mtc->freeBlockList[bufIndex];\r
238	CriticalSection_Leave(&mtc->cs);\r
239	\r
240	res = mtc->mtCallback->Code(mtc->mtCallbackObject, t->index, bufIndex,\r
241	mtc->inStream ? t->inBuf : inData, size, finished);\r
242	\r
243	// MtProgress_Reinit(&mtc->mtProgress, t->index);\r
244	\r
245	if (res != SZ_OK)\r
246	MtProgress_SetError(&mtc->mtProgress, res);\r
247	}\r
248	\r
249	{\r
250	CMtCoderBlock *block = &mtc->blocks[bi];\r
251	block->res = res;\r
252	block->bufIndex = bufIndex;\r
253	block->finished = finished;\r
254	}\r
255	\r
256	#ifdef MTCODER__USE_WRITE_THREAD\r
257	RINOK_THREAD(Event_Set(&mtc->writeEvents[bi]))\r
258	#else\r
259	{\r
260	unsigned wi;\r
261	{\r
262	CriticalSection_Enter(&mtc->cs);\r
263	wi = mtc->writeIndex;\r
264	if (wi == bi)\r
265	mtc->writeIndex = (unsigned)(int)-1;\r
266	else\r
267	mtc->ReadyBlocks[bi] = True;\r
268	CriticalSection_Leave(&mtc->cs);\r
269	}\r
270	\r
271	if (wi != bi)\r
272	{\r
273	if (res != SZ_OK \|\| finished)\r
274	return 0;\r
275	continue;\r
276	}\r
277	\r
278	if (mtc->writeRes != SZ_OK)\r
279	res = mtc->writeRes;\r
280	\r
281	for (;;)\r
282	{\r
283	if (res == SZ_OK && bufIndex != (unsigned)(int)-1)\r
284	{\r
285	res = mtc->mtCallback->Write(mtc->mtCallbackObject, bufIndex);\r
286	if (res != SZ_OK)\r
287	{\r
288	mtc->writeRes = res;\r
289	MtProgress_SetError(&mtc->mtProgress, res);\r
290	}\r
291	}\r
292	\r
293	if (++wi >= mtc->numBlocksMax)\r
294	wi = 0;\r
295	{\r
296	BoolInt isReady;\r
297	\r
298	CriticalSection_Enter(&mtc->cs);\r
299	\r
300	if (bufIndex != (unsigned)(int)-1)\r
301	{\r
302	mtc->freeBlockList[bufIndex] = mtc->freeBlockHead;\r
303	mtc->freeBlockHead = bufIndex;\r
304	}\r
305	\r
306	isReady = mtc->ReadyBlocks[wi];\r
307	\r
308	if (isReady)\r
309	mtc->ReadyBlocks[wi] = False;\r
310	else\r
311	mtc->writeIndex = wi;\r
312	\r
313	CriticalSection_Leave(&mtc->cs);\r
314	\r
315	RINOK_THREAD(Semaphore_Release1(&mtc->blocksSemaphore))\r
316	\r
317	if (!isReady)\r
318	break;\r
319	}\r
320	\r
321	{\r
322	CMtCoderBlock *block = &mtc->blocks[wi];\r
323	if (res == SZ_OK && block->res != SZ_OK)\r
324	res = block->res;\r
325	bufIndex = block->bufIndex;\r
326	finished = block->finished;\r
327	}\r
328	}\r
329	}\r
330	#endif\r
331	\r
332	if (finished \|\| res != SZ_OK)\r
333	return 0;\r
334	}\r
335	}\r
336	\r
337	\r
338	static THREAD_FUNC_DECL ThreadFunc(void *pp)\r
339	{\r
340	CMtCoderThread t = (CMtCoderThread )pp;\r
341	for (;;)\r
342	{\r
343	if (Event_Wait(&t->startEvent) != 0)\r
344	return (THREAD_FUNC_RET_TYPE)SZ_ERROR_THREAD;\r
345	if (t->stop)\r
346	return 0;\r
347	{\r
348	SRes res = ThreadFunc2(t);\r
349	CMtCoder *mtc = t->mtCoder;\r
350	if (res != SZ_OK)\r
351	{\r
352	MtProgress_SetError(&mtc->mtProgress, res);\r
353	}\r
354	\r
355	#ifndef MTCODER__USE_WRITE_THREAD\r
356	{\r
357	unsigned numFinished = (unsigned)InterlockedIncrement(&mtc->numFinishedThreads);\r
358	if (numFinished == mtc->numStartedThreads)\r
359	if (Event_Set(&mtc->finishedEvent) != 0)\r
360	return (THREAD_FUNC_RET_TYPE)SZ_ERROR_THREAD;\r
361	}\r
362	#endif\r
363	}\r
364	}\r
365	}\r
366	\r
367	\r
368	\r
369	void MtCoder_Construct(CMtCoder *p)\r
370	{\r
371	unsigned i;\r
372	\r
373	p->blockSize = 0;\r
374	p->numThreadsMax = 0;\r
375	p->expectedDataSize = (UInt64)(Int64)-1;\r
376	\r
377	p->inStream = NULL;\r
378	p->inData = NULL;\r
379	p->inDataSize = 0;\r
380	\r
381	p->progress = NULL;\r
382	p->allocBig = NULL;\r
383	\r
384	p->mtCallback = NULL;\r
385	p->mtCallbackObject = NULL;\r
386	\r
387	p->allocatedBufsSize = 0;\r
388	\r
389	Event_Construct(&p->readEvent);\r
390	Semaphore_Construct(&p->blocksSemaphore);\r
391	\r
392	for (i = 0; i < MTCODER__THREADS_MAX; i++)\r
393	{\r
394	CMtCoderThread *t = &p->threads[i];\r
395	t->mtCoder = p;\r
396	t->index = i;\r
397	t->inBuf = NULL;\r
398	t->stop = False;\r
399	Event_Construct(&t->startEvent);\r
400	Thread_Construct(&t->thread);\r
401	}\r
402	\r
403	#ifdef MTCODER__USE_WRITE_THREAD\r
404	for (i = 0; i < MTCODER__BLOCKS_MAX; i++)\r
405	Event_Construct(&p->writeEvents[i]);\r
406	#else\r
407	Event_Construct(&p->finishedEvent);\r
408	#endif\r
409	\r
410	CriticalSection_Init(&p->cs);\r
411	CriticalSection_Init(&p->mtProgress.cs);\r
412	}\r
413	\r
414	\r
415	\r
416	\r
417	static void MtCoder_Free(CMtCoder *p)\r
418	{\r
419	unsigned i;\r
420	\r
421	/*\r
422	p->stopReading = True;\r
423	if (Event_IsCreated(&p->readEvent))\r
424	Event_Set(&p->readEvent);\r
425	*/\r
426	\r
427	for (i = 0; i < MTCODER__THREADS_MAX; i++)\r
428	MtCoderThread_Destruct(&p->threads[i]);\r
429	\r
430	Event_Close(&p->readEvent);\r
431	Semaphore_Close(&p->blocksSemaphore);\r
432	\r
433	#ifdef MTCODER__USE_WRITE_THREAD\r
434	for (i = 0; i < MTCODER__BLOCKS_MAX; i++)\r
435	Event_Close(&p->writeEvents[i]);\r
436	#else\r
437	Event_Close(&p->finishedEvent);\r
438	#endif\r
439	}\r
440	\r
441	\r
442	void MtCoder_Destruct(CMtCoder *p)\r
443	{\r
444	MtCoder_Free(p);\r
445	\r
446	CriticalSection_Delete(&p->cs);\r
447	CriticalSection_Delete(&p->mtProgress.cs);\r
448	}\r
449	\r
450	\r
451	SRes MtCoder_Code(CMtCoder *p)\r
452	{\r
453	unsigned numThreads = p->numThreadsMax;\r
454	unsigned numBlocksMax;\r
455	unsigned i;\r
456	SRes res = SZ_OK;\r
457	\r
458	if (numThreads > MTCODER__THREADS_MAX)\r
459	numThreads = MTCODER__THREADS_MAX;\r
460	numBlocksMax = MTCODER__GET_NUM_BLOCKS_FROM_THREADS(numThreads);\r
461	\r
462	if (p->blockSize < ((UInt32)1 << 26)) numBlocksMax++;\r
463	if (p->blockSize < ((UInt32)1 << 24)) numBlocksMax++;\r
464	if (p->blockSize < ((UInt32)1 << 22)) numBlocksMax++;\r
465	\r
466	if (numBlocksMax > MTCODER__BLOCKS_MAX)\r
467	numBlocksMax = MTCODER__BLOCKS_MAX;\r
468	\r
469	if (p->blockSize != p->allocatedBufsSize)\r
470	{\r
471	for (i = 0; i < MTCODER__THREADS_MAX; i++)\r
472	{\r
473	CMtCoderThread *t = &p->threads[i];\r
474	if (t->inBuf)\r
475	{\r
476	ISzAlloc_Free(p->allocBig, t->inBuf);\r
477	t->inBuf = NULL;\r
478	}\r
479	}\r
480	p->allocatedBufsSize = p->blockSize;\r
481	}\r
482	\r
483	p->readRes = SZ_OK;\r
484	\r
485	MtProgress_Init(&p->mtProgress, p->progress);\r
486	\r
487	#ifdef MTCODER__USE_WRITE_THREAD\r
488	for (i = 0; i < numBlocksMax; i++)\r
489	{\r
490	RINOK_THREAD(ArEvent_OptCreate_And_Reset(&p->writeEvents[i]));\r
491	}\r
492	#else\r
493	RINOK_THREAD(ArEvent_OptCreate_And_Reset(&p->finishedEvent));\r
494	#endif\r
495	\r
496	{\r
497	RINOK_THREAD(ArEvent_OptCreate_And_Reset(&p->readEvent));\r
498	RINOK_THREAD(Semaphore_OptCreateInit(&p->blocksSemaphore, numBlocksMax, numBlocksMax));\r
499	}\r
500	\r
501	for (i = 0; i < MTCODER__BLOCKS_MAX - 1; i++)\r
502	p->freeBlockList[i] = i + 1;\r
503	p->freeBlockList[MTCODER__BLOCKS_MAX - 1] = (unsigned)(int)-1;\r
504	p->freeBlockHead = 0;\r
505	\r
506	p->readProcessed = 0;\r
507	p->blockIndex = 0;\r
508	p->numBlocksMax = numBlocksMax;\r
509	p->stopReading = False;\r
510	\r
511	#ifndef MTCODER__USE_WRITE_THREAD\r
512	p->writeIndex = 0;\r
513	p->writeRes = SZ_OK;\r
514	for (i = 0; i < MTCODER__BLOCKS_MAX; i++)\r
515	p->ReadyBlocks[i] = False;\r
516	p->numFinishedThreads = 0;\r
517	#endif\r
518	\r
519	p->numStartedThreadsLimit = numThreads;\r
520	p->numStartedThreads = 0;\r
521	\r
522	// for (i = 0; i < numThreads; i++)\r
523	{\r
524	CMtCoderThread *nextThread = &p->threads[p->numStartedThreads++];\r
525	RINOK(MtCoderThread_CreateAndStart(nextThread));\r
526	}\r
527	\r
528	RINOK_THREAD(Event_Set(&p->readEvent))\r
529	\r
530	#ifdef MTCODER__USE_WRITE_THREAD\r
531	{\r
532	unsigned bi = 0;\r
533	\r
534	for (;; bi++)\r
535	{\r
536	if (bi >= numBlocksMax)\r
537	bi = 0;\r
538	\r
539	RINOK_THREAD(Event_Wait(&p->writeEvents[bi]))\r
540	\r
541	{\r
542	const CMtCoderBlock *block = &p->blocks[bi];\r
543	unsigned bufIndex = block->bufIndex;\r
544	BoolInt finished = block->finished;\r
545	if (res == SZ_OK && block->res != SZ_OK)\r
546	res = block->res;\r
547	\r
548	if (bufIndex != (unsigned)(int)-1)\r
549	{\r
550	if (res == SZ_OK)\r
551	{\r
552	res = p->mtCallback->Write(p->mtCallbackObject, bufIndex);\r
553	if (res != SZ_OK)\r
554	MtProgress_SetError(&p->mtProgress, res);\r
555	}\r
556	\r
557	CriticalSection_Enter(&p->cs);\r
558	{\r
559	p->freeBlockList[bufIndex] = p->freeBlockHead;\r
560	p->freeBlockHead = bufIndex;\r
561	}\r
562	CriticalSection_Leave(&p->cs);\r
563	}\r
564	\r
565	RINOK_THREAD(Semaphore_Release1(&p->blocksSemaphore))\r
566	\r
567	if (finished)\r
568	break;\r
569	}\r
570	}\r
571	}\r
572	#else\r
573	{\r
574	WRes wres = Event_Wait(&p->finishedEvent);\r
575	res = MY_SRes_HRESULT_FROM_WRes(wres);\r
576	}\r
577	#endif\r
578	\r
579	if (res == SZ_OK)\r
580	res = p->readRes;\r
581	\r
582	if (res == SZ_OK)\r
583	res = p->mtProgress.res;\r
584	\r
585	#ifndef MTCODER__USE_WRITE_THREAD\r
586	if (res == SZ_OK)\r
587	res = p->writeRes;\r
588	#endif\r
589	\r
590	if (res != SZ_OK)\r
591	MtCoder_Free(p);\r
592	return res;\r
593	}\r
594	\r
595	#endif\r