[pcsx_rearmed.git] / deps / lzma-16.04 / C / MtCoder.c

/* MtCoder.c -- Multi-thread Coder\r
2015-10-13 : Igor Pavlov : Public domain */\r
\r
#include "Precomp.h"\r
\r
#include "MtCoder.h"\r
\r
void LoopThread_Construct(CLoopThread *p)\r
{\r
  Thread_Construct(&p->thread);\r
  Event_Construct(&p->startEvent);\r
  Event_Construct(&p->finishedEvent);\r
}\r
\r
void LoopThread_Close(CLoopThread *p)\r
{\r
  Thread_Close(&p->thread);\r
  Event_Close(&p->startEvent);\r
  Event_Close(&p->finishedEvent);\r
}\r
\r
static THREAD_FUNC_RET_TYPE THREAD_FUNC_CALL_TYPE LoopThreadFunc(void *pp)\r
{\r
  CLoopThread *p = (CLoopThread *)pp;\r
  for (;;)\r
  {\r
    if (Event_Wait(&p->startEvent) != 0)\r
      return SZ_ERROR_THREAD;\r
    if (p->stop)\r
      return 0;\r
    p->res = p->func(p->param);\r
    if (Event_Set(&p->finishedEvent) != 0)\r
      return SZ_ERROR_THREAD;\r
  }\r
}\r
\r
WRes LoopThread_Create(CLoopThread *p)\r
{\r
  p->stop = 0;\r
  RINOK(AutoResetEvent_CreateNotSignaled(&p->startEvent));\r
  RINOK(AutoResetEvent_CreateNotSignaled(&p->finishedEvent));\r
  return Thread_Create(&p->thread, LoopThreadFunc, p);\r
}\r
\r
WRes LoopThread_StopAndWait(CLoopThread *p)\r
{\r
  p->stop = 1;\r
  if (Event_Set(&p->startEvent) != 0)\r
    return SZ_ERROR_THREAD;\r
  return Thread_Wait(&p->thread);\r
}\r
\r
WRes LoopThread_StartSubThread(CLoopThread *p) { return Event_Set(&p->startEvent); }\r
WRes LoopThread_WaitSubThread(CLoopThread *p) { return Event_Wait(&p->finishedEvent); }\r
\r
static SRes Progress(ICompressProgress *p, UInt64 inSize, UInt64 outSize)\r
{\r
  return (p && p->Progress(p, inSize, outSize) != SZ_OK) ? SZ_ERROR_PROGRESS : SZ_OK;\r
}\r
\r
static void MtProgress_Init(CMtProgress *p, ICompressProgress *progress)\r
{\r
  unsigned i;\r
  for (i = 0; i < NUM_MT_CODER_THREADS_MAX; i++)\r
    p->inSizes[i] = p->outSizes[i] = 0;\r
  p->totalInSize = p->totalOutSize = 0;\r
  p->progress = progress;\r
  p->res = SZ_OK;\r
}\r
\r
static void MtProgress_Reinit(CMtProgress *p, unsigned index)\r
{\r
  p->inSizes[index] = 0;\r
  p->outSizes[index] = 0;\r
}\r
\r
#define UPDATE_PROGRESS(size, prev, total) \\r
  if (size != (UInt64)(Int64)-1) { total += size - prev; prev = size; }\r
