liblzma/common/stream_encoder_mt.c

52 	/// Input buffer of coder->block_size bytes. The main thread will
68 	lzma_stream_coder *coder;
207 	mythread_sync(thr->coder->mutex) {
208 		if (thr->coder->thread_error == LZMA_OK)
209 			thr->coder->thread_error = ret;
211 		mythread_cond_signal(&thr->coder->cond);
227 		.check = thr->coder->stream_flags.check,
229 		.uncompressed_size = thr->coder->block_size,
295 				thr->block_encoder.coder, thr->allocator,
408 		mythread_sync(thr->coder->mutex) {
417 			thr->coder->progress_in
419 			thr->coder->progress_out += out_pos;
424 			thr->next = thr->coder->threads_free;
425 			thr->coder->threads_free = thr;
427 			mythread_cond_signal(&thr->coder->cond);
445 threads_stop(lzma_stream_coder *coder, bool wait_for_threads)
448 	for (uint32_t i = 0; i < coder->threads_initialized; ++i) {
449 		mythread_sync(coder->threads[i].mutex) {
450 			coder->threads[i].state = THR_STOP;
451 			mythread_cond_signal(&coder->threads[i].cond);
459 	for (uint32_t i = 0; i < coder->threads_initialized; ++i) {
460 		mythread_sync(coder->threads[i].mutex) {
461 			while (coder->threads[i].state != THR_IDLE)
462 				mythread_cond_wait(&coder->threads[i].cond,
463 						&coder->threads[i].mutex);
474 threads_end(lzma_stream_coder *coder, const lzma_allocator *allocator)
476 	for (uint32_t i = 0; i < coder->threads_initialized; ++i) {
477 		mythread_sync(coder->threads[i].mutex) {
478 			coder->threads[i].state = THR_EXIT;
479 			mythread_cond_signal(&coder->threads[i].cond);
483 	for (uint32_t i = 0; i < coder->threads_initialized; ++i) {
484 		int ret = mythread_join(coder->threads[i].thread_id);
489 	lzma_free(coder->threads, allocator);
496 initialize_new_thread(lzma_stream_coder *coder,
499 	worker_thread *thr = &coder->threads[coder->threads_initialized];
501 	thr->in = lzma_alloc(coder->block_size, allocator);
513 	thr->coder = coder;
522 	++coder->threads_initialized;
523 	coder->thr = thr;
540 get_thread(lzma_stream_coder *coder, const lzma_allocator *allocator)
544 	if (!lzma_outq_has_buf(&coder->outq))
549 	return_if_error(lzma_outq_prealloc_buf(&coder->outq, allocator,
550 			coder->outbuf_alloc_size));
555 	if (coder->filters_cache[0].id == LZMA_VLI_UNKNOWN)
557 			coder->filters, coder->filters_cache, allocator));
560 	mythread_sync(coder->mutex) {
561 		if (coder->threads_free != NULL) {
562 			coder->thr = coder->threads_free;
563 			coder->threads_free = coder->threads_free->next;
567 	if (coder->thr == NULL) {
569 		if (coder->threads_initialized == coder->threads_max)
573 		return_if_error(initialize_new_thread(coder, allocator));
578 	mythread_sync(coder->thr->mutex) {
579 		coder->thr->state = THR_RUN;
580 		coder->thr->in_size = 0;
581 		coder->thr->outbuf = lzma_outq_get_buf(&coder->outq, NULL);
585 		// coder->filters_cache[]. Then mark the cache as empty.
586 		lzma_filters_free(coder->thr->filters, allocator);
587 		memcpy(coder->thr->filters, coder->filters_cache,
588 				sizeof(coder->filters_cache));
589 		coder->filters_cache[0].id = LZMA_VLI_UNKNOWN;
591 		mythread_cond_signal(&coder->thr->cond);
599 stream_encode_in(lzma_stream_coder *coder, const lzma_allocator *allocator,
604 			|| (coder->thr != NULL && action != LZMA_RUN)) {
605 		if (coder->thr == NULL) {
607 			const lzma_ret ret = get_thread(coder, allocator);
608 			if (coder->thr == NULL)
613 		size_t thr_in_size = coder->thr->in_size;
614 		lzma_bufcpy(in, in_pos, in_size, coder->thr->in,
615 				&thr_in_size, coder->block_size);
623 		const bool finish = thr_in_size == coder->block_size
628 		mythread_sync(coder->thr->mutex) {
629 			if (coder->thr->state == THR_IDLE) {
631 				// encoder. It has set coder->thread_error
637 				coder->thr->in_size = thr_in_size;
640 					coder->thr->state = THR_FINISH;
642 				mythread_cond_signal(&coder->thr->cond);
649 			mythread_sync(coder->mutex) {
650 				ret = coder->thread_error;
657 			coder->thr = NULL;
667 wait_for_work(lzma_stream_coder *coder, mythread_condtime *wait_abs,
670 	if (coder->timeout != 0 && !*has_blocked) {
684 		mythread_condtime_set(wait_abs, &coder->cond, coder->timeout);
689 	mythread_sync(coder->mutex) {
697 		while ((!has_input || coder->threads_free == NULL
698 					|| !lzma_outq_has_buf(&coder->outq))
699 				&& !lzma_outq_is_readable(&coder->outq)
700 				&& coder->thread_error == LZMA_OK
702 			if (coder->timeout != 0)
704 						&coder->cond, &coder->mutex,
707 				mythread_cond_wait(&coder->cond,
708 						&coder->mutex);
722 	lzma_stream_coder *coder = coder_ptr;
724 	switch (coder->sequence) {
726 		lzma_bufcpy(coder->header, &coder->header_pos,
727 				sizeof(coder->header),
729 		if (coder->header_pos < sizeof(coder->header))
732 		coder->header_pos = 0;
733 		coder->sequence = SEQ_BLOCK;
748 			mythread_sync(coder->mutex) {
750 				ret = coder->thread_error;
757 				ret = lzma_outq_read(&coder->outq, allocator,
765 				ret = lzma_index_append(coder->index,
769 					threads_stop(coder, false);
781 				// coder->thread_error was set.
782 				threads_stop(coder, false);
787 			ret = stream_encode_in(coder, allocator,
790 				threads_stop(coder, false);
813 				if (lzma_outq_is_empty(&coder->outq)) {
836 			if (wait_for_work(coder, &wait_abs, &has_blocked,
844 				&coder->index_encoder, allocator,
845 				coder->index));
846 		coder->sequence = SEQ_INDEX;
852 		coder->progress_out += lzma_index_size(coder->index)
861 		const lzma_ret ret = coder->index_encoder.code(
862 				coder->index_encoder.coder, allocator,
868 		// Encode the Stream Footer into coder->buffer.
869 		coder->stream_flags.backward_size
870 				= lzma_index_size(coder->index);
871 		if (lzma_stream_footer_encode(&coder->stream_flags,
872 				coder->header) != LZMA_OK)
875 		coder->sequence = SEQ_STREAM_FOOTER;
881 		lzma_bufcpy(coder->header, &coder->header_pos,
882 				sizeof(coder->header),
884 		return coder->header_pos < sizeof(coder->header)
896 	lzma_stream_coder *coder = coder_ptr;
899 	threads_end(coder, allocator);
900 	lzma_outq_end(&coder->outq, allocator);
902 	lzma_filters_free(coder->filters, allocator);
903 	lzma_filters_free(coder->filters_cache, allocator);
905 	lzma_next_end(&coder->index_encoder, allocator);
906 	lzma_index_end(coder->index, allocator);
908 	mythread_cond_destroy(&coder->cond);
909 	mythread_mutex_destroy(&coder->mutex);
911 	lzma_free(coder, allocator);
922 	lzma_stream_coder *coder = coder_ptr;
926 	if (coder->sequence > SEQ_BLOCK)
931 	if (coder->thr != NULL)
945 	lzma_filters_free(coder->filters, allocator);
946 	lzma_filters_free(coder->filters_cache, allocator);
949 	memcpy(coder->filters, temp, sizeof(temp));
1008 	lzma_stream_coder *coder = coder_ptr;
1010 	// Lock coder->mutex to prevent finishing threads from moving their
1012 	mythread_sync(coder->mutex) {
1013 		*progress_in = coder->progress_in;
1014 		*progress_out = coder->progress_out;
1016 		for (size_t i = 0; i < coder->threads_initialized; ++i) {
1017 			mythread_sync(coder->threads[i].mutex) {
1018 				*progress_in += coder->threads[i].progress_in;
1019 				*progress_out += coder->threads[i]
1065 	lzma_stream_coder *coder = next->coder;
1066 	if (coder == NULL) {
1067 		coder = lzma_alloc(sizeof(lzma_stream_coder), allocator);
1068 		if (coder == NULL)
1071 		next->coder = coder;
1077 		if (mythread_mutex_init(&coder->mutex)) {
1078 			lzma_free(coder, allocator);
1079 			next->coder = NULL;
1083 		if (mythread_cond_init(&coder->cond)) {
1084 			mythread_mutex_destroy(&coder->mutex);
1085 			lzma_free(coder, allocator);
1086 			next->coder = NULL;
1095 		coder->filters[0].id = LZMA_VLI_UNKNOWN;
1096 		coder->filters_cache[0].id = LZMA_VLI_UNKNOWN;
1097 		coder->index_encoder = LZMA_NEXT_CODER_INIT;
1098 		coder->index = NULL;
1099 		memzero(&coder->outq, sizeof(coder->outq));
1100 		coder->threads = NULL;
1101 		coder->threads_max = 0;
1102 		coder->threads_initialized = 0;
1106 	coder->sequence = SEQ_STREAM_HEADER;
1107 	coder->block_size = (size_t)(block_size);
1108 	coder->outbuf_alloc_size = (size_t)(outbuf_size_max);
1109 	coder->thread_error = LZMA_OK;
1110 	coder->thr = NULL;
1114 	if (coder->threads_max != options->threads) {
1115 		threads_end(coder, allocator);
1117 		coder->threads = NULL;
1118 		coder->threads_max = 0;
1120 		coder->threads_initialized = 0;
1121 		coder->threads_free = NULL;
1123 		coder->threads = lzma_alloc(
1126 		if (coder->threads == NULL)
1129 		coder->threads_max = options->threads;
1133 		threads_stop(coder, true);
1137 	return_if_error(lzma_outq_init(&coder->outq, allocator,
1141 	coder->timeout = options->timeout;
1144 	lzma_filters_free(coder->filters, allocator);
1145 	lzma_filters_free(coder->filters_cache, allocator);
1149 			filters, coder->filters, allocator));
1152 	lzma_index_end(coder->index, allocator);
1153 	coder->index = lzma_index_init(allocator);
1154 	if (coder->index == NULL)
1158 	coder->stream_flags.version = 0;
1159 	coder->stream_flags.check = options->check;
1161 			&coder->stream_flags, coder->header));
1163 	coder->header_pos = 0;
1166 	coder->progress_in = 0;
1167 	coder->progress_out = LZMA_STREAM_HEADER_SIZE;