brotli/enc/threading/
mod.rs

1use alloc::{Allocator, SliceWrapper, SliceWrapperMut};
2#[cfg(feature = "std")]
3use core::any;
4use core::marker::PhantomData;
5use core::mem;
6use core::ops::Range;
7#[cfg(feature = "std")]
8use std;
9
10use super::backward_references::{AnyHasher, BrotliEncoderParams, CloneWithAlloc, UnionHasher};
11use super::encode::{
12    hasher_setup, BrotliEncoderDestroyInstance, BrotliEncoderMaxCompressedSize,
13    BrotliEncoderOperation, SanitizeParams,
14};
15use super::BrotliAlloc;
16use crate::concat::{BroCatli, BroCatliResult};
17use crate::enc::combined_alloc::{alloc_default, allocate};
18use crate::enc::encode::BrotliEncoderStateStruct;
19
20pub type PoisonedThreadError = ();
21
22#[cfg(feature = "std")]
23pub type LowLevelThreadError = std::boxed::Box<dyn any::Any + Send + 'static>;
24#[cfg(not(feature = "std"))]
25pub type LowLevelThreadError = ();
26
27pub trait AnyBoxConstructor {
28    fn new(data: LowLevelThreadError) -> Self;
29}
30
31pub trait Joinable<T: Send + 'static, U: Send + 'static>: Sized {
32    fn join(self) -> Result<T, U>;
33}
34#[derive(Debug)]
35pub enum BrotliEncoderThreadError {
36    InsufficientOutputSpace,
37    ConcatenationDidNotProcessFullFile,
38    ConcatenationError(BroCatliResult),
39    ConcatenationFinalizationError(BroCatliResult),
40    OtherThreadPanic,
41    ThreadExecError(LowLevelThreadError),
42}
43
44impl AnyBoxConstructor for BrotliEncoderThreadError {
45    fn new(data: LowLevelThreadError) -> Self {
46        BrotliEncoderThreadError::ThreadExecError(data)
47    }
48}
49
50fn set_pending_error(
51    pending_error: &mut Option<BrotliEncoderThreadError>,
52    error: BrotliEncoderThreadError,
53) {
54    if pending_error.is_none() {
55        *pending_error = Some(error);
56    }
57}
58
59pub struct CompressedFileChunk<Alloc: BrotliAlloc + Send + 'static>
60where
61    <Alloc as Allocator<u8>>::AllocatedMemory: Send,
62{
63    data_backing: <Alloc as Allocator<u8>>::AllocatedMemory,
64    data_size: usize,
65}
66pub struct CompressionThreadResult<Alloc: BrotliAlloc + Send + 'static>
67where
68    <Alloc as Allocator<u8>>::AllocatedMemory: Send,
69{
70    compressed: Result<CompressedFileChunk<Alloc>, BrotliEncoderThreadError>,
71    alloc: Alloc,
72}
73pub enum InternalSendAlloc<
74    ReturnVal: Send + 'static,
75    ExtraInput: Send + 'static,
76    Alloc: BrotliAlloc + Send + 'static,
77    Join: Joinable<ReturnVal, BrotliEncoderThreadError>,
78> where
79    <Alloc as Allocator<u8>>::AllocatedMemory: Send,
80{
81    A(Alloc, ExtraInput),
82    Join(Join),
83    SpawningOrJoining(PhantomData<ReturnVal>),
84}
85impl<
86        ReturnVal: Send + 'static,
87        ExtraInput: Send + 'static,
88        Alloc: BrotliAlloc + Send + 'static,
89        Join: Joinable<ReturnVal, BrotliEncoderThreadError>,
90    > InternalSendAlloc<ReturnVal, ExtraInput, Alloc, Join>
91where
92    <Alloc as Allocator<u8>>::AllocatedMemory: Send,
93{
94    fn unwrap_input(&mut self) -> (&mut Alloc, &mut ExtraInput) {
95        match *self {
96            InternalSendAlloc::A(ref mut alloc, ref mut extra) => (alloc, extra),
97            _ => panic!("Bad state for allocator"),
98        }
99    }
100}
101
102pub struct SendAlloc<
103    ReturnValue: Send + 'static,
104    ExtraInput: Send + 'static,
105    Alloc: BrotliAlloc + Send + 'static,
106    Join: Joinable<ReturnValue, BrotliEncoderThreadError>,
107>(pub InternalSendAlloc<ReturnValue, ExtraInput, Alloc, Join>)
108//FIXME pub
109where
110    <Alloc as Allocator<u8>>::AllocatedMemory: Send;
111
112impl<
113        ReturnValue: Send + 'static,
114        ExtraInput: Send + 'static,
115        Alloc: BrotliAlloc + Send + 'static,
116        Join: Joinable<ReturnValue, BrotliEncoderThreadError>,
117    > SendAlloc<ReturnValue, ExtraInput, Alloc, Join>
118where
119    <Alloc as Allocator<u8>>::AllocatedMemory: Send,
120{
121    pub fn new(alloc: Alloc, extra_input: ExtraInput) -> Self {
122        SendAlloc::<ReturnValue, ExtraInput, Alloc, Join>(InternalSendAlloc::A(alloc, extra_input))
123    }
124    pub fn unwrap_or(self, other: Alloc, other_extra: ExtraInput) -> (Alloc, ExtraInput) {
125        match self.0 {
126            InternalSendAlloc::A(alloc, extra_input) => (alloc, extra_input),
127            InternalSendAlloc::SpawningOrJoining(_) | InternalSendAlloc::Join(_) => {
128                (other, other_extra)
129            }
130        }
131    }
132    fn unwrap_view_mut(&mut self) -> (&mut Alloc, &mut ExtraInput) {
133        match self.0 {
134            InternalSendAlloc::A(ref mut alloc, ref mut extra_input) => (alloc, extra_input),
135            InternalSendAlloc::SpawningOrJoining(_) | InternalSendAlloc::Join(_) => {
136                panic!("Item permanently borrowed/leaked")
137            }
138        }
139    }
140    pub fn unwrap(self) -> (Alloc, ExtraInput) {
141        match self.0 {
142            InternalSendAlloc::A(alloc, extra_input) => (alloc, extra_input),
143            InternalSendAlloc::SpawningOrJoining(_) | InternalSendAlloc::Join(_) => {
144                panic!("Item permanently borrowed/leaked")
145            }
146        }
147    }
148    pub fn replace_with_default(&mut self) -> (Alloc, ExtraInput) {
149        match mem::replace(
150            &mut self.0,
151            InternalSendAlloc::SpawningOrJoining(PhantomData),
152        ) {
153            InternalSendAlloc::A(alloc, extra_input) => (alloc, extra_input),
154            InternalSendAlloc::SpawningOrJoining(_) | InternalSendAlloc::Join(_) => {
155                panic!("Item permanently borrowed/leaked")
156            }
157        }
158    }
159}
160
161pub enum InternalOwned<T> {
162    // FIXME pub
163    Item(T),
164    Borrowed,
165}
166
167pub struct Owned<T>(pub InternalOwned<T>); // FIXME pub
168impl<T> Owned<T> {
169    pub fn new(data: T) -> Self {
170        Owned::<T>(InternalOwned::Item(data))
171    }
172    pub fn unwrap_or(self, other: T) -> T {
173        if let InternalOwned::Item(x) = self.0 {
174            x
175        } else {
176            other
177        }
178    }
179    pub fn unwrap(self) -> T {
180        if let InternalOwned::Item(x) = self.0 {
181            x
182        } else {
183            panic!("Item permanently borrowed")
184        }
185    }
186    pub fn view(&self) -> &T {
187        if let InternalOwned::Item(ref x) = self.0 {
188            x
189        } else {
190            panic!("Item permanently borrowed")
191        }
192    }
193}
194
195pub trait OwnedRetriever<U: Send + 'static> {
196    fn view<T, F: FnOnce(&U) -> T>(&self, f: F) -> Result<T, PoisonedThreadError>;
197    fn unwrap(self) -> Result<U, PoisonedThreadError>;
198}
199
200#[cfg(feature = "std")]
201impl<U: Send + 'static> OwnedRetriever<U> for std::sync::Arc<std::sync::RwLock<U>> {
202    fn view<T, F: FnOnce(&U) -> T>(&self, f: F) -> Result<T, PoisonedThreadError> {
203        match self.read() {
204            Ok(ref u) => Ok(f(u)),
205            Err(_) => Err(PoisonedThreadError::default()),
206        }
207    }
208    fn unwrap(self) -> Result<U, PoisonedThreadError> {
209        match std::sync::Arc::try_unwrap(self) {
210            Ok(rwlock) => match rwlock.into_inner() {
211                Ok(u) => Ok(u),
212                Err(_) => Err(PoisonedThreadError::default()),
213            },
214            Err(_) => Err(PoisonedThreadError::default()),
215        }
216    }
217}
218
219pub trait BatchSpawnable<
220    ReturnValue: Send + 'static,
221    ExtraInput: Send + 'static,
222    Alloc: BrotliAlloc + Send + 'static,
223    U: Send + 'static + Sync,
224> where
225    <Alloc as Allocator<u8>>::AllocatedMemory: Send + 'static,
226{
227    type JoinHandle: Joinable<ReturnValue, BrotliEncoderThreadError>;
228    type FinalJoinHandle: OwnedRetriever<U>;
229    // this function takes in an input slice
230    // a SendAlloc per thread and converts them all into JoinHandle
231    // the input is borrowed until the joins complete
232    // owned is set to borrowed
233    // the final join handle is a r/w lock which will return the SliceW to the owner
234    // the FinalJoinHandle is only to be called when each individual JoinHandle has been examined
235    // the function is called with the thread_index, the num_threads, a reference to the slice under a read lock,
236    // and an allocator from the alloc_per_thread
237    fn make_spawner(&mut self, input: &mut Owned<U>) -> Self::FinalJoinHandle;
238    fn spawn<F: Fn(ExtraInput, usize, usize, &U, Alloc) -> ReturnValue + Send + 'static + Copy>(
239        &mut self,
240        handle: &mut Self::FinalJoinHandle,
241        alloc: &mut SendAlloc<ReturnValue, ExtraInput, Alloc, Self::JoinHandle>,
242        index: usize,
243        num_threads: usize,
244        f: F,
245    );
246}
247
248pub trait BatchSpawnableLite<
249    ReturnValue: Send + 'static,
250    ExtraInput: Send + 'static,
251    Alloc: BrotliAlloc + Send + 'static,
252    U: Send + 'static + Sync,
253> where
254    <Alloc as Allocator<u8>>::AllocatedMemory: Send + 'static,
255{
256    type JoinHandle: Joinable<ReturnValue, BrotliEncoderThreadError>;
257    type FinalJoinHandle: OwnedRetriever<U>;
258    fn make_spawner(&mut self, input: &mut Owned<U>) -> Self::FinalJoinHandle;
259    fn spawn(
260        &mut self,
261        handle: &mut Self::FinalJoinHandle,
262        alloc_per_thread: &mut SendAlloc<ReturnValue, ExtraInput, Alloc, Self::JoinHandle>,
263        index: usize,
264        num_threads: usize,
265        f: fn(ExtraInput, usize, usize, &U, Alloc) -> ReturnValue,
266    );
267}
268/*
269impl<ReturnValue:Send+'static,
270     ExtraInput:Send+'static,
271     Alloc:BrotliAlloc+Send+'static,
272     U:Send+'static+Sync>
273     BatchSpawnableLite<T, Alloc, U> for BatchSpawnable<T, Alloc, U> {
274  type JoinHandle = <Self as BatchSpawnable<T, Alloc, U>>::JoinHandle;
275  type FinalJoinHandle = <Self as BatchSpawnable<T, Alloc, U>>::FinalJoinHandle;
276  fn batch_spawn(
277    &mut self,
278    input: &mut Owned<U>,
279    alloc_per_thread:&mut [SendAlloc<ReturnValue, ExtraInput, Alloc, Self::JoinHandle>],
280    f: fn(usize, usize, &U, Alloc) -> T,
281  ) -> Self::FinalJoinHandle {
282   <Self as BatchSpawnable<ReturnValue, ExtraInput,  Alloc, U>>::batch_spawn(self, input, alloc_per_thread, f)
283  }
284}*/
285
286pub fn CompressMultiSlice<
287    Alloc: BrotliAlloc + Send + 'static,
288    Spawner: BatchSpawnableLite<
289        CompressionThreadResult<Alloc>,
290        UnionHasher<Alloc>,
291        Alloc,
292        (
293            <Alloc as Allocator<u8>>::AllocatedMemory,
294            BrotliEncoderParams,
295        ),
296    >,
297>(
298    params: &BrotliEncoderParams,
299    input_slice: &[u8],
300    output: &mut [u8],
301    alloc_per_thread: &mut [SendAlloc<
302        CompressionThreadResult<Alloc>,
303        UnionHasher<Alloc>,
304        Alloc,
305        Spawner::JoinHandle,
306    >],
307    thread_spawner: &mut Spawner,
308) -> Result<usize, BrotliEncoderThreadError>
309where
310    <Alloc as Allocator<u8>>::AllocatedMemory: Send + Sync,
311    <Alloc as Allocator<u16>>::AllocatedMemory: Send + Sync,
312    <Alloc as Allocator<u32>>::AllocatedMemory: Send + Sync,
313{
314    let input = if let InternalSendAlloc::A(ref mut alloc, ref _extra) = alloc_per_thread[0].0 {
315        let mut input = allocate::<u8, _>(alloc, input_slice.len());
316        input.slice_mut().clone_from_slice(input_slice);
317        input
318    } else {
319        alloc_default::<u8, Alloc>()
320    };
321    let mut owned_input = Owned::new(input);
322    let ret = CompressMulti(
323        params,
324        &mut owned_input,
325        output,
326        alloc_per_thread,
327        thread_spawner,
328    );
329    if let InternalSendAlloc::A(ref mut alloc, ref _extra) = alloc_per_thread[0].0 {
330        <Alloc as Allocator<u8>>::free_cell(alloc, owned_input.unwrap());
331    }
332    ret
333}
334
335fn get_range(thread_index: usize, num_threads: usize, file_size: usize) -> Range<usize> {
336    ((thread_index * file_size) / num_threads)..(((thread_index + 1) * file_size) / num_threads)
337}
338
339fn compress_part<Alloc: BrotliAlloc + Send + 'static, SliceW: SliceWrapper<u8>>(
340    hasher: UnionHasher<Alloc>,
341    thread_index: usize,
342    num_threads: usize,
343    input_and_params: &(SliceW, BrotliEncoderParams),
344    mut alloc: Alloc,
345) -> CompressionThreadResult<Alloc>
346where
347    <Alloc as Allocator<u8>>::AllocatedMemory: Send + 'static,
348{
349    let mut range = get_range(thread_index, num_threads, input_and_params.0.len());
350    let mut mem = allocate::<u8, _>(
351        &mut alloc,
352        BrotliEncoderMaxCompressedSize(range.end - range.start),
353    );
354    let mut state = BrotliEncoderStateStruct::new(alloc);
355    state.params = input_and_params.1.clone();
356    if thread_index != 0 {
357        state.params.catable = true; // make sure we can concatenate this to the other work results
358        state.params.magic_number = false; // no reason to pepper this around
359    }
360    state.params.appendable = true; // make sure we are at least appendable, so that future items can be catted in
361    if thread_index != 0 {
362        state.set_custom_dictionary_with_optional_precomputed_hasher(
363            range.start,
364            &input_and_params.0.slice()[..range.start],
365            hasher,
366            true,
367        );
368    }
369    let mut out_offset = 0usize;
370    let compression_result;
371    let mut available_out = mem.len();
372    loop {
373        let mut next_in_offset = 0usize;
374        let mut available_in = range.end - range.start;
375        let result = state.compress_stream(
376            BrotliEncoderOperation::BROTLI_OPERATION_FINISH,
377            &mut available_in,
378            &input_and_params.0.slice()[range.clone()],
379            &mut next_in_offset,
380            &mut available_out,
381            mem.slice_mut(),
382            &mut out_offset,
383            &mut None,
384            &mut |_a, _b, _c, _d| (),
385        );
386        let new_range = range.start + next_in_offset..range.end;
387        range = new_range;
388        if result {
389            compression_result = Ok(out_offset);
390            break;
391        } else if available_out == 0 {
392            compression_result = Err(BrotliEncoderThreadError::InsufficientOutputSpace); // mark no space??
393            break;
394        }
395    }
396    BrotliEncoderDestroyInstance(&mut state);
397    match compression_result {
398        Ok(size) => CompressionThreadResult::<Alloc> {
399            compressed: Ok(CompressedFileChunk {
400                data_backing: mem,
401                data_size: size,
402            }),
403            alloc: state.m8,
404        },
405        Err(e) => {
406            <Alloc as Allocator<u8>>::free_cell(&mut state.m8, mem);
407            CompressionThreadResult::<Alloc> {
408                compressed: Err(e),
409                alloc: state.m8,
410            }
411        }
412    }
413}
414
415pub fn CompressMulti<
416    Alloc: BrotliAlloc + Send + 'static,
417    SliceW: SliceWrapper<u8> + Send + 'static + Sync,
418    Spawner: BatchSpawnableLite<
419        CompressionThreadResult<Alloc>,
420        UnionHasher<Alloc>,
421        Alloc,
422        (SliceW, BrotliEncoderParams),
423    >,
424>(
425    params: &BrotliEncoderParams,
426    owned_input: &mut Owned<SliceW>,
427    output: &mut [u8],
428    alloc_per_thread: &mut [SendAlloc<
429        CompressionThreadResult<Alloc>,
430        UnionHasher<Alloc>,
431        Alloc,
432        Spawner::JoinHandle,
433    >],
434    thread_spawner: &mut Spawner,
435) -> Result<usize, BrotliEncoderThreadError>
436where
437    <Alloc as Allocator<u8>>::AllocatedMemory: Send,
438    <Alloc as Allocator<u16>>::AllocatedMemory: Send,
439    <Alloc as Allocator<u32>>::AllocatedMemory: Send,
440{
441    let num_threads = alloc_per_thread.len();
442    let actually_owned_mem = mem::replace(owned_input, Owned(InternalOwned::Borrowed));
443    let mut owned_input_pair = Owned::new((actually_owned_mem.unwrap(), params.clone()));
444    // start thread spawner
445    let mut spawner_and_input = thread_spawner.make_spawner(&mut owned_input_pair);
446    if num_threads > 1 {
447        // spawn first thread without "custom dictionary" while we compute the custom dictionary for other work items
448        thread_spawner.spawn(
449            &mut spawner_and_input,
450            &mut alloc_per_thread[0],
451            0,
452            num_threads,
453            compress_part,
454        );
455    }
456    // populate all hashers at once, cloning them one by one
457    let mut compression_last_thread_result;
458    if num_threads > 1 && params.favor_cpu_efficiency {
459        let mut local_params = params.clone();
460        SanitizeParams(&mut local_params);
461        let mut hasher = UnionHasher::Uninit;
462        hasher_setup(
463            alloc_per_thread[num_threads - 1].0.unwrap_input().0,
464            &mut hasher,
465            &mut local_params,
466            None, // No unwrappable custom dict used here.
467            &[],
468            0,
469            0,
470            false,
471        );
472        let mut setup_error = false;
473        for thread_index in 1..num_threads {
474            let res = spawner_and_input.view(|input_and_params: &(SliceW, BrotliEncoderParams)| {
475                let range = get_range(thread_index - 1, num_threads, input_and_params.0.len());
476                let overlap = hasher.StoreLookahead().wrapping_sub(1);
477                if range.end - range.start > overlap {
478                    hasher.BulkStoreRange(
479                        input_and_params.0.slice(),
480                        usize::MAX,
481                        if range.start > overlap {
482                            range.start - overlap
483                        } else {
484                            0
485                        },
486                        range.end - overlap,
487                    );
488                }
489            });
490            if let Err(_e) = res {
491                setup_error = true;
492                break;
493            }
494            if thread_index + 1 != num_threads {
495                {
496                    let (alloc, out_hasher) = alloc_per_thread[thread_index].unwrap_view_mut();
497                    *out_hasher = hasher.clone_with_alloc(alloc);
498                }
499                thread_spawner.spawn(
500                    &mut spawner_and_input,
501                    &mut alloc_per_thread[thread_index],
502                    thread_index,
503                    num_threads,
504                    compress_part,
505                );
506            }
507        }
508        if setup_error {
509            let mut setup_result = Err(BrotliEncoderThreadError::OtherThreadPanic);
510            for thread in alloc_per_thread.iter_mut() {
511                match mem::replace(
512                    &mut thread.0,
513                    InternalSendAlloc::SpawningOrJoining(PhantomData),
514                ) {
515                    InternalSendAlloc::Join(join) => match join.join() {
516                        Ok(mut thread_result) => {
517                            if let Ok(compressed_out) = thread_result.compressed {
518                                <Alloc as Allocator<u8>>::free_cell(
519                                    &mut thread_result.alloc,
520                                    compressed_out.data_backing,
521                                );
522                            }
523                            thread.0 =
524                                InternalSendAlloc::A(thread_result.alloc, UnionHasher::Uninit);
525                        }
526                        Err(join_error) => setup_result = Err(join_error),
527                    },
528                    other => thread.0 = other,
529                }
530            }
531            if let Ok(retrieved_owned_input) = spawner_and_input.unwrap() {
532                *owned_input = Owned::new(retrieved_owned_input.0);
533            }
534            return setup_result;
535        }
536        let (alloc, _extra) = alloc_per_thread[num_threads - 1].replace_with_default();
537        compression_last_thread_result = spawner_and_input.view(move |input_and_params:&(SliceW, BrotliEncoderParams)| -> CompressionThreadResult<Alloc> {
538        compress_part(hasher,
539                      num_threads - 1,
540                      num_threads,
541                      input_and_params,
542                      alloc,
543        )
544      });
545    } else {
546        if num_threads > 1 {
547            for thread_index in 1..num_threads - 1 {
548                thread_spawner.spawn(
549                    &mut spawner_and_input,
550                    &mut alloc_per_thread[thread_index],
551                    thread_index,
552                    num_threads,
553                    compress_part,
554                );
555            }
556        }
557        let (alloc, _extra) = alloc_per_thread[num_threads - 1].replace_with_default();
558        compression_last_thread_result = spawner_and_input.view(move |input_and_params:&(SliceW, BrotliEncoderParams)| -> CompressionThreadResult<Alloc> {
559        compress_part(UnionHasher::Uninit,
560                      num_threads - 1,
561                      num_threads,
562                      input_and_params,
563                      alloc,
564        )
565      });
566    }
567    let mut compression_result = Ok(0usize);
568    let mut pending_error = None;
569    let mut out_file_size = 0usize;
570    let mut bro_cat_li = BroCatli::new();
571    for (index, thread) in alloc_per_thread.iter_mut().enumerate() {
572        let cur_result = if index + 1 == num_threads {
573            match mem::replace(&mut compression_last_thread_result, Err(())) {
574                Ok(result) => Some(result),
575                Err(_err) => {
576                    set_pending_error(
577                        &mut pending_error,
578                        BrotliEncoderThreadError::OtherThreadPanic,
579                    );
580                    None
581                }
582            }
583        } else {
584            match mem::replace(
585                &mut thread.0,
586                InternalSendAlloc::SpawningOrJoining(PhantomData),
587            ) {
588                InternalSendAlloc::A(_, _) | InternalSendAlloc::SpawningOrJoining(_) => {
589                    panic!("Thread not properly spawned")
590                }
591                InternalSendAlloc::Join(join) => match join.join() {
592                    Ok(result) => Some(result),
593                    Err(err) => {
594                        set_pending_error(&mut pending_error, err);
595                        None
596                    }
597                },
598            }
599        };
600        if let Some(mut cur_result) = cur_result {
601            match cur_result.compressed {
602                Ok(compressed_out) => {
603                    if pending_error.is_none() {
604                        bro_cat_li.new_brotli_file();
605                        let mut in_offset = 0usize;
606                        let cat_result = bro_cat_li.stream(
607                            &compressed_out.data_backing.slice()[..compressed_out.data_size],
608                            &mut in_offset,
609                            output,
610                            &mut out_file_size,
611                        );
612                        match cat_result {
613                            BroCatliResult::Success | BroCatliResult::NeedsMoreInput => {
614                                compression_result = Ok(out_file_size);
615                            }
616                            BroCatliResult::NeedsMoreOutput => {
617                                set_pending_error(
618                                    &mut pending_error,
619                                    BrotliEncoderThreadError::InsufficientOutputSpace,
620                                );
621                                // not enough space
622                            }
623                            err => {
624                                set_pending_error(
625                                    &mut pending_error,
626                                    BrotliEncoderThreadError::ConcatenationError(err),
627                                );
628                                // misc error
629                            }
630                        }
631                    }
632                    <Alloc as Allocator<u8>>::free_cell(
633                        &mut cur_result.alloc,
634                        compressed_out.data_backing,
635                    );
636                }
637                Err(e) => {
638                    set_pending_error(&mut pending_error, e);
639                }
640            }
641            thread.0 = InternalSendAlloc::A(cur_result.alloc, UnionHasher::Uninit);
642        }
643    }
644    if let Some(error) = pending_error {
645        compression_result = Err(error);
646    }
647    if compression_result.is_ok() {
648        match bro_cat_li.finish(output, &mut out_file_size) {
649            BroCatliResult::Success => compression_result = Ok(out_file_size),
650            err => {
651                compression_result = Err(BrotliEncoderThreadError::ConcatenationFinalizationError(
652                    err,
653                ))
654            }
655        }
656    }
657    if let Ok(retrieved_owned_input) = spawner_and_input.unwrap() {
658        *owned_input = Owned::new(retrieved_owned_input.0); // return the input to its rightful owner before returning
659    } else if compression_result.is_ok() {
660        compression_result = Err(BrotliEncoderThreadError::OtherThreadPanic);
661    }
662    compression_result
663}
664
665mod test;
brotli/enc/threading/mod.rs

brotli/enc/threading/
mod.rs
