lsst::ap::io::CompressedFileReader Class Reference

A sequential reader for compressed files that uses asynchronous IO to overlap IO with decompression. More...

#include <FileIo.h>

Inheritance diagram for lsst::ap::io::CompressedFileReader:

Public Member Functions
	CompressedFileReader (std::string const &fileName, std::size_t const blockSize=262144)
virtual	~CompressedFileReader ()
virtual std::size_t	read (unsigned char *const buf, std::size_t const size)
std::size_t	getBlockSize () const
Public Member Functions inherited from lsst::ap::io::SequentialIoBase
	SequentialIoBase ()
virtual	~SequentialIoBase ()=0
bool	finished () const
	Returns true if there are no more bytes available for reading. More...
bool	failed () const
	Returns true if a read operation failed. More...
State	getState () const
	Returns the state of the SequentialReader. More...

Private Member Functions
void	cleanup ()
void	cleanup (State const state)

Private Attributes
boost::scoped_array< unsigned char >	_memory
unsigned char *	_buffers
	aligned input buffers More...
::z_stream	_stream
	zlib state More...
::aiocb	_request
	Outstanding IO request. More...
std::size_t const	_blockSize
	read granularity More...
std::size_t	_fileSize
	Size of the file being read. More...
std::size_t	_remaining
	Bytes that haven't yet been read. More...
int	_fd
	file descriptor More...

Additional Inherited Members
Public Types inherited from lsst::ap::io::SequentialIoBase
enum	State { IN_PROGRESS = 0, FINISHED, FAILED }
Protected Attributes inherited from lsst::ap::io::SequentialIoBase
State	_state

Detailed Description

A sequential reader for compressed files that uses asynchronous IO to overlap IO with decompression.

Gzip compatible files, or files written with zlib compression can be read by this class.

Definition at line 165 of file FileIo.h.

Constructor & Destructor Documentation

lsst::ap::io::CompressedFileReader::CompressedFileReader ( std::string const &  fileName, std::size_t const  blockSize = 262144  )

explicit

Definition at line 225 of file FileIo.cc.

  :
    _memory(0),
    _buffers(0),
    _blockSize(blockSize),
    _fileSize(0),
    _remaining(0),
    _fd(-1)
{
    if (blockSize < 1024 || blockSize > 16777216 || (blockSize & (blockSize - 1)) != 0) {
        throw LSST_EXCEPT(ex::InvalidParameterError,
                          "I/O block size must be a power of 2 between 2^10 and 2^24 bytes");
    }

    std::memset(&_stream,  0, sizeof(::z_stream));
    std::memset(&_request, 0, sizeof(::aiocb));
    _request.aio_fildes = -1;

    // allocate IO buffers, store pointer to 8k aligned location inside buffer
    boost::scoped_array<unsigned char> mem(new unsigned char[2*blockSize + 8192]);
    std::size_t buf = reinterpret_cast<std::size_t>(mem.get());
    _buffers = reinterpret_cast<unsigned char *>((buf + 8191) & ~static_cast<std::size_t>(8191));

    // open file
    int const fd = openFile(fileName, O_RDONLY);
    if (fd == -1) {
        _state = FINISHED;
        return;
    }
    ScopeGuard fdGuard(boost::bind(::close, fd));

    // determine file size
    struct stat sb;
    if (::fstat(fd, &sb) != 0) {
        throw LSST_EXCEPT(ex::IoError,
            (boost::format("fstat() failed on file %1%, errno: %2%") % fileName % errno).str());
    }
    _fileSize = static_cast<std::size_t>(sb.st_size);
    if (_fileSize == 0) {
        _state = FINISHED;
        return;
    }
    _remaining = _fileSize;

    // setup zlib (specify auto-detection of zlib/gzip header)
    if (inflateInit2(&_stream, MAX_WBITS + 32) != Z_OK) {
        throw LSST_EXCEPT(ex::RuntimeError,
                          "[zlib] inflateInit2() failed to initialize decompression stream");
    }
    ScopeGuard zlibGuard(boost::bind(::inflateEnd, &_stream));

    // issue first async read
    std::size_t const nb = std::min(blockSize, _fileSize);
    _request.aio_fildes = fd;
    _request.aio_buf    = _buffers;
    _request.aio_nbytes = nb;
    if (::aio_read(&_request) != 0) {
        throw LSST_EXCEPT(ex::IoError,
            (boost::format("aio_read() failed to enqueue IO request for file %1%, errno: %2%") %
                fileName % errno).str());
    }
    _fd = fd;
    using std::swap;
    swap(_memory, mem);
    zlibGuard.dismiss();
    fdGuard.dismiss();
}

