lsst::ap::io::CompressedFileWriter Class Reference

A sequential writer for compressed files that uses asynchronous IO to overlap IO with compression. More...

#include <FileIo.h>

Inheritance diagram for lsst::ap::io::CompressedFileWriter:

Public Member Functions
	CompressedFileWriter (std::string const &fileName, bool const overwrite=false, std::size_t const blockSize=262144)
virtual	~CompressedFileWriter ()
virtual void	write (unsigned char const *const buf, std::size_t const size)
	Writes len bytes from buf to the underlying storage device. More...
virtual void	finish ()
	Moves modified data to the underlying storage device and marks the SequentialWriter as finished. More...
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 output buffers More...
::z_stream	_stream
	zlib state More...
::aiocb	_request
	Outstanding IO request. More...
std::size_t const	_blockSize
	read granularity More...
int	_fd
bool	_started

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 writer for compressed files that uses asynchronous IO to overlap IO with compression.

Gzip compatible files are written by this class.

Definition at line 204 of file FileIo.h.

Constructor & Destructor Documentation

lsst::ap::io::CompressedFileWriter::CompressedFileWriter ( std::string const &  fileName, bool const  overwrite = false, std::size_t const  blockSize = 262144  )

explicit

Definition at line 450 of file FileIo.cc.

  :
    _memory(0),
    _buffers(0),
    _blockSize(blockSize),
    _fd(-1),
    _started(false)
{
    if (blockSize < 1024 || blockSize > 16777216) {
        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_WRONLY | O_CREAT | O_APPEND | (overwrite ? O_TRUNC : O_EXCL),
        S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH
    );
    assert(fd != -1);
    ScopeGuard fdGuard(boost::bind(::close, fd));

    // setup zlib (15 window bits, add 16 to indicate a gzip compatible header is desired)
    if (deflateInit2(&_stream, 1, Z_DEFLATED, MAX_WBITS + 16, MAX_MEM_LEVEL, Z_DEFAULT_STRATEGY) != Z_OK) {
        throw LSST_EXCEPT(ex::RuntimeError,
                          "[zlib] deflateInit2() failed to initialize compression stream");
    }

    _request.aio_fildes = fd;
    _request.aio_buf    = _buffers;
    _request.aio_nbytes = blockSize;
    _fd                 = fd;
    using std::swap;
    swap(_memory, mem);
    fdGuard.dismiss();
}

lsst::ap::io::CompressedFileWriter::~CompressedFileWriter ( )

virtual

Definition at line 500 of file FileIo.cc.

{ cleanup(); }

Member Function Documentation

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

private

Definition at line 503 of file FileIo.cc.

                                           {
    if (_fd != -1) {
        if (_started) {
            ::aio_cancel(_fd, 0);
        }
        ::close(_fd);
        ::deflateEnd(&_stream);            
        _fd = -1;
    }
}

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

inlineprivate

Definition at line 232 of file FileIo.h.

                                    {
        cleanup();
        _state = state;
    }

void lsst::ap::io::CompressedFileWriter::finish ( )

virtual

Moves modified data to the underlying storage device and marks the SequentialWriter as finished.

Implements lsst::ap::io::SequentialWriter.

Definition at line 606 of file FileIo.cc.

                                          {

    if (_state != IN_PROGRESS) {
        throw LSST_EXCEPT(ex::IoError,
            "finish() called on a finished or failed CompressedFileWriter");
    }

    if (_stream.avail_out == 0) {
        unsigned char * buf = static_cast<unsigned char *>(const_cast<void *>(_request.aio_buf));
        _stream.next_out  = (buf == _buffers) ? buf + _blockSize : _buffers;
        _stream.avail_out = _blockSize;
    }

    while (true) {

        int const zret = ::deflate(&_stream, Z_FINISH);
        if (zret != Z_STREAM_END && zret != Z_OK) {
            cleanup(FAILED);
            throw LSST_EXCEPT(ex::RuntimeError,
                (boost::format("[zlib] deflate() failed, return code: %1%") % zret).str());
        }
        if (_started) {
            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_write() 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 write that just completed
            if (::aio_return(&_request) != static_cast<int>(_blockSize)) {
                cleanup(FAILED);
                throw LSST_EXCEPT(ex::IoError,
                    (boost::format("aio_write() failed, errno: %1%") %
                        ::aio_error(&_request)).str());
            }
        }
        if (zret == Z_STREAM_END) {
            break;
        }

        unsigned char * buf = static_cast<unsigned char *>(const_cast<void *>(_request.aio_buf));
        _stream.next_out  = buf;
        _stream.avail_out = _blockSize;
        // issue asynchronous write of the deflated data
        _request.aio_buf = (buf == _buffers) ? buf + _blockSize : _buffers;
        if (::aio_write(&_request) != 0) {
            cleanup(FAILED);
            throw LSST_EXCEPT(ex::IoError,
                (boost::format("aio_write() failed to enqueue IO request") % errno).str());
        }
        _started = true;

    }

    // no outstanding IO remains, perform final (blocking) write
    std::size_t bytesRemaining = _blockSize - _stream.avail_out;
    unsigned char * buf = _stream.next_out - bytesRemaining;
    while (bytesRemaining > 0) {
        ::ssize_t nb = ::write(_fd, buf, bytesRemaining);
        if (nb < 0) {
            cleanup(FAILED);
            throw LSST_EXCEPT(ex::IoError,
                              (boost::format("write() failed, errno: %1%") % errno).str());
        }
        bytesRemaining -= static_cast<std::size_t>(nb);
        buf += nb;
    }
    // flush both userland and kernel buffers
    while (::fsync(_fd) != 0) {
        if (errno != EINTR) {
            cleanup(FAILED);
            throw LSST_EXCEPT(ex::IoError,
                              (boost::format("fsync() failed, errno: %1%") % errno).str());
        }
        errno = 0;
    }
    cleanup(FINISHED);
}

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

