+ return msgs;
+}
+
+static void
+main_writer_thread_destroy_ctx(struct main_writer_thread_ctx *ctx)
+{
+ while (ctx->num_outstanding_messages--)
+ shared_queue_get(ctx->compressed_res_queue);
+ free_messages(ctx->msgs, ctx->num_messages);
+ FREE(ctx->cur_chunk_tab);
+}
+
+static int
+main_writer_thread_init_ctx(struct main_writer_thread_ctx *ctx)
+{
+ /* Pre-allocate all the buffers that will be needed to do the chunk
+ * compression. */
+ ctx->msgs = allocate_messages(ctx->num_messages);
+ if (!ctx->msgs)
+ return WIMLIB_ERR_NOMEM;
+
+ /* Initially, all the messages are available to use. */
+ INIT_LIST_HEAD(&ctx->available_msgs);
+ for (size_t i = 0; i < ctx->num_messages; i++)
+ list_add_tail(&ctx->msgs[i].list, &ctx->available_msgs);
+
+ /* outstanding_streams is the list of streams that currently have had
+ * chunks sent off for compression.
+ *
+ * The first stream in outstanding_streams is the stream that is
+ * currently being written.
+ *
+ * The last stream in outstanding_streams is the stream that is
+ * currently being read and having chunks fed to the compressor threads.
+ * */
+ INIT_LIST_HEAD(&ctx->outstanding_streams);
+ ctx->num_outstanding_messages = 0;
+
+ ctx->next_msg = NULL;
+
+ /* Resources that don't need any chunks compressed are added to this
+ * list and written directly by the main thread. */
+ INIT_LIST_HEAD(&ctx->serial_streams);
+
+ ctx->cur_chunk_tab = NULL;
+
+ return 0;
+}
+
+static int
+receive_compressed_chunks(struct main_writer_thread_ctx *ctx)
+{
+ struct message *msg;
+ struct wim_lookup_table_entry *cur_lte;
+ int ret;
+
+ wimlib_assert(!list_empty(&ctx->outstanding_streams));
+ wimlib_assert(ctx->num_outstanding_messages != 0);
+
+ cur_lte = container_of(ctx->outstanding_streams.next,
+ struct wim_lookup_table_entry,
+ being_compressed_list);
+
+ /* Get the next message from the queue and process it.
+ * The message will contain 1 or more data chunks that have been
+ * compressed. */
+ msg = shared_queue_get(ctx->compressed_res_queue);
+ msg->complete = true;
+ --ctx->num_outstanding_messages;
+
+ /* Is this the next chunk in the current resource? If it's not
+ * (i.e., an earlier chunk in a same or different resource
+ * hasn't been compressed yet), do nothing, and keep this
+ * message around until all earlier chunks are received.
+ *
+ * Otherwise, write all the chunks we can. */
+ while (cur_lte != NULL &&
+ !list_empty(&cur_lte->msg_list)
+ && (msg = container_of(cur_lte->msg_list.next,
+ struct message,
+ list))->complete)
+ {
+ list_move(&msg->list, &ctx->available_msgs);
+ if (msg->begin_chunk == 0) {
+ /* This is the first set of chunks. Leave space
+ * for the chunk table in the output file. */
+ off_t cur_offset = filedes_offset(ctx->out_fd);
+ if (cur_offset == -1)
+ return WIMLIB_ERR_WRITE;
+ ret = begin_wim_resource_chunk_tab(cur_lte,
+ ctx->out_fd,
+ cur_offset,
+ &ctx->cur_chunk_tab);
+ if (ret)
+ return ret;
+ }
+
+ /* Write the compressed chunks from the message. */
+ ret = write_wim_chunks(msg, ctx->out_fd, ctx->cur_chunk_tab);
+ if (ret)
+ return ret;
+
+ /* Was this the last chunk of the stream? If so, finish
+ * it. */
+ if (list_empty(&cur_lte->msg_list) &&
+ msg->begin_chunk + msg->num_chunks == ctx->cur_chunk_tab->num_chunks)
+ {
+ u64 res_csize;
+ off_t offset;
+
+ ret = finish_wim_resource_chunk_tab(ctx->cur_chunk_tab,
+ ctx->out_fd,
+ &res_csize);
+ if (ret)
+ return ret;
+
+ list_del(&cur_lte->being_compressed_list);
+
+ /* Grab the offset of this stream in the output file
+ * from the chunk table before we free it. */
+ offset = ctx->cur_chunk_tab->file_offset;
+
+ FREE(ctx->cur_chunk_tab);
+ ctx->cur_chunk_tab = NULL;
+
+ if (res_csize >= wim_resource_size(cur_lte)) {
+ /* Oops! We compressed the resource to
+ * larger than the original size. Write
+ * the resource uncompressed instead. */
+ DEBUG("Compressed %"PRIu64" => %"PRIu64" bytes; "
+ "writing uncompressed instead",
+ wim_resource_size(cur_lte), res_csize);
+ ret = seek_and_truncate(ctx->out_fd, offset);
+ if (ret)
+ return ret;
+ ret = write_wim_resource(cur_lte,
+ ctx->out_fd,
+ WIMLIB_COMPRESSION_TYPE_NONE,
+ &cur_lte->output_resource_entry,
+ ctx->write_resource_flags);
+ if (ret)
+ return ret;
+ } else {
+ cur_lte->output_resource_entry.size =
+ res_csize;
+
+ cur_lte->output_resource_entry.original_size =
+ cur_lte->resource_entry.original_size;
+
+ cur_lte->output_resource_entry.offset =
+ offset;
+
+ cur_lte->output_resource_entry.flags =
+ cur_lte->resource_entry.flags |
+ WIM_RESHDR_FLAG_COMPRESSED;
+ }
+
+ do_write_streams_progress(ctx->progress,
+ ctx->progress_func,
+ wim_resource_size(cur_lte),
+ false);
+
+ /* Since we just finished writing a stream, write any
+ * streams that have been added to the serial_streams
+ * list for direct writing by the main thread (e.g.
+ * resources that don't need to be compressed because
+ * the desired compression type is the same as the
+ * previous compression type). */
+ if (!list_empty(&ctx->serial_streams)) {
+ ret = do_write_stream_list_serial(&ctx->serial_streams,
+ ctx->lookup_table,
+ ctx->out_fd,
+ ctx->out_ctype,
+ ctx->write_resource_flags,
+ ctx->progress_func,
+ ctx->progress);
+ if (ret)
+ return ret;
+ }
+
+ /* Advance to the next stream to write. */
+ if (list_empty(&ctx->outstanding_streams)) {
+ cur_lte = NULL;
+ } else {
+ cur_lte = container_of(ctx->outstanding_streams.next,
+ struct wim_lookup_table_entry,
+ being_compressed_list);