other-licenses/7zstub/src/C/MtDec.c

   1 /* MtDec.c -- Multi-thread Decoder
   2 2018-03-02 : Igor Pavlov : Public domain */
   3
   4 #include "Precomp.h"
   5
   6 // #define SHOW_DEBUG_INFO
   7
   8 // #include <stdio.h>
   9
  10 #ifdef SHOW_DEBUG_INFO
  11 #include <stdio.h>
  12 #endif
  13
  14 #ifdef SHOW_DEBUG_INFO
  15 #define PRF(x) x
  16 #else
  17 #define PRF(x)
  18 #endif
  19
  20 #define PRF_STR_INT(s, d) PRF(printf("\n" s " %d\n", (unsigned)d))
  21
  22 #include "MtDec.h"
  23
  24 #ifndef _7ZIP_ST
  25
  26 void MtProgress_Init(CMtProgress *p, ICompressProgress *progress)
  27 {
  28   p->progress = progress;
  29   p->res = SZ_OK;
  30   p->totalInSize = 0;
  31   p->totalOutSize = 0;
  32 }
  33
  34
  35 SRes MtProgress_Progress_ST(CMtProgress *p)
  36 {
  37   if (p->res == SZ_OK && p->progress)
  38     if (ICompressProgress_Progress(p->progress, p->totalInSize, p->totalOutSize) != SZ_OK)
  39       p->res = SZ_ERROR_PROGRESS;
  40   return p->res;
  41 }
  42
  43
  44 SRes MtProgress_ProgressAdd(CMtProgress *p, UInt64 inSize, UInt64 outSize)
  45 {
  46   SRes res;
  47   CriticalSection_Enter(&p->cs);
  48
  49   p->totalInSize += inSize;
  50   p->totalOutSize += outSize;
  51   if (p->res == SZ_OK && p->progress)
  52     if (ICompressProgress_Progress(p->progress, p->totalInSize, p->totalOutSize) != SZ_OK)
  53       p->res = SZ_ERROR_PROGRESS;
  54   res = p->res;
  55
  56   CriticalSection_Leave(&p->cs);
  57   return res;
  58 }
  59
  60
  61 SRes MtProgress_GetError(CMtProgress *p)
  62 {
  63   SRes res;
  64   CriticalSection_Enter(&p->cs);
  65   res = p->res;
  66   CriticalSection_Leave(&p->cs);
  67   return res;
  68 }
  69
  70
  71 void MtProgress_SetError(CMtProgress *p, SRes res)
  72 {
  73   CriticalSection_Enter(&p->cs);
  74   if (p->res == SZ_OK)
  75     p->res = res;
  76   CriticalSection_Leave(&p->cs);
  77 }
  78
  79
  80 #define RINOK_THREAD(x) RINOK(x)
  81
  82
  83 static WRes ArEvent_OptCreate_And_Reset(CEvent *p)
  84 {
  85   if (Event_IsCreated(p))
  86     return Event_Reset(p);
  87   return AutoResetEvent_CreateNotSignaled(p);
  88 }
  89
  90
  91
  92 typedef struct
  93 {
  94   void *next;
  95   void *pad[3];
  96 } CMtDecBufLink;
  97
  98 #define MTDEC__LINK_DATA_OFFSET sizeof(CMtDecBufLink)
  99 #define MTDEC__DATA_PTR_FROM_LINK(link) ((Byte *)(link) + MTDEC__LINK_DATA_OFFSET)
 100
 101
 102
 103 static THREAD_FUNC_RET_TYPE THREAD_FUNC_CALL_TYPE ThreadFunc(void *pp);
 104
 105
 106 static WRes MtDecThread_CreateEvents(CMtDecThread *t)
 107 {
 108   WRes wres = ArEvent_OptCreate_And_Reset(&t->canWrite);
 109   if (wres == 0)
 110   {
 111     wres = ArEvent_OptCreate_And_Reset(&t->canRead);
 112     if (wres == 0)
 113       return SZ_OK;
 114   }
 115   return wres;
 116 }
 117
 118
 119 static SRes MtDecThread_CreateAndStart(CMtDecThread *t)
 120 {
 121   WRes wres = MtDecThread_CreateEvents(t);
 122   // wres = 17; // for test
 123   if (wres == 0)
 124   {
 125     if (Thread_WasCreated(&t->thread))
 126       return SZ_OK;
 127     wres = Thread_Create(&t->thread, ThreadFunc, t);
 128     if (wres == 0)
 129       return SZ_OK;
 130   }
 131   return MY_SRes_HRESULT_FROM_WRes(wres);
 132 }
 133
 134
 135 void MtDecThread_FreeInBufs(CMtDecThread *t)
 136 {
 137   if (t->inBuf)
 138   {
 139     void *link = t->inBuf;
 140     t->inBuf = NULL;
 141     do
 142     {
 143       void *next = ((CMtDecBufLink *)link)->next;
 144       ISzAlloc_Free(t->mtDec->alloc, link);
 145       link = next;
 146     }
 147     while (link);
 148   }
 149 }
 150
 151
 152 static void MtDecThread_CloseThread(CMtDecThread *t)
 153 {
 154   if (Thread_WasCreated(&t->thread))
 155   {
 156     Event_Set(&t->canWrite); /* we can disable it. There are no threads waiting canWrite in normal cases */
 157     Event_Set(&t->canRead);
 158     Thread_Wait(&t->thread);
 159     Thread_Close(&t->thread);
 160   }
 161
 162   Event_Close(&t->canRead);
 163   Event_Close(&t->canWrite);
 164 }
 165
 166 static void MtDec_CloseThreads(CMtDec *p)
 167 {
 168   unsigned i;
 169   for (i = 0; i < MTDEC__THREADS_MAX; i++)
 170     MtDecThread_CloseThread(&p->threads[i]);
 171 }
 172
 173 static void MtDecThread_Destruct(CMtDecThread *t)
 174 {
 175   MtDecThread_CloseThread(t);
 176   MtDecThread_FreeInBufs(t);
 177 }
 178
 179
 180
 181 static SRes FullRead(ISeqInStream *stream, Byte *data, size_t *processedSize)
 182 {
 183   size_t size = *processedSize;
 184   *processedSize = 0;
 185   while (size != 0)
 186   {
 187     size_t cur = size;
 188     SRes res = ISeqInStream_Read(stream, data, &cur);
 189     *processedSize += cur;
 190     data += cur;
 191     size -= cur;
 192     RINOK(res);
 193     if (cur == 0)
 194       return SZ_OK;
 195   }
 196   return SZ_OK;
 197 }
 198
 199
 200 static SRes MtDec_GetError_Spec(CMtDec *p, UInt64 interruptIndex, Bool *wasInterrupted)
 201 {
 202   SRes res;
 203   CriticalSection_Enter(&p->mtProgress.cs);
 204   *wasInterrupted = (p->needInterrupt && interruptIndex > p->interruptIndex);
 205   res = p->mtProgress.res;
 206   CriticalSection_Leave(&p->mtProgress.cs);
 207   return res;
 208 }
 209
 210 static SRes MtDec_Progress_GetError_Spec(CMtDec *p, UInt64 inSize, UInt64 outSize, UInt64 interruptIndex, Bool *wasInterrupted)
 211 {
 212   SRes res;
 213   CriticalSection_Enter(&p->mtProgress.cs);
 214
 215   p->mtProgress.totalInSize += inSize;
 216   p->mtProgress.totalOutSize += outSize;
 217   if (p->mtProgress.res == SZ_OK && p->mtProgress.progress)
 218     if (ICompressProgress_Progress(p->mtProgress.progress, p->mtProgress.totalInSize, p->mtProgress.totalOutSize) != SZ_OK)
 219       p->mtProgress.res = SZ_ERROR_PROGRESS;
 220
 221   *wasInterrupted = (p->needInterrupt && interruptIndex > p->interruptIndex);
 222   res = p->mtProgress.res;
 223
 224   CriticalSection_Leave(&p->mtProgress.cs);
 225
 226   return res;
 227 }
 228
 229 static void MtDec_Interrupt(CMtDec *p, UInt64 interruptIndex)
 230 {
 231   CriticalSection_Enter(&p->mtProgress.cs);
 232   if (!p->needInterrupt || interruptIndex < p->interruptIndex)
 233   {
 234     p->interruptIndex = interruptIndex;
 235     p->needInterrupt = True;
 236   }
 237   CriticalSection_Leave(&p->mtProgress.cs);
 238 }
 239
 240 Byte *MtDec_GetCrossBuff(CMtDec *p)
 241 {
 242   Byte *cr = p->crossBlock;
 243   if (!cr)
 244   {
 245     cr = (Byte *)ISzAlloc_Alloc(p->alloc, MTDEC__LINK_DATA_OFFSET + p->inBufSize);
 246     if (!cr)
 247       return NULL;
 248     p->crossBlock = cr;
 249   }
 250   return MTDEC__DATA_PTR_FROM_LINK(cr);
 251 }
 252
 253
 254 /*
 255   ThreadFunc2() returns:
 256   0      - in all normal cases (even for stream error or memory allocation error)
 257   (!= 0) - WRes error return by system threading function
 258 */
 259
 260 // #define MTDEC_ProgessStep (1 << 22)
 261 #define MTDEC_ProgessStep (1 << 0)
 262
 263 static WRes ThreadFunc2(CMtDecThread *t)
 264 {
 265   CMtDec *p = t->mtDec;
 266
 267   PRF_STR_INT("ThreadFunc2", t->index);
 268
 269   // SetThreadAffinityMask(GetCurrentThread(), 1 << t->index);
 270
 271   for (;;)
 272   {
 273     SRes res, codeRes;
 274     Bool wasInterrupted, isAllocError, overflow, finish;
 275     SRes threadingErrorSRes;
 276     Bool needCode, needWrite, needContinue;
 277
 278     size_t inDataSize_Start;
 279     UInt64 inDataSize;
 280     // UInt64 inDataSize_Full;
 281
 282     UInt64 blockIndex;
 283
 284     UInt64 inPrev = 0;
 285     UInt64 outPrev = 0;
 286     UInt64 inCodePos;
 287     UInt64 outCodePos;
 288
 289     Byte *afterEndData = NULL;
 290     size_t afterEndData_Size = 0;
 291
 292     Bool canCreateNewThread = False;
 293     // CMtDecCallbackInfo parse;
 294     CMtDecThread *nextThread;
 295
 296     PRF_STR_INT("Event_Wait(&t->canRead)", t->index);
 297
 298     RINOK_THREAD(Event_Wait(&t->canRead));
 299     if (p->exitThread)
 300       return 0;
 301
 302     PRF_STR_INT("after Event_Wait(&t->canRead)", t->index);
 303
 304     // if (t->index == 3) return 19; // for test
 305
 306     blockIndex = p->blockIndex++;
 307
 308     // PRF(printf("\ncanRead\n"))
 309
 310     res = MtDec_Progress_GetError_Spec(p, 0, 0, blockIndex, &wasInterrupted);
 311
 312     finish = p->readWasFinished;
 313     needCode = False;
 314     needWrite = False;
 315     isAllocError = False;
 316     overflow = False;
 317
 318     inDataSize_Start = 0;
 319     inDataSize = 0;
 320     // inDataSize_Full = 0;
 321
 322     if (res == SZ_OK && !wasInterrupted)
 323     {
 324       // if (p->inStream)
 325       {
 326         CMtDecBufLink *prev = NULL;
 327         CMtDecBufLink *link = (CMtDecBufLink *)t->inBuf;
 328         size_t crossSize = p->crossEnd - p->crossStart;
 329
 330         PRF(printf("\ncrossSize = %d\n", crossSize));
 331
 332         for (;;)
 333         {
 334           if (!link)
 335           {
 336             link = (CMtDecBufLink *)ISzAlloc_Alloc(p->alloc, MTDEC__LINK_DATA_OFFSET + p->inBufSize);
 337             if (!link)
 338             {
 339               finish = True;
 340               // p->allocError_for_Read_BlockIndex = blockIndex;
 341               isAllocError = True;
 342               break;
 343             }
 344             link->next = NULL;
 345             if (prev)
 346             {
 347               // static unsigned g_num = 0;
 348               // printf("\n%6d : %x", ++g_num, (unsigned)(size_t)((Byte *)link - (Byte *)prev));
 349               prev->next = link;
 350             }
 351             else
 352               t->inBuf = (void *)link;
 353           }
 354
 355           {
 356             Byte *data = MTDEC__DATA_PTR_FROM_LINK(link);
 357             Byte *parseData = data;
 358             size_t size;
 359
 360             if (crossSize != 0)
 361             {
 362               inDataSize = crossSize;
 363               // inDataSize_Full = inDataSize;
 364               inDataSize_Start = crossSize;
 365               size = crossSize;
 366               parseData = MTDEC__DATA_PTR_FROM_LINK(p->crossBlock) + p->crossStart;
 367               PRF(printf("\ncross : crossStart = %7d  crossEnd = %7d finish = %1d",
 368                   (int)p->crossStart, (int)p->crossEnd, (int)finish));
 369             }
 370             else
 371             {
 372               size = p->inBufSize;
 373
 374               res = FullRead(p->inStream, data, &size);
 375
 376               // size = 10; // test
 377
 378               inDataSize += size;
 379               // inDataSize_Full = inDataSize;
 380               if (!prev)
 381                 inDataSize_Start = size;
 382
 383               p->readProcessed += size;
 384               finish = (size != p->inBufSize);
 385               if (finish)
 386                 p->readWasFinished = True;
 387
 388               // res = E_INVALIDARG; // test
 389
 390               if (res != SZ_OK)
 391               {
 392                 // PRF(printf("\nRead error = %d\n", res))
 393                 // we want to decode all data before error
 394                 p->readRes = res;
 395                 // p->readError_BlockIndex = blockIndex;
 396                 p->readWasFinished = True;
 397                 finish = True;
 398                 res = SZ_OK;
 399                 // break;
 400               }
 401
 402               if (inDataSize - inPrev >= MTDEC_ProgessStep)
 403               {
 404                 res = MtDec_Progress_GetError_Spec(p, 0, 0, blockIndex, &wasInterrupted);
 405                 if (res != SZ_OK || wasInterrupted)
 406                   break;
 407                 inPrev = inDataSize;
 408               }
 409             }
 410
 411             {
 412               CMtDecCallbackInfo parse;
 413
 414               parse.startCall = (prev == NULL);
 415               parse.src = parseData;
 416               parse.srcSize = size;
 417               parse.srcFinished = finish;
 418               parse.canCreateNewThread = True;
 419
 420               // PRF(printf("\nParse size = %d\n", (unsigned)size))
 421
 422               p->mtCallback->Parse(p->mtCallbackObject, t->index, &parse);
 423
 424               needWrite = True;
 425               canCreateNewThread = parse.canCreateNewThread;
 426
 427               // printf("\n\n%12I64u %12I64u", (UInt64)p->mtProgress.totalInSize, (UInt64)p->mtProgress.totalOutSize);
 428
 429               if (
 430                   // parseRes != SZ_OK ||
 431                   // inDataSize - (size - parse.srcSize) > p->inBlockMax
 432                   // ||
 433                   parse.state == MTDEC_PARSE_OVERFLOW
 434                   // || wasInterrupted
 435                   )
 436               {
 437                 // Overflow or Parse error - switch from MT decoding to ST decoding
 438                 finish = True;
 439                 overflow = True;
 440
 441                 {
 442                   PRF(printf("\n Overflow"));
 443                   // PRF(printf("\nisBlockFinished = %d", (unsigned)parse.blockWasFinished));
 444                   PRF(printf("\n inDataSize = %d", (unsigned)inDataSize));
 445                 }
 446
 447                 if (crossSize != 0)
 448                   memcpy(data, parseData, size);
 449                 p->crossStart = 0;
 450                 p->crossEnd = 0;
 451                 break;
 452               }
 453
 454               if (crossSize != 0)
 455               {
 456                 memcpy(data, parseData, parse.srcSize);
 457                 p->crossStart += parse.srcSize;
 458               }
 459
 460               if (parse.state != MTDEC_PARSE_CONTINUE || finish)
 461               {
 462                 // we don't need to parse in current thread anymore
 463
 464                 if (parse.state == MTDEC_PARSE_END)
 465                   finish = True;
 466
 467                 needCode = True;
 468                 // p->crossFinished = finish;
 469
 470                 if (parse.srcSize == size)
 471                 {
 472                   // full parsed - no cross transfer
 473                   p->crossStart = 0;
 474                   p->crossEnd = 0;
 475                   break;
 476                 }
 477
 478                 if (parse.state == MTDEC_PARSE_END)
 479                 {
 480                   p->crossStart = 0;
 481                   p->crossEnd = 0;
 482
 483                   if (crossSize != 0)
 484                     memcpy(data + parse.srcSize, parseData + parse.srcSize, size - parse.srcSize); // we need all data
 485                   afterEndData_Size = size - parse.srcSize;
 486                   afterEndData = parseData + parse.srcSize;
 487
 488                   // we reduce data size to required bytes (parsed only)
 489                   inDataSize -= (size - parse.srcSize);
 490                   if (!prev)
 491                     inDataSize_Start = parse.srcSize;
 492                   break;
 493                 }
 494
 495                 {
 496                   // partial parsed - need cross transfer
 497                   if (crossSize != 0)
 498                     inDataSize = parse.srcSize; // it's only parsed now
 499                   else
 500                   {
 501                     // partial parsed - is not in initial cross block - we need to copy new data to cross block
 502                     Byte *cr = MtDec_GetCrossBuff(p);
 503                     if (!cr)
 504                     {
 505                       {
 506                         PRF(printf("\ncross alloc error error\n"));
 507                         // res = SZ_ERROR_MEM;
 508                         finish = True;
 509                         // p->allocError_for_Read_BlockIndex = blockIndex;
 510                         isAllocError = True;
 511                         break;
 512                       }
 513                     }
 514
 515                     {
 516                       size_t crSize = size - parse.srcSize;
 517                       inDataSize -= crSize;
 518                       p->crossEnd = crSize;
 519                       p->crossStart = 0;
 520                       memcpy(cr, parseData + parse.srcSize, crSize);
 521                     }
 522                   }
 523
 524                   // inDataSize_Full = inDataSize;
 525                   if (!prev)
 526                     inDataSize_Start = parse.srcSize; // it's partial size (parsed only)
 527
 528                   finish = False;
 529                   break;
 530                 }
 531               }
 532
 533               if (parse.srcSize != size)
 534               {
 535                 res = SZ_ERROR_FAIL;
 536                 PRF(printf("\nfinished error SZ_ERROR_FAIL = %d\n", res));
 537                 break;
 538               }
 539             }
 540           }
 541
 542           prev = link;
 543           link = link->next;
 544
 545           if (crossSize != 0)
 546           {
 547             crossSize = 0;
 548             p->crossStart = 0;
 549             p->crossEnd = 0;
 550           }
 551         }
 552       }
 553
 554       if (res == SZ_OK)
 555         res = MtDec_GetError_Spec(p, blockIndex, &wasInterrupted);
 556     }
 557
 558     codeRes = SZ_OK;
 559
 560     if (res == SZ_OK && needCode && !wasInterrupted)
 561     {
 562       codeRes = p->mtCallback->PreCode(p->mtCallbackObject, t->index);
 563       if (codeRes != SZ_OK)
 564       {
 565         needCode = False;
 566         finish = True;
 567         // SZ_ERROR_MEM is expected error here.
 568         //   if (codeRes == SZ_ERROR_MEM) - we will try single-thread decoding later.
 569         //   if (codeRes != SZ_ERROR_MEM) - we can stop decoding or try single-thread decoding.
 570       }
 571     }
 572
 573     if (res != SZ_OK || wasInterrupted)
 574       finish = True;
 575
 576     nextThread = NULL;
 577     threadingErrorSRes = SZ_OK;
 578
 579     if (!finish)
 580     {
 581       if (p->numStartedThreads < p->numStartedThreads_Limit && canCreateNewThread)
 582       {
 583         SRes res2 = MtDecThread_CreateAndStart(&p->threads[p->numStartedThreads]);
 584         if (res2 == SZ_OK)
 585         {
 586           // if (p->numStartedThreads % 1000 == 0) PRF(printf("\n numStartedThreads=%d\n", p->numStartedThreads));
 587           p->numStartedThreads++;
 588         }
 589         else
 590         {
 591           PRF(printf("\nERROR: numStartedThreads=%d\n", p->numStartedThreads));
 592           if (p->numStartedThreads == 1)
 593           {
 594             // if only one thread is possible, we leave muti-threading code
 595             finish = True;
 596             needCode = False;
 597             threadingErrorSRes = res2;
 598           }
 599           else
 600             p->numStartedThreads_Limit = p->numStartedThreads;
 601         }
 602       }
 603
 604       if (!finish)
 605       {
 606         unsigned nextIndex = t->index + 1;
 607         nextThread = &p->threads[nextIndex >= p->numStartedThreads ? 0 : nextIndex];
 608         RINOK_THREAD(Event_Set(&nextThread->canRead))
 609         // We have started executing for new iteration (with next thread)
 610         // And that next thread now is responsible for possible exit from decoding (threading_code)
 611       }
 612     }
 613
 614     // each call of Event_Set(&nextThread->canRead) must be followed by call of Event_Set(&nextThread->canWrite)
 615     // if ( !finish ) we must call Event_Set(&nextThread->canWrite) in any case
 616     // if (  finish ) we switch to single-thread mode and there are 2 ways at the end of current iteration (current block):
 617     //   - if (needContinue) after Write(&needContinue), we restore decoding with new iteration
 618     //   - otherwise we stop decoding and exit from ThreadFunc2()
 619
 620     // Don't change (finish) variable in the further code
 621
 622
 623     // ---------- CODE ----------
 624
 625     inPrev = 0;
 626     outPrev = 0;
 627     inCodePos = 0;
 628     outCodePos = 0;
 629
 630     if (res == SZ_OK && needCode && codeRes == SZ_OK)
 631     {
 632       Bool isStartBlock = True;
 633       CMtDecBufLink *link = (CMtDecBufLink *)t->inBuf;
 634
 635       for (;;)
 636       {
 637         size_t inSize;
 638         int stop;
 639
 640         if (isStartBlock)
 641           inSize = inDataSize_Start;
 642         else
 643         {
 644           UInt64 rem = inDataSize - inCodePos;
 645           inSize = p->inBufSize;
 646           if (inSize > rem)
 647             inSize = (size_t)rem;
 648         }
 649
 650         inCodePos += inSize;
 651         stop = True;
 652
 653         codeRes = p->mtCallback->Code(p->mtCallbackObject, t->index,
 654             (const Byte *)MTDEC__DATA_PTR_FROM_LINK(link), inSize,
 655             (inCodePos == inDataSize), // srcFinished
 656             &inCodePos, &outCodePos, &stop);
 657
 658         if (codeRes != SZ_OK)
 659         {
 660           PRF(printf("\nCode Interrupt error = %x\n", codeRes));
 661           // we interrupt only later blocks
 662           MtDec_Interrupt(p, blockIndex);
 663           break;
 664         }
 665
 666         if (stop || inCodePos == inDataSize)
 667           break;
 668
 669         {
 670           const UInt64 inDelta = inCodePos - inPrev;
 671           const UInt64 outDelta = outCodePos - outPrev;
 672           if (inDelta >= MTDEC_ProgessStep || outDelta >= MTDEC_ProgessStep)
 673           {
 674             // Sleep(1);
 675             res = MtDec_Progress_GetError_Spec(p, inDelta, outDelta, blockIndex, &wasInterrupted);
 676             if (res != SZ_OK || wasInterrupted)
 677               break;
 678             inPrev = inCodePos;
 679             outPrev = outCodePos;
 680           }
 681         }
 682
 683         link = link->next;
 684         isStartBlock = False;
 685       }
 686     }
 687
 688
 689     // ---------- WRITE ----------
 690
 691     RINOK_THREAD(Event_Wait(&t->canWrite));
 692
 693   {
 694     Bool isErrorMode = False;
 695     Bool canRecode = True;
 696     Bool needWriteToStream = needWrite;
 697
 698     if (p->exitThread) return 0; // it's never executed in normal cases
 699
 700     if (p->wasInterrupted)
 701       wasInterrupted = True;
 702     else
 703     {
 704       if (codeRes != SZ_OK) // || !needCode // check it !!!
 705       {
 706         p->wasInterrupted = True;
 707         p->codeRes = codeRes;
 708         if (codeRes == SZ_ERROR_MEM)
 709           isAllocError = True;
 710       }
 711
 712       if (threadingErrorSRes)
 713       {
 714         p->wasInterrupted = True;
 715         p->threadingErrorSRes = threadingErrorSRes;
 716         needWriteToStream = False;
 717       }
 718       if (isAllocError)
 719       {
 720         p->wasInterrupted = True;
 721         p->isAllocError = True;
 722         needWriteToStream = False;
 723       }
 724       if (overflow)
 725       {
 726         p->wasInterrupted = True;
 727         p->overflow = True;
 728         needWriteToStream = False;
 729       }
 730     }
 731
 732     if (needCode)
 733     {
 734       if (wasInterrupted)
 735       {
 736         inCodePos = 0;
 737         outCodePos = 0;
 738       }
 739       {
 740         const UInt64 inDelta = inCodePos - inPrev;
 741         const UInt64 outDelta = outCodePos - outPrev;
 742         // if (inDelta != 0 || outDelta != 0)
 743         res = MtProgress_ProgressAdd(&p->mtProgress, inDelta, outDelta);
 744       }
 745     }
 746
 747     needContinue = (!finish);
 748
 749     // if (res == SZ_OK && needWrite && !wasInterrupted)
 750     if (needWrite)
 751     {
 752       // p->inProcessed += inCodePos;
 753
 754       res = p->mtCallback->Write(p->mtCallbackObject, t->index,
 755           res == SZ_OK && needWriteToStream && !wasInterrupted, // needWrite
 756           afterEndData, afterEndData_Size,
 757           &needContinue,
 758           &canRecode);
 759
 760       // res= E_INVALIDARG; // for test
 761
 762       PRF(printf("\nAfter Write needContinue = %d\n", (unsigned)needContinue));
 763       PRF(printf("\nprocessed = %d\n", (unsigned)p->inProcessed));
 764
 765       if (res != SZ_OK)
 766       {
 767         PRF(printf("\nWrite error = %d\n", res));
 768         isErrorMode = True;
 769         p->wasInterrupted = True;
 770       }
 771       if (res != SZ_OK
 772           || (!needContinue && !finish))
 773       {
 774         PRF(printf("\nWrite Interrupt error = %x\n", res));
 775         MtDec_Interrupt(p, blockIndex);
 776       }
 777     }
 778
 779     if (canRecode)
 780     if (!needCode
 781         || res != SZ_OK
 782         || p->wasInterrupted
 783         || codeRes != SZ_OK
 784         || wasInterrupted
 785         || p->numFilledThreads != 0
 786         || isErrorMode)
 787     {
 788       if (p->numFilledThreads == 0)
 789         p->filledThreadStart = t->index;
 790       if (inDataSize != 0 || !finish)
 791       {
 792         t->inDataSize_Start = inDataSize_Start;
 793         t->inDataSize = inDataSize;
 794         p->numFilledThreads++;
 795       }
 796       PRF(printf("\np->numFilledThreads = %d\n", p->numFilledThreads));
 797       PRF(printf("p->filledThreadStart = %d\n", p->filledThreadStart));
 798     }
 799
 800     if (!finish)
 801     {
 802       RINOK_THREAD(Event_Set(&nextThread->canWrite));
 803     }
 804     else
 805     {
 806       if (needContinue)
 807       {
 808         // we restore decoding with new iteration
 809         RINOK_THREAD(Event_Set(&p->threads[0].canWrite));
 810       }
 811       else
 812       {
 813         // we exit from decoding
 814         if (t->index == 0)
 815           return SZ_OK;
 816         p->exitThread = True;
 817       }
 818       RINOK_THREAD(Event_Set(&p->threads[0].canRead));
 819     }
 820   }
 821   }
 822 }
 823
 824 #ifdef _WIN32
 825 #define USE_ALLOCA
 826 #endif
 827
 828 #ifdef USE_ALLOCA
 829 #ifdef _WIN32
 830 #include <malloc.h>
 831 #else
 832 #include <stdlib.h>
 833 #endif
 834 #endif
 835
 836
 837 static THREAD_FUNC_RET_TYPE THREAD_FUNC_CALL_TYPE ThreadFunc1(void *pp)
 838 {
 839   WRes res;
 840
 841   CMtDecThread *t = (CMtDecThread *)pp;
 842   CMtDec *p;
 843
 844   // fprintf(stdout, "\n%d = %p\n", t->index, &t);
 845
 846   res = ThreadFunc2(t);
 847   p = t->mtDec;
 848   if (res == 0)
 849     return p->exitThreadWRes;
 850   {
 851     // it's unexpected situation for some threading function error
 852     if (p->exitThreadWRes == 0)
 853       p->exitThreadWRes = res;
 854     PRF(printf("\nthread exit error = %d\n", res));
 855     p->exitThread = True;
 856     Event_Set(&p->threads[0].canRead);
 857     Event_Set(&p->threads[0].canWrite);
 858     MtProgress_SetError(&p->mtProgress, MY_SRes_HRESULT_FROM_WRes(res));
 859   }
 860   return res;
 861 }
 862
 863 static MY_NO_INLINE THREAD_FUNC_RET_TYPE THREAD_FUNC_CALL_TYPE ThreadFunc(void *pp)
 864 {
 865   CMtDecThread *t = (CMtDecThread *)pp;
 866
 867   // fprintf(stderr, "\n%d = %p - before", t->index, &t);
 868   #ifdef USE_ALLOCA
 869   t->allocaPtr = alloca(t->index * 128);
 870   #endif
 871   return ThreadFunc1(pp);
 872 }
 873
 874
 875 int MtDec_PrepareRead(CMtDec *p)
 876 {
 877   if (p->crossBlock && p->crossStart == p->crossEnd)
 878   {
 879     ISzAlloc_Free(p->alloc, p->crossBlock);
 880     p->crossBlock = NULL;
 881   }
 882
 883   {
 884     unsigned i;
 885     for (i = 0; i < MTDEC__THREADS_MAX; i++)
 886       if (i > p->numStartedThreads
 887           || p->numFilledThreads <=
 888             (i >= p->filledThreadStart ?
 889               i - p->filledThreadStart :
 890               i + p->numStartedThreads - p->filledThreadStart))
 891         MtDecThread_FreeInBufs(&p->threads[i]);
 892   }
 893
 894   return (p->numFilledThreads != 0) || (p->crossStart != p->crossEnd);
 895 }
 896
 897
 898 const Byte *MtDec_Read(CMtDec *p, size_t *inLim)
 899 {
 900   while (p->numFilledThreads != 0)
 901   {
 902     CMtDecThread *t = &p->threads[p->filledThreadStart];
 903
 904     if (*inLim != 0)
 905     {
 906       {
 907         void *link = t->inBuf;
 908         void *next = ((CMtDecBufLink *)link)->next;
 909         ISzAlloc_Free(p->alloc, link);
 910         t->inBuf = next;
 911       }
 912
 913       if (t->inDataSize == 0)
 914       {
 915         MtDecThread_FreeInBufs(t);
 916         if (--p->numFilledThreads == 0)
 917           break;
 918         if (++p->filledThreadStart == p->numStartedThreads)
 919           p->filledThreadStart = 0;
 920         t = &p->threads[p->filledThreadStart];
 921       }
 922     }
 923
 924     {
 925       size_t lim = t->inDataSize_Start;
 926       if (lim != 0)
 927         t->inDataSize_Start = 0;
 928       else
 929       {
 930         UInt64 rem = t->inDataSize;
 931         lim = p->inBufSize;
 932         if (lim > rem)
 933           lim = (size_t)rem;
 934       }
 935       t->inDataSize -= lim;
 936       *inLim = lim;
 937       return (const Byte *)MTDEC__DATA_PTR_FROM_LINK(t->inBuf);
 938     }
 939   }
 940
 941   {
 942     size_t crossSize = p->crossEnd - p->crossStart;
 943     if (crossSize != 0)
 944     {
 945       const Byte *data = MTDEC__DATA_PTR_FROM_LINK(p->crossBlock) + p->crossStart;
 946       *inLim = crossSize;
 947       p->crossStart = 0;
 948       p->crossEnd = 0;
 949       return data;
 950     }
 951     *inLim = 0;
 952     if (p->crossBlock)
 953     {
 954       ISzAlloc_Free(p->alloc, p->crossBlock);
 955       p->crossBlock = NULL;
 956     }
 957     return NULL;
 958   }
 959 }
 960
 961
 962 void MtDec_Construct(CMtDec *p)
 963 {
 964   unsigned i;
 965
 966   p->inBufSize = (size_t)1 << 18;
 967
 968   p->numThreadsMax = 0;
 969
 970   p->inStream = NULL;
 971
 972   // p->inData = NULL;
 973   // p->inDataSize = 0;
 974
 975   p->crossBlock = NULL;
 976   p->crossStart = 0;
 977   p->crossEnd = 0;
 978
 979   p->numFilledThreads = 0;
 980
 981   p->progress = NULL;
 982   p->alloc = NULL;
 983
 984   p->mtCallback = NULL;
 985   p->mtCallbackObject = NULL;
 986
 987   p->allocatedBufsSize = 0;
 988
 989   for (i = 0; i < MTDEC__THREADS_MAX; i++)
 990   {
 991     CMtDecThread *t = &p->threads[i];
 992     t->mtDec = p;
 993     t->index = i;
 994     t->inBuf = NULL;
 995     Event_Construct(&t->canRead);
 996     Event_Construct(&t->canWrite);
 997     Thread_Construct(&t->thread);
 998   }
 999
1000   // Event_Construct(&p->finishedEvent);
1001
1002   CriticalSection_Init(&p->mtProgress.cs);
1003 }
1004
1005
1006 static void MtDec_Free(CMtDec *p)
1007 {
1008   unsigned i;
1009
1010   p->exitThread = True;
1011
1012   for (i = 0; i < MTDEC__THREADS_MAX; i++)
1013     MtDecThread_Destruct(&p->threads[i]);
1014
1015   // Event_Close(&p->finishedEvent);
1016
1017   if (p->crossBlock)
1018   {
1019     ISzAlloc_Free(p->alloc, p->crossBlock);
1020     p->crossBlock = NULL;
1021   }
1022 }
1023
1024
1025 void MtDec_Destruct(CMtDec *p)
1026 {
1027   MtDec_Free(p);
1028
1029   CriticalSection_Delete(&p->mtProgress.cs);
1030 }
1031
1032
1033 SRes MtDec_Code(CMtDec *p)
1034 {
1035   unsigned i;
1036
1037   p->inProcessed = 0;
1038
1039   p->blockIndex = 1; // it must be larger than not_defined index (0)
1040   p->isAllocError = False;
1041   p->overflow = False;
1042   p->threadingErrorSRes = SZ_OK;
1043
1044   p->needContinue = True;
1045
1046   p->readWasFinished = False;
1047   p->needInterrupt = False;
1048   p->interruptIndex = (UInt64)(Int64)-1;
1049
1050   p->readProcessed = 0;
1051   p->readRes = SZ_OK;
1052   p->codeRes = SZ_OK;
1053   p->wasInterrupted = False;
1054
1055   p->crossStart = 0;
1056   p->crossEnd = 0;
1057
1058   p->filledThreadStart = 0;
1059   p->numFilledThreads = 0;
1060
1061   {
1062     unsigned numThreads = p->numThreadsMax;
1063     if (numThreads > MTDEC__THREADS_MAX)
1064       numThreads = MTDEC__THREADS_MAX;
1065     p->numStartedThreads_Limit = numThreads;
1066     p->numStartedThreads = 0;
1067   }
1068
1069   if (p->inBufSize != p->allocatedBufsSize)
1070   {
1071     for (i = 0; i < MTDEC__THREADS_MAX; i++)
1072     {
1073       CMtDecThread *t = &p->threads[i];
1074       if (t->inBuf)
1075         MtDecThread_FreeInBufs(t);
1076     }
1077     if (p->crossBlock)
1078     {
1079       ISzAlloc_Free(p->alloc, p->crossBlock);
1080       p->crossBlock = NULL;
1081     }
1082
1083     p->allocatedBufsSize = p->inBufSize;
1084   }
1085
1086   MtProgress_Init(&p->mtProgress, p->progress);
1087
1088   // RINOK_THREAD(ArEvent_OptCreate_And_Reset(&p->finishedEvent));
1089   p->exitThread = False;
1090   p->exitThreadWRes = 0;
1091
1092   {
1093     WRes wres;
1094     WRes sres;
1095     CMtDecThread *nextThread = &p->threads[p->numStartedThreads++];
1096     // wres = MtDecThread_CreateAndStart(nextThread);
1097     wres = MtDecThread_CreateEvents(nextThread);
1098     if (wres == 0) { wres = Event_Set(&nextThread->canWrite);
1099     if (wres == 0) { wres = Event_Set(&nextThread->canRead);
1100     if (wres == 0) { wres = ThreadFunc(nextThread);
1101     if (wres != 0)
1102     {
1103       p->needContinue = False;
1104       MtDec_CloseThreads(p);
1105     }}}}
1106
1107     // wres = 17; // for test
1108     // wres = Event_Wait(&p->finishedEvent);
1109
1110     sres = MY_SRes_HRESULT_FROM_WRes(wres);
1111
1112     if (sres != 0)
1113       p->threadingErrorSRes = sres;
1114
1115     if (
1116         // wres == 0
1117         // wres != 0
1118         // || p->mtc.codeRes == SZ_ERROR_MEM
1119         p->isAllocError
1120         || p->threadingErrorSRes != SZ_OK
1121         || p->overflow)
1122     {
1123       // p->needContinue = True;
1124     }
1125     else
1126       p->needContinue = False;
1127
1128     if (p->needContinue)
1129       return SZ_OK;
1130
1131     // if (sres != SZ_OK)
1132       return sres;
1133     // return E_FAIL;
1134   }
1135 }
1136
1137 #endif