1 /* MtDec.h -- Multi-thread Decoder
\r
2 2020-03-05 : Igor Pavlov : Public domain */
\r
10 #include "Threads.h"
\r
18 #define MTDEC__THREADS_MAX 32
\r
20 #define MTDEC__THREADS_MAX 1
\r
26 ICompressProgress *progress;
\r
29 UInt64 totalOutSize;
\r
30 CCriticalSection cs;
\r
33 void MtProgress_Init(CMtProgress *p, ICompressProgress *progress);
\r
34 SRes MtProgress_Progress_ST(CMtProgress *p);
\r
35 SRes MtProgress_ProgressAdd(CMtProgress *p, UInt64 inSize, UInt64 outSize);
\r
36 SRes MtProgress_GetError(CMtProgress *p);
\r
37 void MtProgress_SetError(CMtProgress *p, SRes res);
\r
43 struct _CMtDec *mtDec;
\r
47 size_t inDataSize_Start; // size of input data in start block
\r
48 UInt64 inDataSize; // total size of input data in all blocks
\r
51 CAutoResetEvent canRead;
\r
52 CAutoResetEvent canWrite;
\r
56 void MtDecThread_FreeInBufs(CMtDecThread *t);
\r
61 MTDEC_PARSE_CONTINUE, // continue this block with more input data
\r
62 MTDEC_PARSE_OVERFLOW, // MT buffers overflow, need switch to single-thread
\r
63 MTDEC_PARSE_NEW, // new block
\r
64 MTDEC_PARSE_END // end of block threading. But we still can return to threading after Write(&needContinue)
\r
73 // in : (srcSize == 0) is allowed
\r
74 // out : it's allowed to return less that actually was used ?
\r
78 EMtDecParseState state;
\r
79 BoolInt canCreateNewThread;
\r
80 UInt64 outPos; // check it (size_t)
\r
81 } CMtDecCallbackInfo;
\r
86 void (*Parse)(void *p, unsigned coderIndex, CMtDecCallbackInfo *ci);
\r
88 // PreCode() and Code():
\r
89 // (SRes_return_result != SZ_OK) means stop decoding, no need another blocks
\r
90 SRes (*PreCode)(void *p, unsigned coderIndex);
\r
91 SRes (*Code)(void *p, unsigned coderIndex,
\r
92 const Byte *src, size_t srcSize, int srcFinished,
\r
93 UInt64 *inCodePos, UInt64 *outCodePos, int *stop);
\r
94 // stop - means stop another Code calls
\r
97 /* Write() must be called, if Parse() was called
\r
100 && (was not interrupted by progress)
\r
101 && (was not interrupted in previous block)
\r
105 if (*needContinue), decoder still need to continue decoding with new iteration,
\r
106 even after MTDEC_PARSE_END
\r
107 if (*canRecode), we didn't flush current block data, so we still can decode current block later.
\r
109 SRes (*Write)(void *p, unsigned coderIndex,
\r
110 BoolInt needWriteToStream,
\r
111 const Byte *src, size_t srcSize, BoolInt isCross,
\r
112 // int srcFinished,
\r
113 BoolInt *needContinue,
\r
114 BoolInt *canRecode);
\r
120 typedef struct _CMtDec
\r
122 /* input variables */
\r
124 size_t inBufSize; /* size of input block */
\r
125 unsigned numThreadsMax;
\r
126 // size_t inBlockMax;
\r
127 unsigned numThreadsMax_2;
\r
129 ISeqInStream *inStream;
\r
130 // const Byte *inData;
\r
131 // size_t inDataSize;
\r
133 ICompressProgress *progress;
\r
136 IMtDecCallback2 *mtCallback;
\r
137 void *mtCallbackObject;
\r
140 /* internal variables */
\r
142 size_t allocatedBufsSize;
\r
144 BoolInt exitThread;
\r
145 WRes exitThreadWRes;
\r
148 BoolInt isAllocError;
\r
150 SRes threadingErrorSRes;
\r
152 BoolInt needContinue;
\r
154 // CAutoResetEvent finishedEvent;
\r
159 BoolInt wasInterrupted;
\r
161 unsigned numStartedThreads_Limit;
\r
162 unsigned numStartedThreads;
\r
167 UInt64 readProcessed;
\r
168 BoolInt readWasFinished;
\r
169 UInt64 inProcessed;
\r
171 unsigned filledThreadStart;
\r
172 unsigned numFilledThreads;
\r
175 BoolInt needInterrupt;
\r
176 UInt64 interruptIndex;
\r
177 CMtProgress mtProgress;
\r
178 CMtDecThread threads[MTDEC__THREADS_MAX];
\r
183 void MtDec_Construct(CMtDec *p);
\r
184 void MtDec_Destruct(CMtDec *p);
\r
187 MtDec_Code() returns:
\r
188 SZ_OK - in most cases
\r
189 MY_SRes_HRESULT_FROM_WRes(WRes_error) - in case of unexpected error in threading function
\r
192 SRes MtDec_Code(CMtDec *p);
\r
193 Byte *MtDec_GetCrossBuff(CMtDec *p);
\r
195 int MtDec_PrepareRead(CMtDec *p);
\r
196 const Byte *MtDec_Read(CMtDec *p, size_t *inLim);
\r