lsst::ap::io::CompressedFileReader::~CompressedFileReader ( )

virtual

Definition at line 296 of file FileIo.cc.

                                                    {
    cleanup();
}

Member Function Documentation

void lsst::ap::io::CompressedFileReader::cleanup ( )

private

Definition at line 301 of file FileIo.cc.

                                           {
    if (_fd != -1) {
        if (_state == IN_PROGRESS) {
            ::aio_cancel(_fd, 0);
        }
        ::close(_fd);
        ::inflateEnd(&_stream);
        _fd = -1;
    }
}

void lsst::ap::io::CompressedFileReader::cleanup ( State const  state )

inlineprivate

Definition at line 191 of file FileIo.h.

                                    {
        cleanup();
        _state = state;
    }

std::size_t lsst::ap::io::CompressedFileReader::getBlockSize ( ) const

inline

Definition at line 177 of file FileIo.h.

{ return _blockSize; }

std::size_t lsst::ap::io::CompressedFileReader::read ( unsigned char *const  buf, std::size_t const  len  )

virtual

Reads up to len bytes from an underlying storage device into buf and returns the number of bytes actually read.

Implements lsst::ap::io::SequentialReader.

Definition at line 313 of file FileIo.cc.

  {
    if (len == 0) {
        return 0;
    }
    if (buf == 0) {
        throw LSST_EXCEPT(ex::InvalidParameterError, "null pointer to read destination");
    }
    if (_state == FAILED) {
        throw LSST_EXCEPT(ex::IoError, "read() called on a failed CompressedFileReader");
    } else if (_state == FINISHED) {
        return 0;
    }

    // if the current block has not been fully transferred to the user,
    // decompress as much as possible of what remains
    _stream.next_out  = buf;
    _stream.avail_out = len;
    if (_stream.avail_in > 0) {
        int const zret = ::inflate(&_stream, Z_SYNC_FLUSH);
        if (zret == Z_STREAM_END) {
            if (_remaining != 0) {
                cleanup(FAILED);
                throw LSST_EXCEPT(ex::RuntimeError,
                                  "[zlib] inflate(): unexpected end of stream");
            }
            if (_stream.avail_in != 0) {
                cleanup(FAILED);
                throw LSST_EXCEPT(ex::RuntimeError,
                                  "[zlib] inflate() failed to consume input");
            }
            cleanup(FINISHED);
            return len - _stream.avail_out;
        } else if (zret == Z_OK) {
            if (_stream.avail_out == 0) {
                return len;
            } else if (_remaining == 0) {
                cleanup(FAILED);
                throw LSST_EXCEPT(ex::RuntimeError,
                                  "[zlib] inflate(): expected end of stream");
            } else if (_stream.avail_in != 0) {
                cleanup(FAILED);
                throw LSST_EXCEPT(ex::RuntimeError,
                                  "[zlib] inflate() failed to consume input");
            }
            // user has space for even more data
        } else {
            cleanup(FAILED);
            throw LSST_EXCEPT(ex::RuntimeError,
                (boost::format("[zlib] inflate() failed, return code: %1%") % zret).str());
        }
    }

    // Current block is consumed, so block on outstanding request
    while (::aio_error(&_request) == EINPROGRESS) {
        TimeSpec timeout;
        ::aiocb * const list = &_request;
        timeout.tv_sec = 10; // wait up to 10 seconds
        if (::aio_suspend(&list, 1, &timeout) != 0) {
            if (errno != EINTR) {
                if (errno == EAGAIN) {
                    cleanup(FAILED);
                    throw LSST_EXCEPT(ex::TimeoutError, "aio_read() timed out");
                } else {
                    cleanup(FAILED);
                    throw LSST_EXCEPT(ex::IoError,
                        (boost::format("aio_suspend() failed, errno: %1%") % errno).str());
                }
            }
            errno = 0;
        }
    }

    // check status of the read that just completed
    int const nb = ::aio_return(&_request);
    if (nb < 0) {
        cleanup(FAILED);
        throw LSST_EXCEPT(ex::IoError,
            (boost::format("aio_read() failed, errno: %1%") % ::aio_error(&_request)).str());
    } else if (nb == 0) {
        cleanup(FAILED);
        throw LSST_EXCEPT(ex::IoError, "aio_read(): unexpected end of file reached");
    }

    unsigned char * ready = static_cast<unsigned char *>(const_cast<void *>(_request.aio_buf));
    _remaining -= static_cast<std::size_t>(nb);
    if (_remaining > 0) {
        // issue prefetch for the next block
        _request.aio_buf     = (ready == _buffers) ? ready + _blockSize : _buffers;
        _request.aio_offset  = _fileSize - _remaining;
        _request.aio_nbytes  = std::min(_blockSize, _remaining);
        if (::aio_read(&_request) != 0) {
            cleanup(FAILED);
            throw LSST_EXCEPT(ex::IoError,
                (boost::format("aio_read() failed to enqueue IO request, errno: %1%") % errno).str());
        }
    }

    // Decompress into user-supplied buffer
    _stream.next_in  = ready;
    _stream.avail_in = nb;
    int const zret = ::inflate(&_stream, Z_SYNC_FLUSH);
    if (zret == Z_STREAM_END) {
        if (_remaining != 0) {
            cleanup(FAILED);
            throw LSST_EXCEPT(ex::RuntimeError,
                              "[zlib] inflate(): unexpected end of stream");
        }
        if (_stream.avail_in != 0) {
            cleanup(FAILED);
            throw LSST_EXCEPT(ex::RuntimeError,
                              "[zlib] inflate() failed to consume input");
        }
        cleanup(FINISHED);
    } else if (zret != Z_OK) {
        cleanup(FAILED);
        throw LSST_EXCEPT(ex::RuntimeError,
            (boost::format("[zlib] inflate() failed, return code: %1%") % zret).str());
    } else if (_stream.avail_out != 0) {
        if (_remaining == 0) {
            cleanup(FAILED);
            throw LSST_EXCEPT(ex::RuntimeError,
                              "[zlib] inflate(): expected end of stream");
        } else if (_stream.avail_in != 0) {
            cleanup(FAILED);
            throw LSST_EXCEPT(ex::RuntimeError,
                              "[zlib] inflate() failed to consume input");
        }
    }
    return len - _stream.avail_out;
}

