1 ///////////////////////////////////////////////////////////////////////////////
2 //
3 /// \file outqueue.c
4 /// \brief Output queue handling in multithreaded coding
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 "outqueue.h"
14
15
16 /// Get the maximum number of buffers that may be allocated based
17 /// on the number of threads. For now this is twice the number of threads.
18 /// It's a compromise between RAM usage and keeping the worker threads busy
19 /// when buffers finish out of order.
20 #define GET_BUFS_LIMIT(threads) (2 * (threads))
21
22
23 extern uint64_t
24 lzma_outq_memusage(uint64_t buf_size_max, uint32_t threads)
25 {
26 // This is to ease integer overflow checking: We may allocate up to
27 // GET_BUFS_LIMIT(LZMA_THREADS_MAX) buffers and we need some extra
28 // memory for other data structures too (that's the /2).
29 //
30 // lzma_outq_prealloc_buf() will still accept bigger buffers than this.
31 const uint64_t limit
32 = UINT64_MAX / GET_BUFS_LIMIT(LZMA_THREADS_MAX) / 2;
33
34 if (threads > LZMA_THREADS_MAX || buf_size_max > limit)
35 return UINT64_MAX;
36
37 return GET_BUFS_LIMIT(threads)
38 * lzma_outq_outbuf_memusage(buf_size_max);
39 }
40
41
42 static void
43 move_head_to_cache(lzma_outq *outq, const lzma_allocator *allocator)
44 {
45 assert(outq->head != NULL);
46 assert(outq->tail != NULL);
47 assert(outq->bufs_in_use > 0);
48
49 lzma_outbuf *buf = outq->head;
50 outq->head = buf->next;
51 if (outq->head == NULL)
52 outq->tail = NULL;
53
54 if (outq->cache != NULL && outq->cache->allocated != buf->allocated)
55 lzma_outq_clear_cache(outq, allocator);
56
57 buf->next = outq->cache;
58 outq->cache = buf;
59
60 --outq->bufs_in_use;
61 outq->mem_in_use -= lzma_outq_outbuf_memusage(buf->allocated);
62
63 return;
64 }
65
66
67 static void
68 free_one_cached_buffer(lzma_outq *outq, const lzma_allocator *allocator)
69 {
70 assert(outq->cache != NULL);
71
72 lzma_outbuf *buf = outq->cache;
73 outq->cache = buf->next;
74
75 --outq->bufs_allocated;
76 outq->mem_allocated -= lzma_outq_outbuf_memusage(buf->allocated);
77
78 lzma_free(buf, allocator);
79 return;
80 }
81
82
83 extern void
84 lzma_outq_clear_cache(lzma_outq *outq, const lzma_allocator *allocator)
85 {
86 while (outq->cache != NULL)
87 free_one_cached_buffer(outq, allocator);
88
89 return;
90 }
91
92
93 extern void
94 lzma_outq_clear_cache2(lzma_outq *outq, const lzma_allocator *allocator,
95 size_t keep_size)
96 {
97 if (outq->cache == NULL)
98 return;
99
100 // Free all but one.
101 while (outq->cache->next != NULL)
102 free_one_cached_buffer(outq, allocator);
103
104 // Free the last one only if its size doesn't equal to keep_size.
105 if (outq->cache->allocated != keep_size)
106 free_one_cached_buffer(outq, allocator);
107
108 return;
109 }
110
111
112 extern lzma_ret
113 lzma_outq_init(lzma_outq *outq, const lzma_allocator *allocator,
114 uint32_t threads)
115 {
116 if (threads > LZMA_THREADS_MAX)
117 return LZMA_OPTIONS_ERROR;
118
119 const uint32_t bufs_limit = GET_BUFS_LIMIT(threads);
120
121 // Clear head/tail.
122 while (outq->head != NULL)
123 move_head_to_cache(outq, allocator);
124
125 // If new buf_limit is lower than the old one, we may need to free
126 // a few cached buffers.
127 while (bufs_limit < outq->bufs_allocated)
128 free_one_cached_buffer(outq, allocator);
129
130 outq->bufs_limit = bufs_limit;
131 outq->read_pos = 0;
132
133 return LZMA_OK;
134 }
135
136
137 extern void
138 lzma_outq_end(lzma_outq *outq, const lzma_allocator *allocator)
139 {
140 while (outq->head != NULL)
141 move_head_to_cache(outq, allocator);
142
143 lzma_outq_clear_cache(outq, allocator);
144 return;
145 }
146
147
148 extern lzma_ret
149 lzma_outq_prealloc_buf(lzma_outq *outq, const lzma_allocator *allocator,
150 size_t size)
151 {
152 // Caller must have checked it with lzma_outq_has_buf().
153 assert(outq->bufs_in_use < outq->bufs_limit);
154
155 // If there already is appropriately-sized buffer in the cache,
156 // we need to do nothing.
157 if (outq->cache != NULL && outq->cache->allocated == size)
158 return LZMA_OK;
159
160 if (size > SIZE_MAX - sizeof(lzma_outbuf))
161 return LZMA_MEM_ERROR;
162
163 const size_t alloc_size = lzma_outq_outbuf_memusage(size);
164
165 // The cache may have buffers but their size is wrong.
166 lzma_outq_clear_cache(outq, allocator);
167
168 outq->cache = lzma_alloc(alloc_size, allocator);
169 if (outq->cache == NULL)
170 return LZMA_MEM_ERROR;
171
172 outq->cache->next = NULL;
173 outq->cache->allocated = size;
174
175 ++outq->bufs_allocated;
176 outq->mem_allocated += alloc_size;
177
178 return LZMA_OK;
179 }
180
181
182 extern lzma_outbuf *
183 lzma_outq_get_buf(lzma_outq *outq, void *worker)
184 {
185 // Caller must have used lzma_outq_prealloc_buf() to ensure these.
186 assert(outq->bufs_in_use < outq->bufs_limit);
187 assert(outq->bufs_in_use < outq->bufs_allocated);
188 assert(outq->cache != NULL);
189
190 lzma_outbuf *buf = outq->cache;
191 outq->cache = buf->next;
192 buf->next = NULL;
193
194 if (outq->tail != NULL) {
195 assert(outq->head != NULL);
196 outq->tail->next = buf;
197 } else {
198 assert(outq->head == NULL);
199 outq->head = buf;
200 }
201
202 outq->tail = buf;
203
204 buf->worker = worker;
205 buf->finished = false;
206 buf->finish_ret = LZMA_STREAM_END;
207 buf->pos = 0;
208 buf->decoder_in_pos = 0;
209
210 buf->unpadded_size = 0;
211 buf->uncompressed_size = 0;
212
213 ++outq->bufs_in_use;
214 outq->mem_in_use += lzma_outq_outbuf_memusage(buf->allocated);
215
216 return buf;
217 }
218
219
220 extern bool
221 lzma_outq_is_readable(const lzma_outq *outq)
222 {
223 if (outq->head == NULL)
224 return false;
225
226 return outq->read_pos < outq->head->pos || outq->head->finished;
227 }
228
229
230 extern lzma_ret
231 lzma_outq_read(lzma_outq *restrict outq,
232 const lzma_allocator *restrict allocator,
233 uint8_t *restrict out, size_t *restrict out_pos,
234 size_t out_size,
235 lzma_vli *restrict unpadded_size,
236 lzma_vli *restrict uncompressed_size)
237 {
238 // There must be at least one buffer from which to read.
239 if (outq->bufs_in_use == 0)
240 return LZMA_OK;
241
242 // Get the buffer.
243 lzma_outbuf *buf = outq->head;
244
245 // Copy from the buffer to output.
246 //
247 // FIXME? In threaded decoder it may be bad to do this copy while
248 // the mutex is being held.
249 lzma_bufcpy(buf->buf, &outq->read_pos, buf->pos,
250 out, out_pos, out_size);
251
252 // Return if we didn't get all the data from the buffer.
253 if (!buf->finished || outq->read_pos < buf->pos)
254 return LZMA_OK;
255
256 // The buffer was finished. Tell the caller its size information.
257 if (unpadded_size != NULL)
258 *unpadded_size = buf->unpadded_size;
259
260 if (uncompressed_size != NULL)
261 *uncompressed_size = buf->uncompressed_size;
262
263 // Remember the return value.
264 const lzma_ret finish_ret = buf->finish_ret;
265
266 // Free this buffer for further use.
267 move_head_to_cache(outq, allocator);
268 outq->read_pos = 0;
269
270 return finish_ret;
271 }
272
273
274 extern void
275 lzma_outq_enable_partial_output(lzma_outq *outq,
276 void (*enable_partial_output)(void *worker))
277 {
278 if (outq->head != NULL && !outq->head->finished
279 && outq->head->worker != NULL) {
280 enable_partial_output(outq->head->worker);
281
282 // Set it to NULL since calling it twice is pointless.
283 outq->head->worker = NULL;
284 }
285
286 return;
287 }