\r
SRes MtProgress_Set(CMtProgress *p, unsigned index, UInt64 inSize, UInt64 outSize)\r
{\r
  SRes res;\r
  CriticalSection_Enter(&p->cs);\r
  UPDATE_PROGRESS(inSize, p->inSizes[index], p->totalInSize)\r
  UPDATE_PROGRESS(outSize, p->outSizes[index], p->totalOutSize)\r
  if (p->res == SZ_OK)\r
    p->res = Progress(p->progress, p->totalInSize, p->totalOutSize);\r
  res = p->res;\r
  CriticalSection_Leave(&p->cs);\r
  return res;\r
}\r
\r
static void MtProgress_SetError(CMtProgress *p, SRes res)\r
{\r
  CriticalSection_Enter(&p->cs);\r
  if (p->res == SZ_OK)\r
    p->res = res;\r
  CriticalSection_Leave(&p->cs);\r
}\r
\r
static void MtCoder_SetError(CMtCoder* p, SRes res)\r
{\r
  CriticalSection_Enter(&p->cs);\r
  if (p->res == SZ_OK)\r
    p->res = res;\r
  CriticalSection_Leave(&p->cs);\r
}\r
\r
/* ---------- MtThread ---------- */\r
\r
void CMtThread_Construct(CMtThread *p, CMtCoder *mtCoder)\r
{\r
  p->mtCoder = mtCoder;\r
  p->outBuf = 0;\r
  p->inBuf = 0;\r
  Event_Construct(&p->canRead);\r
  Event_Construct(&p->canWrite);\r
  LoopThread_Construct(&p->thread);\r
}\r
\r
#define RINOK_THREAD(x) { if ((x) != 0) return SZ_ERROR_THREAD; }\r
\r
static void CMtThread_CloseEvents(CMtThread *p)\r
{\r
  Event_Close(&p->canRead);\r
  Event_Close(&p->canWrite);\r
}\r
\r
static void CMtThread_Destruct(CMtThread *p)\r
{\r
  CMtThread_CloseEvents(p);\r
\r
  if (Thread_WasCreated(&p->thread.thread))\r
  {\r
    LoopThread_StopAndWait(&p->thread);\r
    LoopThread_Close(&p->thread);\r
  }\r
\r
  if (p->mtCoder->alloc)\r
    IAlloc_Free(p->mtCoder->alloc, p->outBuf);\r
  p->outBuf = 0;\r
\r
  if (p->mtCoder->alloc)\r
    IAlloc_Free(p->mtCoder->alloc, p->inBuf);\r
  p->inBuf = 0;\r
}\r
\r
#define MY_BUF_ALLOC(buf, size, newSize) \\r
  if (buf == 0 || size != newSize) \\r
  { IAlloc_Free(p->mtCoder->alloc, buf); \\r
    size = newSize; buf = (Byte *)IAlloc_Alloc(p->mtCoder->alloc, size); \\r
    if (buf == 0) return SZ_ERROR_MEM; }\r