Member Data Documentation

std::size_t const lsst::ap::io::CompressedFileReader::_blockSize

private

read granularity

Definition at line 185 of file FileIo.h.

unsigned char* lsst::ap::io::CompressedFileReader::_buffers

private

aligned input buffers

Definition at line 182 of file FileIo.h.

int lsst::ap::io::CompressedFileReader::_fd

private

file descriptor

Definition at line 188 of file FileIo.h.

std::size_t lsst::ap::io::CompressedFileReader::_fileSize

private

Size of the file being read.

Definition at line 186 of file FileIo.h.

boost::scoped_array<unsigned char> lsst::ap::io::CompressedFileReader::_memory

private

Definition at line 181 of file FileIo.h.

std::size_t lsst::ap::io::CompressedFileReader::_remaining

private

Bytes that haven't yet been read.

Definition at line 187 of file FileIo.h.

::aiocb lsst::ap::io::CompressedFileReader::_request

private

Outstanding IO request.

Definition at line 184 of file FileIo.h.

::z_stream lsst::ap::io::CompressedFileReader::_stream

private

zlib state

Definition at line 183 of file FileIo.h.

---

The documentation for this class was generated from the following files:

• /lsst/home/lsstsw/stack/Linux64/ap/master-g60f1e072bb+1/include/lsst/ap/io/FileIo.h
• /lsst/home/lsstsw/stack/Linux64/ap/master-g60f1e072bb+1/src/io/FileIo.cc