xref: /freebsd/contrib/xz/src/liblzma/common/stream_encoder_mt.c (revision 39ee7a7a6bdd1557b1c3532abf60d139798ac88b)
1 ///////////////////////////////////////////////////////////////////////////////
2 //
3 /// \file       stream_encoder_mt.c
4 /// \brief      Multithreaded .xz Stream encoder
5 //
6 //  Author:     Lasse Collin
7 //
8 //  This file has been put into the public domain.
9 //  You can do whatever you want with this file.
10 //
11 ///////////////////////////////////////////////////////////////////////////////
12 
13 #include "filter_encoder.h"
14 #include "easy_preset.h"
15 #include "block_encoder.h"
16 #include "block_buffer_encoder.h"
17 #include "index_encoder.h"
18 #include "outqueue.h"
19 
20 
21 /// Maximum supported block size. This makes it simpler to prevent integer
22 /// overflows if we are given unusually large block size.
23 #define BLOCK_SIZE_MAX (UINT64_MAX / LZMA_THREADS_MAX)
24 
25 
26 typedef enum {
27 	/// Waiting for work.
28 	THR_IDLE,
29 
30 	/// Encoding is in progress.
31 	THR_RUN,
32 
33 	/// Encoding is in progress but no more input data will
34 	/// be read.
35 	THR_FINISH,
36 
37 	/// The main thread wants the thread to stop whatever it was doing
38 	/// but not exit.
39 	THR_STOP,
40 
41 	/// The main thread wants the thread to exit. We could use
42 	/// cancellation but since there's stopped anyway, this is lazier.
43 	THR_EXIT,
44 
45 } worker_state;
46 
47 
48 typedef struct worker_thread_s worker_thread;
49 struct worker_thread_s {
50 	worker_state state;
51 
52 	/// Input buffer of coder->block_size bytes. The main thread will
53 	/// put new input into this and update in_size accordingly. Once
54 	/// no more input is coming, state will be set to THR_FINISH.
55 	uint8_t *in;
56 
57 	/// Amount of data available in the input buffer. This is modified
58 	/// only by the main thread.
59 	size_t in_size;
60 
61 	/// Output buffer for this thread. This is set by the main
62 	/// thread every time a new Block is started with this thread
63 	/// structure.
64 	lzma_outbuf *outbuf;
65 
66 	/// Pointer to the main structure is needed when putting this
67 	/// thread back to the stack of free threads.
68 	lzma_coder *coder;
69 
70 	/// The allocator is set by the main thread. Since a copy of the
71 	/// pointer is kept here, the application must not change the
72 	/// allocator before calling lzma_end().
73 	const lzma_allocator *allocator;
74 
75 	/// Amount of uncompressed data that has already been compressed.
76 	uint64_t progress_in;
77 
78 	/// Amount of compressed data that is ready.
79 	uint64_t progress_out;
80 
81 	/// Block encoder
82 	lzma_next_coder block_encoder;
83 
84 	/// Compression options for this Block
85 	lzma_block block_options;
86 
87 	/// Next structure in the stack of free worker threads.
88 	worker_thread *next;
89 
90 	mythread_mutex mutex;
91 	mythread_cond cond;
92 
93 	/// The ID of this thread is used to join the thread
94 	/// when it's not needed anymore.
95 	mythread thread_id;
96 };
97 
98 
99 struct lzma_coder_s {
100 	enum {
101 		SEQ_STREAM_HEADER,
102 		SEQ_BLOCK,
103 		SEQ_INDEX,
104 		SEQ_STREAM_FOOTER,
105 	} sequence;
106 
107 	/// Start a new Block every block_size bytes of input unless
108 	/// LZMA_FULL_FLUSH or LZMA_FULL_BARRIER is used earlier.
109 	size_t block_size;
110 
111 	/// The filter chain currently in use
112 	lzma_filter filters[LZMA_FILTERS_MAX + 1];
113 
114 
115 	/// Index to hold sizes of the Blocks
116 	lzma_index *index;
117 
118 	/// Index encoder
119 	lzma_next_coder index_encoder;
120 
121 
122 	/// Stream Flags for encoding the Stream Header and Stream Footer.
123 	lzma_stream_flags stream_flags;
124 
125 	/// Buffer to hold Stream Header and Stream Footer.
126 	uint8_t header[LZMA_STREAM_HEADER_SIZE];
127 
128 	/// Read position in header[]
129 	size_t header_pos;
130 
131 
132 	/// Output buffer queue for compressed data
133 	lzma_outq outq;
134 
135 
136 	/// Maximum wait time if cannot use all the input and cannot
137 	/// fill the output buffer. This is in milliseconds.
138 	uint32_t timeout;
139 
140 
141 	/// Error code from a worker thread
142 	lzma_ret thread_error;
143 
144 	/// Array of allocated thread-specific structures
145 	worker_thread *threads;
146 
147 	/// Number of structures in "threads" above. This is also the
148 	/// number of threads that will be created at maximum.
149 	uint32_t threads_max;
150 
151 	/// Number of thread structures that have been initialized, and
152 	/// thus the number of worker threads actually created so far.
153 	uint32_t threads_initialized;
154 
155 	/// Stack of free threads. When a thread finishes, it puts itself
156 	/// back into this stack. This starts as empty because threads
157 	/// are created only when actually needed.
158 	worker_thread *threads_free;
159 
160 	/// The most recent worker thread to which the main thread writes
161 	/// the new input from the application.
162 	worker_thread *thr;
163 
164 
165 	/// Amount of uncompressed data in Blocks that have already
166 	/// been finished.
167 	uint64_t progress_in;
168 
169 	/// Amount of compressed data in Stream Header + Blocks that
170 	/// have already been finished.
171 	uint64_t progress_out;
172 
173 
174 	mythread_mutex mutex;
175 	mythread_cond cond;
176 };
177 
178 
179 /// Tell the main thread that something has gone wrong.
180 static void
181 worker_error(worker_thread *thr, lzma_ret ret)
182 {
183 	assert(ret != LZMA_OK);
184 	assert(ret != LZMA_STREAM_END);
185 
186 	mythread_sync(thr->coder->mutex) {
187 		if (thr->coder->thread_error == LZMA_OK)
188 			thr->coder->thread_error = ret;
189 
190 		mythread_cond_signal(&thr->coder->cond);
191 	}
192 
193 	return;
194 }
195 
196 
197 static worker_state
198 worker_encode(worker_thread *thr, worker_state state)
199 {
200 	assert(thr->progress_in == 0);
201 	assert(thr->progress_out == 0);
202 
203 	// Set the Block options.
204 	thr->block_options = (lzma_block){
205 		.version = 0,
206 		.check = thr->coder->stream_flags.check,
207 		.compressed_size = thr->coder->outq.buf_size_max,
208 		.uncompressed_size = thr->coder->block_size,
209 
210 		// TODO: To allow changing the filter chain, the filters
211 		// array must be copied to each worker_thread.
212 		.filters = thr->coder->filters,
213 	};
214 
215 	// Calculate maximum size of the Block Header. This amount is
216 	// reserved in the beginning of the buffer so that Block Header
217 	// along with Compressed Size and Uncompressed Size can be
218 	// written there.
219 	lzma_ret ret = lzma_block_header_size(&thr->block_options);
220 	if (ret != LZMA_OK) {
221 		worker_error(thr, ret);
222 		return THR_STOP;
223 	}
224 
225 	// Initialize the Block encoder.
226 	ret = lzma_block_encoder_init(&thr->block_encoder,
227 			thr->allocator, &thr->block_options);
228 	if (ret != LZMA_OK) {
229 		worker_error(thr, ret);
230 		return THR_STOP;
231 	}
232 
233 	size_t in_pos = 0;
234 	size_t in_size = 0;
235 
236 	thr->outbuf->size = thr->block_options.header_size;
237 	const size_t out_size = thr->coder->outq.buf_size_max;
238 
239 	do {
240 		mythread_sync(thr->mutex) {
241 			// Store in_pos and out_pos into *thr so that
242 			// an application may read them via
243 			// lzma_get_progress() to get progress information.
244 			//
245 			// NOTE: These aren't updated when the encoding
246 			// finishes. Instead, the final values are taken
247 			// later from thr->outbuf.
248 			thr->progress_in = in_pos;
249 			thr->progress_out = thr->outbuf->size;
250 
251 			while (in_size == thr->in_size
252 					&& thr->state == THR_RUN)
253 				mythread_cond_wait(&thr->cond, &thr->mutex);
254 
255 			state = thr->state;
256 			in_size = thr->in_size;
257 		}
258 
259 		// Return if we were asked to stop or exit.
260 		if (state >= THR_STOP)
261 			return state;
262 
263 		lzma_action action = state == THR_FINISH
264 				? LZMA_FINISH : LZMA_RUN;
265 
266 		// Limit the amount of input given to the Block encoder
267 		// at once. This way this thread can react fairly quickly
268 		// if the main thread wants us to stop or exit.
269 		static const size_t in_chunk_max = 16384;
270 		size_t in_limit = in_size;
271 		if (in_size - in_pos > in_chunk_max) {
272 			in_limit = in_pos + in_chunk_max;
273 			action = LZMA_RUN;
274 		}
275 
276 		ret = thr->block_encoder.code(
277 				thr->block_encoder.coder, thr->allocator,
278 				thr->in, &in_pos, in_limit, thr->outbuf->buf,
279 				&thr->outbuf->size, out_size, action);
280 	} while (ret == LZMA_OK && thr->outbuf->size < out_size);
281 
282 	switch (ret) {
283 	case LZMA_STREAM_END:
284 		assert(state == THR_FINISH);
285 
286 		// Encode the Block Header. By doing it after
287 		// the compression, we can store the Compressed Size
288 		// and Uncompressed Size fields.
289 		ret = lzma_block_header_encode(&thr->block_options,
290 				thr->outbuf->buf);
291 		if (ret != LZMA_OK) {
292 			worker_error(thr, ret);
293 			return THR_STOP;
294 		}
295 
296 		break;
297 
298 	case LZMA_OK:
299 		// The data was incompressible. Encode it using uncompressed
300 		// LZMA2 chunks.
301 		//
302 		// First wait that we have gotten all the input.
303 		mythread_sync(thr->mutex) {
304 			while (thr->state == THR_RUN)
305 				mythread_cond_wait(&thr->cond, &thr->mutex);
306 
307 			state = thr->state;
308 			in_size = thr->in_size;
309 		}
310 
311 		if (state >= THR_STOP)
312 			return state;
313 
314 		// Do the encoding. This takes care of the Block Header too.
315 		thr->outbuf->size = 0;
316 		ret = lzma_block_uncomp_encode(&thr->block_options,
317 				thr->in, in_size, thr->outbuf->buf,
318 				&thr->outbuf->size, out_size);
319 
320 		// It shouldn't fail.
321 		if (ret != LZMA_OK) {
322 			worker_error(thr, LZMA_PROG_ERROR);
323 			return THR_STOP;
324 		}
325 
326 		break;
327 
328 	default:
329 		worker_error(thr, ret);
330 		return THR_STOP;
331 	}
332 
333 	// Set the size information that will be read by the main thread
334 	// to write the Index field.
335 	thr->outbuf->unpadded_size
336 			= lzma_block_unpadded_size(&thr->block_options);
337 	assert(thr->outbuf->unpadded_size != 0);
338 	thr->outbuf->uncompressed_size = thr->block_options.uncompressed_size;
339 
340 	return THR_FINISH;
341 }
342 
343 
344 static MYTHREAD_RET_TYPE
345 worker_start(void *thr_ptr)
346 {
347 	worker_thread *thr = thr_ptr;
348 	worker_state state = THR_IDLE; // Init to silence a warning
349 
350 	while (true) {
351 		// Wait for work.
352 		mythread_sync(thr->mutex) {
353 			while (true) {
354 				// The thread is already idle so if we are
355 				// requested to stop, just set the state.
356 				if (thr->state == THR_STOP) {
357 					thr->state = THR_IDLE;
358 					mythread_cond_signal(&thr->cond);
359 				}
360 
361 				state = thr->state;
362 				if (state != THR_IDLE)
363 					break;
364 
365 				mythread_cond_wait(&thr->cond, &thr->mutex);
366 			}
367 		}
368 
369 		assert(state != THR_IDLE);
370 		assert(state != THR_STOP);
371 
372 		if (state <= THR_FINISH)
373 			state = worker_encode(thr, state);
374 
375 		if (state == THR_EXIT)
376 			break;
377 
378 		// Mark the thread as idle unless the main thread has
379 		// told us to exit. Signal is needed for the case
380 		// where the main thread is waiting for the threads to stop.
381 		mythread_sync(thr->mutex) {
382 			if (thr->state != THR_EXIT) {
383 				thr->state = THR_IDLE;
384 				mythread_cond_signal(&thr->cond);
385 			}
386 		}
387 
388 		mythread_sync(thr->coder->mutex) {
389 			// Mark the output buffer as finished if
390 			// no errors occurred.
391 			thr->outbuf->finished = state == THR_FINISH;
392 
393 			// Update the main progress info.
394 			thr->coder->progress_in
395 					+= thr->outbuf->uncompressed_size;
396 			thr->coder->progress_out += thr->outbuf->size;
397 			thr->progress_in = 0;
398 			thr->progress_out = 0;
399 
400 			// Return this thread to the stack of free threads.
401 			thr->next = thr->coder->threads_free;
402 			thr->coder->threads_free = thr;
403 
404 			mythread_cond_signal(&thr->coder->cond);
405 		}
406 	}
407 
408 	// Exiting, free the resources.
409 	mythread_mutex_destroy(&thr->mutex);
410 	mythread_cond_destroy(&thr->cond);
411 
412 	lzma_next_end(&thr->block_encoder, thr->allocator);
413 	lzma_free(thr->in, thr->allocator);
414 	return MYTHREAD_RET_VALUE;
415 }
416 
417 
418 /// Make the threads stop but not exit. Optionally wait for them to stop.
419 static void
420 threads_stop(lzma_coder *coder, bool wait_for_threads)
421 {
422 	// Tell the threads to stop.
423 	for (uint32_t i = 0; i < coder->threads_initialized; ++i) {
424 		mythread_sync(coder->threads[i].mutex) {
425 			coder->threads[i].state = THR_STOP;
426 			mythread_cond_signal(&coder->threads[i].cond);
427 		}
428 	}
429 
430 	if (!wait_for_threads)
431 		return;
432 
433 	// Wait for the threads to settle in the idle state.
434 	for (uint32_t i = 0; i < coder->threads_initialized; ++i) {
435 		mythread_sync(coder->threads[i].mutex) {
436 			while (coder->threads[i].state != THR_IDLE)
437 				mythread_cond_wait(&coder->threads[i].cond,
438 						&coder->threads[i].mutex);
439 		}
440 	}
441 
442 	return;
443 }
444 
445 
446 /// Stop the threads and free the resources associated with them.
447 /// Wait until the threads have exited.
448 static void
449 threads_end(lzma_coder *coder, const lzma_allocator *allocator)
450 {
451 	for (uint32_t i = 0; i < coder->threads_initialized; ++i) {
452 		mythread_sync(coder->threads[i].mutex) {
453 			coder->threads[i].state = THR_EXIT;
454 			mythread_cond_signal(&coder->threads[i].cond);
455 		}
456 	}
457 
458 	for (uint32_t i = 0; i < coder->threads_initialized; ++i) {
459 		int ret = mythread_join(coder->threads[i].thread_id);
460 		assert(ret == 0);
461 		(void)ret;
462 	}
463 
464 	lzma_free(coder->threads, allocator);
465 	return;
466 }
467 
468 
469 /// Initialize a new worker_thread structure and create a new thread.
470 static lzma_ret
471 initialize_new_thread(lzma_coder *coder, const lzma_allocator *allocator)
472 {
473 	worker_thread *thr = &coder->threads[coder->threads_initialized];
474 
475 	thr->in = lzma_alloc(coder->block_size, allocator);
476 	if (thr->in == NULL)
477 		return LZMA_MEM_ERROR;
478 
479 	if (mythread_mutex_init(&thr->mutex))
480 		goto error_mutex;
481 
482 	if (mythread_cond_init(&thr->cond))
483 		goto error_cond;
484 
485 	thr->state = THR_IDLE;
486 	thr->allocator = allocator;
487 	thr->coder = coder;
488 	thr->progress_in = 0;
489 	thr->progress_out = 0;
490 	thr->block_encoder = LZMA_NEXT_CODER_INIT;
491 
492 	if (mythread_create(&thr->thread_id, &worker_start, thr))
493 		goto error_thread;
494 
495 	++coder->threads_initialized;
496 	coder->thr = thr;
497 
498 	return LZMA_OK;
499 
500 error_thread:
501 	mythread_cond_destroy(&thr->cond);
502 
503 error_cond:
504 	mythread_mutex_destroy(&thr->mutex);
505 
506 error_mutex:
507 	lzma_free(thr->in, allocator);
508 	return LZMA_MEM_ERROR;
509 }
510 
511 
512 static lzma_ret
513 get_thread(lzma_coder *coder, const lzma_allocator *allocator)
514 {
515 	// If there are no free output subqueues, there is no
516 	// point to try getting a thread.
517 	if (!lzma_outq_has_buf(&coder->outq))
518 		return LZMA_OK;
519 
520 	// If there is a free structure on the stack, use it.
521 	mythread_sync(coder->mutex) {
522 		if (coder->threads_free != NULL) {
523 			coder->thr = coder->threads_free;
524 			coder->threads_free = coder->threads_free->next;
525 		}
526 	}
527 
528 	if (coder->thr == NULL) {
529 		// If there are no uninitialized structures left, return.
530 		if (coder->threads_initialized == coder->threads_max)
531 			return LZMA_OK;
532 
533 		// Initialize a new thread.
534 		return_if_error(initialize_new_thread(coder, allocator));
535 	}
536 
537 	// Reset the parts of the thread state that have to be done
538 	// in the main thread.
539 	mythread_sync(coder->thr->mutex) {
540 		coder->thr->state = THR_RUN;
541 		coder->thr->in_size = 0;
542 		coder->thr->outbuf = lzma_outq_get_buf(&coder->outq);
543 		mythread_cond_signal(&coder->thr->cond);
544 	}
545 
546 	return LZMA_OK;
547 }
548 
549 
550 static lzma_ret
551 stream_encode_in(lzma_coder *coder, const lzma_allocator *allocator,
552 		const uint8_t *restrict in, size_t *restrict in_pos,
553 		size_t in_size, lzma_action action)
554 {
555 	while (*in_pos < in_size
556 			|| (coder->thr != NULL && action != LZMA_RUN)) {
557 		if (coder->thr == NULL) {
558 			// Get a new thread.
559 			const lzma_ret ret = get_thread(coder, allocator);
560 			if (coder->thr == NULL)
561 				return ret;
562 		}
563 
564 		// Copy the input data to thread's buffer.
565 		size_t thr_in_size = coder->thr->in_size;
566 		lzma_bufcpy(in, in_pos, in_size, coder->thr->in,
567 				&thr_in_size, coder->block_size);
568 
569 		// Tell the Block encoder to finish if
570 		//  - it has got block_size bytes of input; or
571 		//  - all input was used and LZMA_FINISH, LZMA_FULL_FLUSH,
572 		//    or LZMA_FULL_BARRIER was used.
573 		//
574 		// TODO: LZMA_SYNC_FLUSH and LZMA_SYNC_BARRIER.
575 		const bool finish = thr_in_size == coder->block_size
576 				|| (*in_pos == in_size && action != LZMA_RUN);
577 
578 		bool block_error = false;
579 
580 		mythread_sync(coder->thr->mutex) {
581 			if (coder->thr->state == THR_IDLE) {
582 				// Something has gone wrong with the Block
583 				// encoder. It has set coder->thread_error
584 				// which we will read a few lines later.
585 				block_error = true;
586 			} else {
587 				// Tell the Block encoder its new amount
588 				// of input and update the state if needed.
589 				coder->thr->in_size = thr_in_size;
590 
591 				if (finish)
592 					coder->thr->state = THR_FINISH;
593 
594 				mythread_cond_signal(&coder->thr->cond);
595 			}
596 		}
597 
598 		if (block_error) {
599 			lzma_ret ret;
600 
601 			mythread_sync(coder->mutex) {
602 				ret = coder->thread_error;
603 			}
604 
605 			return ret;
606 		}
607 
608 		if (finish)
609 			coder->thr = NULL;
610 	}
611 
612 	return LZMA_OK;
613 }
614 
615 
616 /// Wait until more input can be consumed, more output can be read, or
617 /// an optional timeout is reached.
618 static bool
619 wait_for_work(lzma_coder *coder, mythread_condtime *wait_abs,
620 		bool *has_blocked, bool has_input)
621 {
622 	if (coder->timeout != 0 && !*has_blocked) {
623 		// Every time when stream_encode_mt() is called via
624 		// lzma_code(), *has_blocked starts as false. We set it
625 		// to true here and calculate the absolute time when
626 		// we must return if there's nothing to do.
627 		//
628 		// The idea of *has_blocked is to avoid unneeded calls
629 		// to mythread_condtime_set(), which may do a syscall
630 		// depending on the operating system.
631 		*has_blocked = true;
632 		mythread_condtime_set(wait_abs, &coder->cond, coder->timeout);
633 	}
634 
635 	bool timed_out = false;
636 
637 	mythread_sync(coder->mutex) {
638 		// There are four things that we wait. If one of them
639 		// becomes possible, we return.
640 		//  - If there is input left, we need to get a free
641 		//    worker thread and an output buffer for it.
642 		//  - Data ready to be read from the output queue.
643 		//  - A worker thread indicates an error.
644 		//  - Time out occurs.
645 		while ((!has_input || coder->threads_free == NULL
646 					|| !lzma_outq_has_buf(&coder->outq))
647 				&& !lzma_outq_is_readable(&coder->outq)
648 				&& coder->thread_error == LZMA_OK
649 				&& !timed_out) {
650 			if (coder->timeout != 0)
651 				timed_out = mythread_cond_timedwait(
652 						&coder->cond, &coder->mutex,
653 						wait_abs) != 0;
654 			else
655 				mythread_cond_wait(&coder->cond,
656 						&coder->mutex);
657 		}
658 	}
659 
660 	return timed_out;
661 }
662 
663 
664 static lzma_ret
665 stream_encode_mt(lzma_coder *coder, const lzma_allocator *allocator,
666 		const uint8_t *restrict in, size_t *restrict in_pos,
667 		size_t in_size, uint8_t *restrict out,
668 		size_t *restrict out_pos, size_t out_size, lzma_action action)
669 {
670 	switch (coder->sequence) {
671 	case SEQ_STREAM_HEADER:
672 		lzma_bufcpy(coder->header, &coder->header_pos,
673 				sizeof(coder->header),
674 				out, out_pos, out_size);
675 		if (coder->header_pos < sizeof(coder->header))
676 			return LZMA_OK;
677 
678 		coder->header_pos = 0;
679 		coder->sequence = SEQ_BLOCK;
680 
681 	// Fall through
682 
683 	case SEQ_BLOCK: {
684 		// Initialized to silence warnings.
685 		lzma_vli unpadded_size = 0;
686 		lzma_vli uncompressed_size = 0;
687 		lzma_ret ret = LZMA_OK;
688 
689 		// These are for wait_for_work().
690 		bool has_blocked = false;
691 		mythread_condtime wait_abs;
692 
693 		while (true) {
694 			mythread_sync(coder->mutex) {
695 				// Check for Block encoder errors.
696 				ret = coder->thread_error;
697 				if (ret != LZMA_OK) {
698 					assert(ret != LZMA_STREAM_END);
699 					break;
700 				}
701 
702 				// Try to read compressed data to out[].
703 				ret = lzma_outq_read(&coder->outq,
704 						out, out_pos, out_size,
705 						&unpadded_size,
706 						&uncompressed_size);
707 			}
708 
709 			if (ret == LZMA_STREAM_END) {
710 				// End of Block. Add it to the Index.
711 				ret = lzma_index_append(coder->index,
712 						allocator, unpadded_size,
713 						uncompressed_size);
714 
715 				// If we didn't fill the output buffer yet,
716 				// try to read more data. Maybe the next
717 				// outbuf has been finished already too.
718 				if (*out_pos < out_size)
719 					continue;
720 			}
721 
722 			if (ret != LZMA_OK) {
723 				// coder->thread_error was set or
724 				// lzma_index_append() failed.
725 				threads_stop(coder, false);
726 				return ret;
727 			}
728 
729 			// Try to give uncompressed data to a worker thread.
730 			ret = stream_encode_in(coder, allocator,
731 					in, in_pos, in_size, action);
732 			if (ret != LZMA_OK) {
733 				threads_stop(coder, false);
734 				return ret;
735 			}
736 
737 			// See if we should wait or return.
738 			//
739 			// TODO: LZMA_SYNC_FLUSH and LZMA_SYNC_BARRIER.
740 			if (*in_pos == in_size) {
741 				// LZMA_RUN: More data is probably coming
742 				// so return to let the caller fill the
743 				// input buffer.
744 				if (action == LZMA_RUN)
745 					return LZMA_OK;
746 
747 				// LZMA_FULL_BARRIER: The same as with
748 				// LZMA_RUN but tell the caller that the
749 				// barrier was completed.
750 				if (action == LZMA_FULL_BARRIER)
751 					return LZMA_STREAM_END;
752 
753 				// Finishing or flushing isn't completed until
754 				// all input data has been encoded and copied
755 				// to the output buffer.
756 				if (lzma_outq_is_empty(&coder->outq)) {
757 					// LZMA_FINISH: Continue to encode
758 					// the Index field.
759 					if (action == LZMA_FINISH)
760 						break;
761 
762 					// LZMA_FULL_FLUSH: Return to tell
763 					// the caller that flushing was
764 					// completed.
765 					if (action == LZMA_FULL_FLUSH)
766 						return LZMA_STREAM_END;
767 				}
768 			}
769 
770 			// Return if there is no output space left.
771 			// This check must be done after testing the input
772 			// buffer, because we might want to use a different
773 			// return code.
774 			if (*out_pos == out_size)
775 				return LZMA_OK;
776 
777 			// Neither in nor out has been used completely.
778 			// Wait until there's something we can do.
779 			if (wait_for_work(coder, &wait_abs, &has_blocked,
780 					*in_pos < in_size))
781 				return LZMA_TIMED_OUT;
782 		}
783 
784 		// All Blocks have been encoded and the threads have stopped.
785 		// Prepare to encode the Index field.
786 		return_if_error(lzma_index_encoder_init(
787 				&coder->index_encoder, allocator,
788 				coder->index));
789 		coder->sequence = SEQ_INDEX;
790 
791 		// Update the progress info to take the Index and
792 		// Stream Footer into account. Those are very fast to encode
793 		// so in terms of progress information they can be thought
794 		// to be ready to be copied out.
795 		coder->progress_out += lzma_index_size(coder->index)
796 				+ LZMA_STREAM_HEADER_SIZE;
797 	}
798 
799 	// Fall through
800 
801 	case SEQ_INDEX: {
802 		// Call the Index encoder. It doesn't take any input, so
803 		// those pointers can be NULL.
804 		const lzma_ret ret = coder->index_encoder.code(
805 				coder->index_encoder.coder, allocator,
806 				NULL, NULL, 0,
807 				out, out_pos, out_size, LZMA_RUN);
808 		if (ret != LZMA_STREAM_END)
809 			return ret;
810 
811 		// Encode the Stream Footer into coder->buffer.
812 		coder->stream_flags.backward_size
813 				= lzma_index_size(coder->index);
814 		if (lzma_stream_footer_encode(&coder->stream_flags,
815 				coder->header) != LZMA_OK)
816 			return LZMA_PROG_ERROR;
817 
818 		coder->sequence = SEQ_STREAM_FOOTER;
819 	}
820 
821 	// Fall through
822 
823 	case SEQ_STREAM_FOOTER:
824 		lzma_bufcpy(coder->header, &coder->header_pos,
825 				sizeof(coder->header),
826 				out, out_pos, out_size);
827 		return coder->header_pos < sizeof(coder->header)
828 				? LZMA_OK : LZMA_STREAM_END;
829 	}
830 
831 	assert(0);
832 	return LZMA_PROG_ERROR;
833 }
834 
835 
836 static void
837 stream_encoder_mt_end(lzma_coder *coder, const lzma_allocator *allocator)
838 {
839 	// Threads must be killed before the output queue can be freed.
840 	threads_end(coder, allocator);
841 	lzma_outq_end(&coder->outq, allocator);
842 
843 	for (size_t i = 0; coder->filters[i].id != LZMA_VLI_UNKNOWN; ++i)
844 		lzma_free(coder->filters[i].options, allocator);
845 
846 	lzma_next_end(&coder->index_encoder, allocator);
847 	lzma_index_end(coder->index, allocator);
848 
849 	mythread_cond_destroy(&coder->cond);
850 	mythread_mutex_destroy(&coder->mutex);
851 
852 	lzma_free(coder, allocator);
853 	return;
854 }
855 
856 
857 /// Options handling for lzma_stream_encoder_mt_init() and
858 /// lzma_stream_encoder_mt_memusage()
859 static lzma_ret
860 get_options(const lzma_mt *options, lzma_options_easy *opt_easy,
861 		const lzma_filter **filters, uint64_t *block_size,
862 		uint64_t *outbuf_size_max)
863 {
864 	// Validate some of the options.
865 	if (options == NULL)
866 		return LZMA_PROG_ERROR;
867 
868 	if (options->flags != 0 || options->threads == 0
869 			|| options->threads > LZMA_THREADS_MAX)
870 		return LZMA_OPTIONS_ERROR;
871 
872 	if (options->filters != NULL) {
873 		// Filter chain was given, use it as is.
874 		*filters = options->filters;
875 	} else {
876 		// Use a preset.
877 		if (lzma_easy_preset(opt_easy, options->preset))
878 			return LZMA_OPTIONS_ERROR;
879 
880 		*filters = opt_easy->filters;
881 	}
882 
883 	// Block size
884 	if (options->block_size > 0) {
885 		if (options->block_size > BLOCK_SIZE_MAX)
886 			return LZMA_OPTIONS_ERROR;
887 
888 		*block_size = options->block_size;
889 	} else {
890 		// Determine the Block size from the filter chain.
891 		*block_size = lzma_mt_block_size(*filters);
892 		if (*block_size == 0)
893 			return LZMA_OPTIONS_ERROR;
894 
895 		assert(*block_size <= BLOCK_SIZE_MAX);
896 	}
897 
898 	// Calculate the maximum amount output that a single output buffer
899 	// may need to hold. This is the same as the maximum total size of
900 	// a Block.
901 	*outbuf_size_max = lzma_block_buffer_bound64(*block_size);
902 	if (*outbuf_size_max == 0)
903 		return LZMA_MEM_ERROR;
904 
905 	return LZMA_OK;
906 }
907 
908 
909 static void
910 get_progress(lzma_coder *coder, uint64_t *progress_in, uint64_t *progress_out)
911 {
912 	// Lock coder->mutex to prevent finishing threads from moving their
913 	// progress info from the worker_thread structure to lzma_coder.
914 	mythread_sync(coder->mutex) {
915 		*progress_in = coder->progress_in;
916 		*progress_out = coder->progress_out;
917 
918 		for (size_t i = 0; i < coder->threads_initialized; ++i) {
919 			mythread_sync(coder->threads[i].mutex) {
920 				*progress_in += coder->threads[i].progress_in;
921 				*progress_out += coder->threads[i]
922 						.progress_out;
923 			}
924 		}
925 	}
926 
927 	return;
928 }
929 
930 
931 static lzma_ret
932 stream_encoder_mt_init(lzma_next_coder *next, const lzma_allocator *allocator,
933 		const lzma_mt *options)
934 {
935 	lzma_next_coder_init(&stream_encoder_mt_init, next, allocator);
936 
937 	// Get the filter chain.
938 	lzma_options_easy easy;
939 	const lzma_filter *filters;
940 	uint64_t block_size;
941 	uint64_t outbuf_size_max;
942 	return_if_error(get_options(options, &easy, &filters,
943 			&block_size, &outbuf_size_max));
944 
945 #if SIZE_MAX < UINT64_MAX
946 	if (block_size > SIZE_MAX)
947 		return LZMA_MEM_ERROR;
948 #endif
949 
950 	// Validate the filter chain so that we can give an error in this
951 	// function instead of delaying it to the first call to lzma_code().
952 	// The memory usage calculation verifies the filter chain as
953 	// a side effect so we take advatange of that.
954 	if (lzma_raw_encoder_memusage(filters) == UINT64_MAX)
955 		return LZMA_OPTIONS_ERROR;
956 
957 	// Validate the Check ID.
958 	if ((unsigned int)(options->check) > LZMA_CHECK_ID_MAX)
959 		return LZMA_PROG_ERROR;
960 
961 	if (!lzma_check_is_supported(options->check))
962 		return LZMA_UNSUPPORTED_CHECK;
963 
964 	// Allocate and initialize the base structure if needed.
965 	if (next->coder == NULL) {
966 		next->coder = lzma_alloc(sizeof(lzma_coder), allocator);
967 		if (next->coder == NULL)
968 			return LZMA_MEM_ERROR;
969 
970 		// For the mutex and condition variable initializations
971 		// the error handling has to be done here because
972 		// stream_encoder_mt_end() doesn't know if they have
973 		// already been initialized or not.
974 		if (mythread_mutex_init(&next->coder->mutex)) {
975 			lzma_free(next->coder, allocator);
976 			next->coder = NULL;
977 			return LZMA_MEM_ERROR;
978 		}
979 
980 		if (mythread_cond_init(&next->coder->cond)) {
981 			mythread_mutex_destroy(&next->coder->mutex);
982 			lzma_free(next->coder, allocator);
983 			next->coder = NULL;
984 			return LZMA_MEM_ERROR;
985 		}
986 
987 		next->code = &stream_encode_mt;
988 		next->end = &stream_encoder_mt_end;
989 		next->get_progress = &get_progress;
990 // 		next->update = &stream_encoder_mt_update;
991 
992 		next->coder->filters[0].id = LZMA_VLI_UNKNOWN;
993 		next->coder->index_encoder = LZMA_NEXT_CODER_INIT;
994 		next->coder->index = NULL;
995 		memzero(&next->coder->outq, sizeof(next->coder->outq));
996 		next->coder->threads = NULL;
997 		next->coder->threads_max = 0;
998 		next->coder->threads_initialized = 0;
999 	}
1000 
1001 	// Basic initializations
1002 	next->coder->sequence = SEQ_STREAM_HEADER;
1003 	next->coder->block_size = (size_t)(block_size);
1004 	next->coder->thread_error = LZMA_OK;
1005 	next->coder->thr = NULL;
1006 
1007 	// Allocate the thread-specific base structures.
1008 	assert(options->threads > 0);
1009 	if (next->coder->threads_max != options->threads) {
1010 		threads_end(next->coder, allocator);
1011 
1012 		next->coder->threads = NULL;
1013 		next->coder->threads_max = 0;
1014 
1015 		next->coder->threads_initialized = 0;
1016 		next->coder->threads_free = NULL;
1017 
1018 		next->coder->threads = lzma_alloc(
1019 				options->threads * sizeof(worker_thread),
1020 				allocator);
1021 		if (next->coder->threads == NULL)
1022 			return LZMA_MEM_ERROR;
1023 
1024 		next->coder->threads_max = options->threads;
1025 	} else {
1026 		// Reuse the old structures and threads. Tell the running
1027 		// threads to stop and wait until they have stopped.
1028 		threads_stop(next->coder, true);
1029 	}
1030 
1031 	// Output queue
1032 	return_if_error(lzma_outq_init(&next->coder->outq, allocator,
1033 			outbuf_size_max, options->threads));
1034 
1035 	// Timeout
1036 	next->coder->timeout = options->timeout;
1037 
1038 	// Free the old filter chain and copy the new one.
1039 	for (size_t i = 0; next->coder->filters[i].id != LZMA_VLI_UNKNOWN; ++i)
1040 		lzma_free(next->coder->filters[i].options, allocator);
1041 
1042 	return_if_error(lzma_filters_copy(
1043 			filters, next->coder->filters, allocator));
1044 
1045 	// Index
1046 	lzma_index_end(next->coder->index, allocator);
1047 	next->coder->index = lzma_index_init(allocator);
1048 	if (next->coder->index == NULL)
1049 		return LZMA_MEM_ERROR;
1050 
1051 	// Stream Header
1052 	next->coder->stream_flags.version = 0;
1053 	next->coder->stream_flags.check = options->check;
1054 	return_if_error(lzma_stream_header_encode(
1055 			&next->coder->stream_flags, next->coder->header));
1056 
1057 	next->coder->header_pos = 0;
1058 
1059 	// Progress info
1060 	next->coder->progress_in = 0;
1061 	next->coder->progress_out = LZMA_STREAM_HEADER_SIZE;
1062 
1063 	return LZMA_OK;
1064 }
1065 
1066 
1067 extern LZMA_API(lzma_ret)
1068 lzma_stream_encoder_mt(lzma_stream *strm, const lzma_mt *options)
1069 {
1070 	lzma_next_strm_init(stream_encoder_mt_init, strm, options);
1071 
1072 	strm->internal->supported_actions[LZMA_RUN] = true;
1073 // 	strm->internal->supported_actions[LZMA_SYNC_FLUSH] = true;
1074 	strm->internal->supported_actions[LZMA_FULL_FLUSH] = true;
1075 	strm->internal->supported_actions[LZMA_FULL_BARRIER] = true;
1076 	strm->internal->supported_actions[LZMA_FINISH] = true;
1077 
1078 	return LZMA_OK;
1079 }
1080 
1081 
1082 // This function name is a monster but it's consistent with the older
1083 // monster names. :-( 31 chars is the max that C99 requires so in that
1084 // sense it's not too long. ;-)
1085 extern LZMA_API(uint64_t)
1086 lzma_stream_encoder_mt_memusage(const lzma_mt *options)
1087 {
1088 	lzma_options_easy easy;
1089 	const lzma_filter *filters;
1090 	uint64_t block_size;
1091 	uint64_t outbuf_size_max;
1092 
1093 	if (get_options(options, &easy, &filters, &block_size,
1094 			&outbuf_size_max) != LZMA_OK)
1095 		return UINT64_MAX;
1096 
1097 	// Memory usage of the input buffers
1098 	const uint64_t inbuf_memusage = options->threads * block_size;
1099 
1100 	// Memory usage of the filter encoders
1101 	uint64_t filters_memusage = lzma_raw_encoder_memusage(filters);
1102 	if (filters_memusage == UINT64_MAX)
1103 		return UINT64_MAX;
1104 
1105 	filters_memusage *= options->threads;
1106 
1107 	// Memory usage of the output queue
1108 	const uint64_t outq_memusage = lzma_outq_memusage(
1109 			outbuf_size_max, options->threads);
1110 	if (outq_memusage == UINT64_MAX)
1111 		return UINT64_MAX;
1112 
1113 	// Sum them with overflow checking.
1114 	uint64_t total_memusage = LZMA_MEMUSAGE_BASE + sizeof(lzma_coder)
1115 			+ options->threads * sizeof(worker_thread);
1116 
1117 	if (UINT64_MAX - total_memusage < inbuf_memusage)
1118 		return UINT64_MAX;
1119 
1120 	total_memusage += inbuf_memusage;
1121 
1122 	if (UINT64_MAX - total_memusage < filters_memusage)
1123 		return UINT64_MAX;
1124 
1125 	total_memusage += filters_memusage;
1126 
1127 	if (UINT64_MAX - total_memusage < outq_memusage)
1128 		return UINT64_MAX;
1129 
1130 	return total_memusage + outq_memusage;
1131 }
1132