/// Fill the pool with decompression jobs. Returns the first job that finishes.
Fill the pool with decompression jobs. Returns the first job that finishes.
Fill the pool with decompression jobs . Returns the first job that finishes .
pub fn decompress_next_block(&mut self) -> Option<Result<UncompressedBlock>> {
        // if self.remaining_chunk_count == 0 { return None; }
        assert_eq!( // propagate panics (in release mode unlikely, but possible of course)
                    self.pool.panic_count(), 0,
                    "OpenEXR decompressor thread panicked \
            (maybe a debug assertion failed) - \
            Use non-parallel decompression to see panic messages."
        );
        while self.currently_decompressing_count < self.max_threads {
            let block = self.remaining_chunks.next();
            if let Some(block) = block {
                let block = match block {
                    Ok(block) => block,
                    Err(error) => return Some(Err(error))
                };
                // TODO if no compression, return directly
                /*if self.meta_data().headers.get(block.layer_index)
                    .ok_or_else(|| Error::invalid("header index in block"))?
                    .compression == Compression::Uncompressed
                {
                    if self.remaining_chunk_count > 0 {
                        let next = self.remaining_chunks.next();
                        if next.is_some() { self.remaining_chunk_count -= 1; }
                        return UncompressedBlock::decompress(next, headers); // no actual compression, as data is uncompressed
                    }
                }*/
                let sender = self.sender.clone();
                let meta = self.shared_meta_data_ref.clone();
                let pedantic = self.pedantic;
                self.currently_decompressing_count += 1;
                self.pool.execute(move || {
                    let decompressed_or_err = UncompressedBlock::decompress_chunk(
                        block, &meta, pedantic
                    );
                    // by now, decompressing could have failed in another thread.
                    // the error is then already handled, so we simply
                    // don't send the decompressed block and do nothing
                    let _ = sender.send(decompressed_or_err);
                });
            }
            else {
                // there are no chunks left to decompress
                break;
            }
        }
        if self.currently_decompressing_count > 0 {
            let next = self.receiver.recv()
                .expect("all decompressing senders hung up but more messages were expected");
            self.currently_decompressing_count -= 1;
            Some(next)
        }
        else {
            debug_assert!(self.receiver.try_recv().is_err(), "uncompressed chunks left in channel after decompressing all chunks"); // TODO not reliable
            debug_assert_eq!(self.len(), 0, "compressed chunks left after decompressing all chunks");
            None
        }
    }
pub fn decompress_next_block ( & mut self ) -> Option < Result < UncompressedBlock > > { // if self.remaining_chunk_count == 0 { return None; } assert_eq ! ( // propagate panics (in release mode unlikely, but possible of course) self pool panic_count ( ) 0 " \
 \
 " ) ; while self . currently_decompressing_count < self . max_threads { let block = self . remaining_chunks . next ( ) ; if let Some ( block ) = block { let block = match block { Ok ( block ) => block , Err ( error ) => return Some ( Err ( error ) ) } ; // TODO if no compression, return directly /*if self.meta_data().headers.get(block.layer_index)
                    .ok_or_else(|| Error::invalid("header index in block"))?
                    .compression == Compression::Uncompressed
                {
                    if self.remaining_chunk_count > 0 {
                        let next = self.remaining_chunks.next();
                        if next.is_some() { self.remaining_chunk_count -= 1; }
                        return UncompressedBlock::decompress(next, headers); // no actual compression, as data is uncompressed
                    }
                }*/ let sender = self . sender . clone ( ) ; let meta = self . shared_meta_data_ref . clone ( ) ; let pedantic = self . pedantic ; self . currently_decompressing_count += 1 ; self . pool . execute ( move | | { let decompressed_or_err = UncompressedBlock :: decompress_chunk ( block , & meta , pedantic ) ; // by now, decompressing could have failed in another thread. // the error is then already handled, so we simply // don't send the decompressed block and do nothing let _ = sender . send ( decompressed_or_err ) ; } ) ; } else { // there are no chunks left to decompress break ; } } if self . currently_decompressing_count > 0 { let next = self . receiver . recv ( ) . expect ( " " ) ; self . currently_decompressing_count -= 1 ; Some ( next ) } else { debug_assert ! ( self receiver try_recv ( ) is_err ( ) " " ) ; // TODO not reliable debug_assert_eq ! ( self len ( ) 0 " " ) ; None } }
Fill the pool with decompression jobs.
Fill the pool with decompression jobs .
&mut self