inline

Definition at line 219 of file FileIo.h.

{ return _blockSize; }

void lsst::ap::io::CompressedFileWriter::write ( unsigned char const *const  buf, std::size_t const  len  )

virtual

Writes len bytes from buf to the underlying storage device.

Implements lsst::ap::io::SequentialWriter.

Definition at line 515 of file FileIo.cc.

  {
    if (len == 0) {
        return;
    }
    if (buf == 0) {
        throw LSST_EXCEPT(ex::InvalidParameterError,
                          "null pointer to read destination");
    }
    if (_state != IN_PROGRESS) {
        throw LSST_EXCEPT(ex::IoError,
                          "write() called on a finished or failed CompressedFileWriter");
    }

    _stream.next_in  = const_cast<Bytef *>(buf);
    _stream.avail_in = len;
    if (_stream.avail_out == 0) {
        unsigned char * buf = static_cast<unsigned char *>(const_cast<void *>(_request.aio_buf));
        _stream.next_out = (buf == _buffers) ? buf + _blockSize : _buffers;
        _stream.avail_out = _blockSize;
    }

    // deflate/write until the buffer passed in by the user has been consumed
    do {

        int const zret = ::deflate(&_stream, Z_NO_FLUSH);
        if (zret != Z_OK) {
            cleanup(FAILED);
            throw LSST_EXCEPT(ex::RuntimeError,
                (boost::format("[zlib] deflate() failed, return code: %1%") % zret).str());
        }
        if (_stream.avail_out != 0) {
            if (_stream.avail_in != 0) {
                cleanup(FAILED);
                throw LSST_EXCEPT(ex::RuntimeError,
                                  "[zlib] deflate() failed to consume input");
            }
            break;
        }

        if (_stream.avail_out == 0) {
            if (_started) {
                // wait for pending write to finish
                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_write() 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 write that just completed
                if (::aio_return(&_request) != static_cast<int>(_blockSize)) {
                    cleanup(FAILED);
                    throw LSST_EXCEPT(ex::IoError,
                        (boost::format("aio_write() failed, errno: %1%") %
                            ::aio_error(&_request)).str());
                }
            }
            unsigned char * buf = static_cast<unsigned char *>(const_cast<void *>(_request.aio_buf));
            _request.aio_buf  = _stream.next_out - _blockSize;
            _stream.next_out  = buf;
            _stream.avail_out = _blockSize;
            // issue asynchronous write
            if (aio_write(&_request) != 0) {
                cleanup(FAILED);
                throw LSST_EXCEPT(ex::IoError,
                    (boost::format("aio_write() failed to enqueue IO request") % errno).str());
            }
            _started = true;
        } else if (_stream.avail_in != 0) {
            throw LSST_EXCEPT(ex::RuntimeError,
                              "[zlib] deflate() failed to consume input");
        }

    } while(_stream.avail_in != 0);
}

Member Data Documentation

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

private

read granularity

Definition at line 227 of file FileIo.h.

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

private

aligned output buffers

Definition at line 224 of file FileIo.h.

int lsst::ap::io::CompressedFileWriter::_fd

private

Definition at line 228 of file FileIo.h.

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

private

Definition at line 223 of file FileIo.h.

::aiocb lsst::ap::io::CompressedFileWriter::_request

private

Outstanding IO request.

Definition at line 226 of file FileIo.h.

bool lsst::ap::io::CompressedFileWriter::_started

private

Definition at line 229 of file FileIo.h.

::z_stream lsst::ap::io::CompressedFileWriter::_stream

private

zlib state

Definition at line 225 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