\r
static SRes CMtThread_Prepare(CMtThread *p)\r
{\r
  MY_BUF_ALLOC(p->inBuf, p->inBufSize, p->mtCoder->blockSize)\r
  MY_BUF_ALLOC(p->outBuf, p->outBufSize, p->mtCoder->destBlockSize)\r
\r
  p->stopReading = False;\r
  p->stopWriting = False;\r
  RINOK_THREAD(AutoResetEvent_CreateNotSignaled(&p->canRead));\r
  RINOK_THREAD(AutoResetEvent_CreateNotSignaled(&p->canWrite));\r
\r
  return SZ_OK;\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 curSize = size;\r
    SRes res = stream->Read(stream, data, &curSize);\r
    *processedSize += curSize;\r
    data += curSize;\r
    size -= curSize;\r
    RINOK(res);\r
    if (curSize == 0)\r
      return SZ_OK;\r
  }\r
  return SZ_OK;\r
}\r
\r
#define GET_NEXT_THREAD(p) &p->mtCoder->threads[p->index == p->mtCoder->numThreads  - 1 ? 0 : p->index + 1]\r
\r
static SRes MtThread_Process(CMtThread *p, Bool *stop)\r
{\r
  CMtThread *next;\r
  *stop = True;\r
  if (Event_Wait(&p->canRead) != 0)\r
    return SZ_ERROR_THREAD;\r
  \r
  next = GET_NEXT_THREAD(p);\r
  \r
  if (p->stopReading)\r
  {\r
    next->stopReading = True;\r
    return Event_Set(&next->canRead) == 0 ? SZ_OK : SZ_ERROR_THREAD;\r
  }\r
\r
  {\r
    size_t size = p->mtCoder->blockSize;\r
    size_t destSize = p->outBufSize;\r
\r
    RINOK(FullRead(p->mtCoder->inStream, p->inBuf, &size));\r
    next->stopReading = *stop = (size != p->mtCoder->blockSize);\r
    if (Event_Set(&next->canRead) != 0)\r
      return SZ_ERROR_THREAD;\r
\r
    RINOK(p->mtCoder->mtCallback->Code(p->mtCoder->mtCallback, p->index,\r
        p->outBuf, &destSize, p->inBuf, size, *stop));\r
\r
    MtProgress_Reinit(&p->mtCoder->mtProgress, p->index);\r
\r
    if (Event_Wait(&p->canWrite) != 0)\r
      return SZ_ERROR_THREAD;\r
    if (p->stopWriting)\r
      return SZ_ERROR_FAIL;\r
    if (p->mtCoder->outStream->Write(p->mtCoder->outStream, p->outBuf, destSize) != destSize)\r
      return SZ_ERROR_WRITE;\r
    return Event_Set(&next->canWrite) == 0 ? SZ_OK : SZ_ERROR_THREAD;\r
  }\r
}\r
\r
static THREAD_FUNC_RET_TYPE THREAD_FUNC_CALL_TYPE ThreadFunc(void *pp)\r
{\r
  CMtThread *p = (CMtThread *)pp;\r
  for (;;)\r
  {\r
    Bool stop;\r
    CMtThread *next = GET_NEXT_THREAD(p);\r
    SRes res = MtThread_Process(p, &stop);\r
    if (res != SZ_OK)\r
    {\r
      MtCoder_SetError(p->mtCoder, res);\r
      MtProgress_SetError(&p->mtCoder->mtProgress, res);\r
      next->stopReading = True;\r
      next->stopWriting = True;\r
      Event_Set(&next->canRead);\r
      Event_Set(&next->canWrite);\r
      return res;\r
    }\r
    if (stop)\r
      return 0;\r
  }\r
}\r
\r
void MtCoder_Construct(CMtCoder* p)\r
{\r
  unsigned i;\r
  p->alloc = 0;\r
  for (i = 0; i < NUM_MT_CODER_THREADS_MAX; i++)\r
  {\r
    CMtThread *t = &p->threads[i];\r
    t->index = i;\r
    CMtThread_Construct(t, p);\r
  }\r
  CriticalSection_Init(&p->cs);\r
  CriticalSection_Init(&p->mtProgress.cs);\r
}\r
\r
void MtCoder_Destruct(CMtCoder* p)\r
{\r
  unsigned i;\r
  for (i = 0; i < NUM_MT_CODER_THREADS_MAX; i++)\r
    CMtThread_Destruct(&p->threads[i]);\r
  CriticalSection_Delete(&p->cs);\r
  CriticalSection_Delete(&p->mtProgress.cs);\r
}\r
\r
SRes MtCoder_Code(CMtCoder *p)\r
{\r
  unsigned i, numThreads = p->numThreads;\r
  SRes res = SZ_OK;\r
  p->res = SZ_OK;\r
\r
  MtProgress_Init(&p->mtProgress, p->progress);\r
\r
  for (i = 0; i < numThreads; i++)\r
  {\r
    RINOK(CMtThread_Prepare(&p->threads[i]));\r
  }\r
\r
  for (i = 0; i < numThreads; i++)\r
  {\r
    CMtThread *t = &p->threads[i];\r
    CLoopThread *lt = &t->thread;\r
\r
    if (!Thread_WasCreated(&lt->thread))\r
    {\r
      lt->func = ThreadFunc;\r
      lt->param = t;\r
\r
      if (LoopThread_Create(lt) != SZ_OK)\r
      {\r
        res = SZ_ERROR_THREAD;\r
        break;\r
      }\r
    }\r
  }\r
\r
  if (res == SZ_OK)\r
  {\r
    unsigned j;\r
    for (i = 0; i < numThreads; i++)\r
    {\r
      CMtThread *t = &p->threads[i];\r
      if (LoopThread_StartSubThread(&t->thread) != SZ_OK)\r
      {\r
        res = SZ_ERROR_THREAD;\r
        p->threads[0].stopReading = True;\r
        break;\r
      }\r
    }\r
\r
    Event_Set(&p->threads[0].canWrite);\r
    Event_Set(&p->threads[0].canRead);\r
\r
    for (j = 0; j < i; j++)\r
      LoopThread_WaitSubThread(&p->threads[j].thread);\r
  }\r
\r
  for (i = 0; i < numThreads; i++)\r
    CMtThread_CloseEvents(&p->threads[i]);\r
  return (res == SZ_OK) ? p->res : res;\r
}\r
Commit	Line	Data
ce188d4d	1	/* MtCoder.c -- Multi-thread Coder\r
	2	2015-10-13 : Igor Pavlov : Public domain */\r
	3	\r
	4	#include "Precomp.h"\r
	5	\r
	6	#include "MtCoder.h"\r
	7	\r
	8	void LoopThread_Construct(CLoopThread *p)\r
	9	{\r
	10	Thread_Construct(&p->thread);\r
	11	Event_Construct(&p->startEvent);\r
	12	Event_Construct(&p->finishedEvent);\r
	13	}\r
	14	\r
	15	void LoopThread_Close(CLoopThread *p)\r
	16	{\r
	17	Thread_Close(&p->thread);\r
	18	Event_Close(&p->startEvent);\r
	19	Event_Close(&p->finishedEvent);\r
	20	}\r
	21	\r
	22	static THREAD_FUNC_RET_TYPE THREAD_FUNC_CALL_TYPE LoopThreadFunc(void *pp)\r
	23	{\r
	24	CLoopThread p = (CLoopThread )pp;\r
	25	for (;;)\r
	26	{\r
	27	if (Event_Wait(&p->startEvent) != 0)\r
	28	return SZ_ERROR_THREAD;\r
	29	if (p->stop)\r
	30	return 0;\r
	31	p->res = p->func(p->param);\r
	32	if (Event_Set(&p->finishedEvent) != 0)\r
	33	return SZ_ERROR_THREAD;\r
	34	}\r
	35	}\r
	36	\r
	37	WRes LoopThread_Create(CLoopThread *p)\r
	38	{\r
	39	p->stop = 0;\r
	40	RINOK(AutoResetEvent_CreateNotSignaled(&p->startEvent));\r
	41	RINOK(AutoResetEvent_CreateNotSignaled(&p->finishedEvent));\r
	42	return Thread_Create(&p->thread, LoopThreadFunc, p);\r
	43	}\r
	44	\r
	45	WRes LoopThread_StopAndWait(CLoopThread *p)\r
	46	{\r
	47	p->stop = 1;\r
	48	if (Event_Set(&p->startEvent) != 0)\r
	49	return SZ_ERROR_THREAD;\r
	50	return Thread_Wait(&p->thread);\r
	51	}\r
	52	\r
	53	WRes LoopThread_StartSubThread(CLoopThread *p) { return Event_Set(&p->startEvent); }\r
	54	WRes LoopThread_WaitSubThread(CLoopThread *p) { return Event_Wait(&p->finishedEvent); }\r
	55	\r
	56	static SRes Progress(ICompressProgress *p, UInt64 inSize, UInt64 outSize)\r
	57	{\r
	58	return (p && p->Progress(p, inSize, outSize) != SZ_OK) ? SZ_ERROR_PROGRESS : SZ_OK;\r
	59	}\r
	60	\r
	61	static void MtProgress_Init(CMtProgress p, ICompressProgress progress)\r
	62	{\r
	63	unsigned i;\r
	64	for (i = 0; i < NUM_MT_CODER_THREADS_MAX; i++)\r
65	p->inSizes[i] = p->outSizes[i] = 0;\r
66	p->totalInSize = p->totalOutSize = 0;\r
67	p->progress = progress;\r
68	p->res = SZ_OK;\r
69	}\r
70	\r
71	static void MtProgress_Reinit(CMtProgress *p, unsigned index)\r
72	{\r
73	p->inSizes[index] = 0;\r
74	p->outSizes[index] = 0;\r
75	}\r
76	\r
77	#define UPDATE_PROGRESS(size, prev, total) \\r
78	if (size != (UInt64)(Int64)-1) { total += size - prev; prev = size; }\r
79	\r
80	SRes MtProgress_Set(CMtProgress *p, unsigned index, UInt64 inSize, UInt64 outSize)\r
81	{\r
82	SRes res;\r
83	CriticalSection_Enter(&p->cs);\r
84	UPDATE_PROGRESS(inSize, p->inSizes[index], p->totalInSize)\r
85	UPDATE_PROGRESS(outSize, p->outSizes[index], p->totalOutSize)\r
86	if (p->res == SZ_OK)\r
87	p->res = Progress(p->progress, p->totalInSize, p->totalOutSize);\r
88	res = p->res;\r
89	CriticalSection_Leave(&p->cs);\r
90	return res;\r
91	}\r
92	\r
93	static void MtProgress_SetError(CMtProgress *p, SRes res)\r
94	{\r
95	CriticalSection_Enter(&p->cs);\r
96	if (p->res == SZ_OK)\r
97	p->res = res;\r
98	CriticalSection_Leave(&p->cs);\r
99	}\r
100	\r
101	static void MtCoder_SetError(CMtCoder* p, SRes res)\r
102	{\r
103	CriticalSection_Enter(&p->cs);\r
104	if (p->res == SZ_OK)\r
105	p->res = res;\r
106	CriticalSection_Leave(&p->cs);\r
107	}\r
108	\r
109	/* ---------- MtThread ---------- */\r
110	\r
111	void CMtThread_Construct(CMtThread p, CMtCoder mtCoder)\r
112	{\r
113	p->mtCoder = mtCoder;\r
114	p->outBuf = 0;\r
115	p->inBuf = 0;\r
116	Event_Construct(&p->canRead);\r
117	Event_Construct(&p->canWrite);\r
118	LoopThread_Construct(&p->thread);\r
119	}\r
120	\r
121	#define RINOK_THREAD(x) { if ((x) != 0) return SZ_ERROR_THREAD; }\r
122	\r
123	static void CMtThread_CloseEvents(CMtThread *p)\r
124	{\r
125	Event_Close(&p->canRead);\r
126	Event_Close(&p->canWrite);\r
127	}\r
128	\r
129	static void CMtThread_Destruct(CMtThread *p)\r
130	{\r
131	CMtThread_CloseEvents(p);\r
132	\r
133	if (Thread_WasCreated(&p->thread.thread))\r
134	{\r
135	LoopThread_StopAndWait(&p->thread);\r
136	LoopThread_Close(&p->thread);\r
137	}\r
138	\r
139	if (p->mtCoder->alloc)\r
140	IAlloc_Free(p->mtCoder->alloc, p->outBuf);\r
141	p->outBuf = 0;\r
142	\r
143	if (p->mtCoder->alloc)\r
144	IAlloc_Free(p->mtCoder->alloc, p->inBuf);\r
145	p->inBuf = 0;\r
146	}\r
147	\r
148	#define MY_BUF_ALLOC(buf, size, newSize) \\r
149	if (buf == 0 \|\| size != newSize) \\r
150	{ IAlloc_Free(p->mtCoder->alloc, buf); \\r
151	size = newSize; buf = (Byte *)IAlloc_Alloc(p->mtCoder->alloc, size); \\r
152	if (buf == 0) return SZ_ERROR_MEM; }\r
153	\r
154	static SRes CMtThread_Prepare(CMtThread *p)\r
155	{\r
156	MY_BUF_ALLOC(p->inBuf, p->inBufSize, p->mtCoder->blockSize)\r
157	MY_BUF_ALLOC(p->outBuf, p->outBufSize, p->mtCoder->destBlockSize)\r
158	\r
159	p->stopReading = False;\r
160	p->stopWriting = False;\r
161	RINOK_THREAD(AutoResetEvent_CreateNotSignaled(&p->canRead));\r
162	RINOK_THREAD(AutoResetEvent_CreateNotSignaled(&p->canWrite));\r
163	\r
164	return SZ_OK;\r
165	}\r
166	\r
167	static SRes FullRead(ISeqInStream stream, Byte data, size_t *processedSize)\r
168	{\r
169	size_t size = *processedSize;\r
170	*processedSize = 0;\r
171	while (size != 0)\r
172	{\r
173	size_t curSize = size;\r
174	SRes res = stream->Read(stream, data, &curSize);\r
175	*processedSize += curSize;\r
176	data += curSize;\r
177	size -= curSize;\r
178	RINOK(res);\r
179	if (curSize == 0)\r
180	return SZ_OK;\r
181	}\r
182	return SZ_OK;\r
183	}\r
184	\r
185	#define GET_NEXT_THREAD(p) &p->mtCoder->threads[p->index == p->mtCoder->numThreads - 1 ? 0 : p->index + 1]\r
186	\r
187	static SRes MtThread_Process(CMtThread p, Bool stop)\r
188	{\r
189	CMtThread *next;\r
190	*stop = True;\r
191	if (Event_Wait(&p->canRead) != 0)\r
192	return SZ_ERROR_THREAD;\r
193	\r
194	next = GET_NEXT_THREAD(p);\r
195	\r
196	if (p->stopReading)\r
197	{\r
198	next->stopReading = True;\r
199	return Event_Set(&next->canRead) == 0 ? SZ_OK : SZ_ERROR_THREAD;\r
200	}\r
201	\r
202	{\r
203	size_t size = p->mtCoder->blockSize;\r
204	size_t destSize = p->outBufSize;\r
205	\r
206	RINOK(FullRead(p->mtCoder->inStream, p->inBuf, &size));\r
207	next->stopReading = *stop = (size != p->mtCoder->blockSize);\r
208	if (Event_Set(&next->canRead) != 0)\r
209	return SZ_ERROR_THREAD;\r
210	\r
211	RINOK(p->mtCoder->mtCallback->Code(p->mtCoder->mtCallback, p->index,\r
212	p->outBuf, &destSize, p->inBuf, size, *stop));\r
213	\r
214	MtProgress_Reinit(&p->mtCoder->mtProgress, p->index);\r
215	\r
216	if (Event_Wait(&p->canWrite) != 0)\r
217	return SZ_ERROR_THREAD;\r
218	if (p->stopWriting)\r
219	return SZ_ERROR_FAIL;\r
220	if (p->mtCoder->outStream->Write(p->mtCoder->outStream, p->outBuf, destSize) != destSize)\r
221	return SZ_ERROR_WRITE;\r
222	return Event_Set(&next->canWrite) == 0 ? SZ_OK : SZ_ERROR_THREAD;\r
223	}\r
224	}\r
225	\r
226	static THREAD_FUNC_RET_TYPE THREAD_FUNC_CALL_TYPE ThreadFunc(void *pp)\r
227	{\r
228	CMtThread p = (CMtThread )pp;\r
229	for (;;)\r
230	{\r
231	Bool stop;\r
232	CMtThread *next = GET_NEXT_THREAD(p);\r
233	SRes res = MtThread_Process(p, &stop);\r
234	if (res != SZ_OK)\r
235	{\r
236	MtCoder_SetError(p->mtCoder, res);\r
237	MtProgress_SetError(&p->mtCoder->mtProgress, res);\r
238	next->stopReading = True;\r
239	next->stopWriting = True;\r
240	Event_Set(&next->canRead);\r
241	Event_Set(&next->canWrite);\r
242	return res;\r
243	}\r
244	if (stop)\r
245	return 0;\r
246	}\r
247	}\r
248	\r
249	void MtCoder_Construct(CMtCoder* p)\r
250	{\r
251	unsigned i;\r
252	p->alloc = 0;\r
253	for (i = 0; i < NUM_MT_CODER_THREADS_MAX; i++)\r
254	{\r
255	CMtThread *t = &p->threads[i];\r
256	t->index = i;\r
257	CMtThread_Construct(t, p);\r
258	}\r
259	CriticalSection_Init(&p->cs);\r
260	CriticalSection_Init(&p->mtProgress.cs);\r
261	}\r
262	\r
263	void MtCoder_Destruct(CMtCoder* p)\r
264	{\r
265	unsigned i;\r
266	for (i = 0; i < NUM_MT_CODER_THREADS_MAX; i++)\r
267	CMtThread_Destruct(&p->threads[i]);\r
268	CriticalSection_Delete(&p->cs);\r
269	CriticalSection_Delete(&p->mtProgress.cs);\r
270	}\r
271	\r
272	SRes MtCoder_Code(CMtCoder *p)\r
273	{\r
274	unsigned i, numThreads = p->numThreads;\r
275	SRes res = SZ_OK;\r
276	p->res = SZ_OK;\r
277	\r
278	MtProgress_Init(&p->mtProgress, p->progress);\r
279	\r
280	for (i = 0; i < numThreads; i++)\r
281	{\r
282	RINOK(CMtThread_Prepare(&p->threads[i]));\r
283	}\r
284	\r
285	for (i = 0; i < numThreads; i++)\r
286	{\r
287	CMtThread *t = &p->threads[i];\r
288	CLoopThread *lt = &t->thread;\r
289	\r
290	if (!Thread_WasCreated(&lt->thread))\r
291	{\r
292	lt->func = ThreadFunc;\r
293	lt->param = t;\r
294	\r
295	if (LoopThread_Create(lt) != SZ_OK)\r
296	{\r
297	res = SZ_ERROR_THREAD;\r
298	break;\r
299	}\r
300	}\r
301	}\r
302	\r
303	if (res == SZ_OK)\r
304	{\r
305	unsigned j;\r
306	for (i = 0; i < numThreads; i++)\r
307	{\r
308	CMtThread *t = &p->threads[i];\r
309	if (LoopThread_StartSubThread(&t->thread) != SZ_OK)\r
310	{\r
311	res = SZ_ERROR_THREAD;\r
312	p->threads[0].stopReading = True;\r
313	break;\r
314	}\r
315	}\r
316	\r
317	Event_Set(&p->threads[0].canWrite);\r
318	Event_Set(&p->threads[0].canRead);\r
319	\r
320	for (j = 0; j < i; j++)\r
321	LoopThread_WaitSubThread(&p->threads[j].thread);\r
322	}\r
323	\r
324	for (i = 0; i < numThreads; i++)\r
325	CMtThread_CloseEvents(&p->threads[i]);\r
326	return (res == SZ_OK) ? p->res : res;\r
327	}\r