diff options
Diffstat (limited to 'dec_stream.cc')
| -rw-r--r-- | dec_stream.cc | 71 |
1 files changed, 44 insertions, 27 deletions
diff --git a/dec_stream.cc b/dec_stream.cc index 91659da..64dcce3 100644 --- a/dec_stream.cc +++ b/dec_stream.cc @@ -1,4 +1,4 @@ -/* Plzip - A parallel compressor compatible with lzip +/* Plzip - Parallel compressor compatible with lzip Copyright (C) 2009 Laszlo Ersek. Copyright (C) 2009, 2010, 2011, 2012, 2013 Antonio Diaz Diaz. @@ -248,22 +248,31 @@ extern "C" void * dsplitter_s( void * arg ) Packet_courier & courier = *tmp.courier; const Pretty_print & pp = *tmp.pp; const int infd = tmp.infd; - const int hsize = 6; // header size - const int tsize = 20; // trailer size + const int hsize = File_header::size; + const int tsize = File_trailer::size; const int buffer_size = max_packet_size; const int base_buffer_size = tsize + buffer_size + hsize; uint8_t * const base_buffer = new( std::nothrow ) uint8_t[base_buffer_size]; - if( !base_buffer ) { pp( "Not enough memory" ); fatal(); } + if( !base_buffer ) { pp( "Not enough memory" ); cleanup_and_fail(); } uint8_t * const buffer = base_buffer + tsize; int size = readblock( infd, buffer, buffer_size + hsize ) - hsize; bool at_stream_end = ( size < buffer_size ); if( size != buffer_size && errno ) - { pp(); show_error( "Read error", errno ); fatal(); } - if( size <= tsize ) - { pp( "Error reading member header" ); fatal(); } - if( find_magic( buffer, 0, 4 ) != 0 ) - { pp( "Bad magic number (file not in lzip format)" ); fatal(); } + { pp(); show_error( "Read error", errno ); cleanup_and_fail(); } + if( size + hsize < min_member_size ) + { pp( "Input file is too short" ); cleanup_and_fail( 2 ); } + const File_header & header = *(File_header *)buffer; + if( !header.verify_magic() ) + { pp( "Bad magic number (file not in lzip format)" ); cleanup_and_fail( 2 ); } + if( !header.verify_version() ) + { + if( verbosity >= 0 ) + { pp(); + std::fprintf( stderr, "Version %d member format not supported.\n", + header.version() ); } + cleanup_and_fail( 2 ); + } unsigned long long partial_member_size = 0; while( true ) @@ -274,13 +283,21 @@ extern "C" void * dsplitter_s( void * arg ) newpos = find_magic( buffer, newpos, size + 4 - newpos ); if( newpos <= size ) { - unsigned long long member_size = 0; - for( int i = 1; i <= 8; ++i ) - { member_size <<= 8; member_size += base_buffer[tsize+newpos-i]; } + const File_trailer & trailer = *(File_trailer *)(buffer + newpos - tsize); + const unsigned long long member_size = trailer.member_size(); if( partial_member_size + newpos - pos == member_size ) { // header found + const File_header & header = *(File_header *)(buffer + newpos); + if( !header.verify_version() ) + { + if( verbosity >= 0 ) + { pp(); + std::fprintf( stderr, "Version %d member format not supported.\n", + header.version() ); } + cleanup_and_fail( 2 ); + } uint8_t * const data = new( std::nothrow ) uint8_t[newpos - pos]; - if( !data ) { pp( "Not enough memory" ); fatal(); } + if( !data ) { pp( "Not enough memory" ); cleanup_and_fail(); } std::memcpy( data, buffer + pos, newpos - pos ); courier.receive_packet( data, newpos - pos ); courier.receive_packet( 0, 0 ); // end of member token @@ -293,7 +310,7 @@ extern "C" void * dsplitter_s( void * arg ) if( at_stream_end ) { uint8_t * data = new( std::nothrow ) uint8_t[size + hsize - pos]; - if( !data ) { pp( "Not enough memory" ); fatal(); } + if( !data ) { pp( "Not enough memory" ); cleanup_and_fail(); } std::memcpy( data, buffer + pos, size + hsize - pos ); courier.receive_packet( data, size + hsize - pos ); courier.receive_packet( 0, 0 ); // end of member token @@ -303,7 +320,7 @@ extern "C" void * dsplitter_s( void * arg ) { partial_member_size += buffer_size - pos; uint8_t * data = new( std::nothrow ) uint8_t[buffer_size - pos]; - if( !data ) { pp( "Not enough memory" ); fatal(); } + if( !data ) { pp( "Not enough memory" ); cleanup_and_fail(); } std::memcpy( data, buffer + pos, buffer_size - pos ); courier.receive_packet( data, buffer_size - pos ); } @@ -311,7 +328,7 @@ extern "C" void * dsplitter_s( void * arg ) size = readblock( infd, buffer + hsize, buffer_size ); at_stream_end = ( size < buffer_size ); if( size != buffer_size && errno ) - { pp(); show_error( "Read error", errno ); fatal(); } + { pp(); show_error( "Read error", errno ); cleanup_and_fail(); } } delete[] base_buffer; courier.finish(); // no more packets to send @@ -339,7 +356,7 @@ extern "C" void * dworker_s( void * arg ) uint8_t * new_data = new( std::nothrow ) uint8_t[max_packet_size]; LZ_Decoder * const decoder = LZ_decompress_open(); if( !new_data || !decoder || LZ_decompress_errno( decoder ) != LZ_ok ) - { pp( "Not enough memory" ); fatal(); } + { pp( "Not enough memory" ); cleanup_and_fail(); } int new_pos = 0; bool trailing_garbage_found = false; @@ -370,7 +387,7 @@ extern "C" void * dworker_s( void * arg ) if( LZ_decompress_errno( decoder ) == LZ_header_error ) trailing_garbage_found = true; else - fatal( decompress_read_error( decoder, pp, worker_id ) ); + cleanup_and_fail( decompress_read_error( decoder, pp, worker_id ) ); } else new_pos += rd; if( new_pos > max_packet_size ) @@ -386,7 +403,7 @@ extern "C" void * dworker_s( void * arg ) courier.collect_packet( opacket, worker_id ); new_pos = 0; new_data = new( std::nothrow ) uint8_t[max_packet_size]; - if( !new_data ) { pp( "Not enough memory" ); fatal(); } + if( !new_data ) { pp( "Not enough memory" ); cleanup_and_fail(); } } if( trailing_garbage_found || LZ_decompress_finished( decoder ) == 1 ) @@ -409,9 +426,9 @@ extern "C" void * dworker_s( void * arg ) delete[] new_data; if( LZ_decompress_member_position( decoder ) != 0 ) - { pp( "Error, some data remains in decoder" ); fatal(); } + { pp( "Error, some data remains in decoder" ); cleanup_and_fail(); } if( LZ_decompress_close( decoder ) < 0 ) - { pp( "LZ_decompress_close failed" ); fatal(); } + { pp( "LZ_decompress_close failed" ); cleanup_and_fail(); } return 0; } @@ -431,7 +448,7 @@ void muxer( Packet_courier & courier, const Pretty_print & pp, const int outfd ) { const int wr = writeblock( outfd, opacket->data, opacket->size ); if( wr != opacket->size ) - { pp(); show_error( "Write error", errno ); fatal(); } + { pp(); show_error( "Write error", errno ); cleanup_and_fail(); } } delete[] opacket->data; delete opacket; @@ -462,12 +479,12 @@ int dec_stream( const int num_workers, const int infd, const int outfd, pthread_t splitter_thread; int errcode = pthread_create( &splitter_thread, 0, dsplitter_s, &splitter_arg ); if( errcode ) - { show_error( "Can't create splitter thread", errcode ); fatal(); } + { show_error( "Can't create splitter thread", errcode ); cleanup_and_fail(); } Worker_arg * worker_args = new( std::nothrow ) Worker_arg[num_workers]; pthread_t * worker_threads = new( std::nothrow ) pthread_t[num_workers]; if( !worker_args || !worker_threads ) - { pp( "Not enough memory" ); fatal(); } + { pp( "Not enough memory" ); cleanup_and_fail(); } for( int i = 0; i < num_workers; ++i ) { worker_args[i].courier = &courier; @@ -475,7 +492,7 @@ int dec_stream( const int num_workers, const int infd, const int outfd, worker_args[i].worker_id = i; errcode = pthread_create( &worker_threads[i], 0, dworker_s, &worker_args[i] ); if( errcode ) - { show_error( "Can't create worker threads", errcode ); fatal(); } + { show_error( "Can't create worker threads", errcode ); cleanup_and_fail(); } } muxer( courier, pp, outfd ); @@ -484,14 +501,14 @@ int dec_stream( const int num_workers, const int infd, const int outfd, { errcode = pthread_join( worker_threads[i], 0 ); if( errcode ) - { show_error( "Can't join worker threads", errcode ); fatal(); } + { show_error( "Can't join worker threads", errcode ); cleanup_and_fail(); } } delete[] worker_threads; delete[] worker_args; errcode = pthread_join( splitter_thread, 0 ); if( errcode ) - { show_error( "Can't join splitter thread", errcode ); fatal(); } + { show_error( "Can't join splitter thread", errcode ); cleanup_and_fail(); } if( verbosity >= 2 && out_size > 0 && in_size > 0 ) std::fprintf( stderr, "%6.3f:1, %6.3f bits/byte, %5.2f%% saved. ", |