stream.h

Go to the documentation of this file.

// -*- mode: c++; tab-width: 4; indent-tabs-mode: t; eval: (progn (c-set-style "stroustrup") (c-set-offset 'innamespace 0)); -*-
// vi:set ts=4 sts=4 sw=4 noet :
// Copyright 2013, The TPIE development team
// 
// This file is part of TPIE.
// 
// TPIE is free software: you can redistribute it and/or modify it under
// the terms of the GNU Lesser General Public License as published by the
// Free Software Foundation, either version 3 of the License, or (at your
// option) any later version.
// 
// TPIE is distributed in the hope that it will be useful, but WITHOUT ANY
// WARRANTY; without even the implied warranty of MERCHANTABILITY or
// FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public
// License for more details.
// 
// You should have received a copy of the GNU Lesser General Public License
// along with TPIE.  If not, see <http://www.gnu.org/licenses/>

#ifndef TPIE_COMPRESSED_STREAM_H
#define TPIE_COMPRESSED_STREAM_H


#include <tpie/array.h>
#include <tpie/tpie_assert.h>
#include <tpie/tempname.h>
#include <tpie/file_base_crtp.h>
#include <tpie/file_stream_base.h>
#include <tpie/file_accessor/byte_stream_accessor.h>
#include <tpie/compressed/thread.h>
#include <tpie/compressed/buffer.h>
#include <tpie/compressed/request.h>
#include <tpie/compressed/stream_position.h>
#include <tpie/compressed/direction.h>
#include <tpie/stream_writable.h>

namespace tpie {

struct open {
    enum type {
        read_only =  00000001,
        write_only = 00000002,
        access_normal = 00000004,
        access_random = 00000010,
        compression_normal = 00000020,
        compression_all = 00000040,

        defaults = 0
    };

    friend inline open::type operator|(open::type a, open::type b)
    { return (open::type) ((int) a | (int) b); }
    friend inline open::type operator&(open::type a, open::type b)
    { return (open::type) ((int) a & (int) b); }
    friend inline open::type operator^(open::type a, open::type b)
    { return (open::type) ((int) a ^ (int) b); }
    friend inline open::type operator~(open::type a)
    { return (open::type) ~(int) a; }

    static type translate(access_type accessType, cache_hint cacheHint, compression_flags compressionFlags) {
        return (type) ((

            (accessType == access_read) ? read_only :
            (accessType == access_write) ? write_only :
            defaults) | (

            (cacheHint == tpie::access_normal) ? access_normal :
            (cacheHint == tpie::access_random) ? access_random :
            defaults) | (

            (compressionFlags == tpie::compression_normal) ? compression_normal :
            (compressionFlags == tpie::compression_all) ? compression_all :
            defaults));
    }

    static cache_hint translate_cache(open::type openFlags) {
        const open::type cacheFlags =
            openFlags & (open::access_normal | open::access_random);

        if (cacheFlags == open::access_normal)
            return tpie::access_normal;
        else if (cacheFlags == open::access_random)
            return tpie::access_random;
        else if (!cacheFlags)
            return tpie::access_sequential;
        else
            throw tpie::stream_exception("Invalid cache flags supplied");
    }

    static compression_flags translate_compression(open::type openFlags) {
        const open::type compressionFlags =
            openFlags & (open::compression_normal | open::compression_all);

        if (compressionFlags == open::compression_normal)
            return tpie::compression_normal;
        else if (compressionFlags == open::compression_all)
            return tpie::compression_all;
        else if (!compressionFlags)
            return tpie::compression_none;
        else
            throw tpie::stream_exception("Invalid compression flags supplied");
    }
};


class compressed_stream_base {
public:
    typedef std::shared_ptr<compressor_buffer> buffer_t;

protected:
    struct seek_state {
        enum type {
            none,
            beginning,
            end,
            position
        };
    };

    compressed_stream_base(memory_size_type itemSize,
                           double blockFactor);

    // Non-virtual, protected destructor
    ~compressed_stream_base();

    virtual void flush_block(compressor_thread_lock &) = 0;

    virtual void post_open() = 0;

    void open_inner(const std::string & path,
                    open::type openFlags,
                    memory_size_type userDataSize);

    compressor_thread & compressor() { return the_compressor_thread(); }

public:
    bool is_readable() const throw() { return m_canRead; }

    bool is_writable() const throw() { return m_canWrite; }

    static memory_size_type block_size(double blockFactor) throw ();

    static double calculate_block_factor(memory_size_type blockSize) throw ();

    static memory_size_type block_memory_usage(double blockFactor);

    memory_size_type block_items() const;

    memory_size_type block_size() const;

    template <typename TT>
    void read_user_data(TT & data) {
        if (sizeof(TT) != user_data_size())
            throw stream_exception("Wrong user data size");
        read_user_data(reinterpret_cast<void *>(&data), sizeof(TT));
    }

    memory_size_type read_user_data(void * data, memory_size_type count);

    template <typename TT>
    void write_user_data(const TT & data) {
        if (sizeof(TT) > max_user_data_size())
            throw stream_exception("Wrong user data size");
        write_user_data(reinterpret_cast<const void *>(&data), sizeof(TT));
    }

    void write_user_data(const void * data, memory_size_type count);

    memory_size_type user_data_size() const;

    memory_size_type max_user_data_size() const;

    const std::string & path() const;

    void open(const std::string & path,
              access_type accessType,
              memory_size_type userDataSize = 0,
              cache_hint cacheHint=access_sequential,
              compression_flags compressionFlags=compression_none)
    {
        open(path, open::translate(accessType, cacheHint, compressionFlags), userDataSize);
    }

    void open(memory_size_type userDataSize,
              cache_hint cacheHint=access_sequential,
              compression_flags compressionFlags=compression_none) {
        open(open::translate(access_read_write, cacheHint, compressionFlags), userDataSize);
    }

    void open(temp_file & file,
              access_type accessType,
              memory_size_type userDataSize = 0,
              cache_hint cacheHint=access_sequential,
              compression_flags compressionFlags=compression_none) {
        open(file, open::translate(accessType, cacheHint, compressionFlags), userDataSize);
    }

    void open(const std::string & path, compression_flags compressionFlags) {
        const memory_size_type userDataSize = 0;
        open(path, open::translate(access_read_write, access_sequential, compressionFlags), userDataSize);
    }

    void open(compression_flags compressionFlags) {
        const memory_size_type userDataSize = 0;
        open(open::translate(access_read_write, access_sequential, compressionFlags), userDataSize);
    }

    void open(temp_file & file, compression_flags compressionFlags) {
        const memory_size_type userDataSize = 0;
        open(file, open::translate(access_read_write, access_sequential, compressionFlags), userDataSize);
    }

    void open(const std::string & path, open::type openFlags=open::defaults, memory_size_type userDataSize=0);

    void open(open::type openFlags=open::defaults, memory_size_type userDataSize=0);

    void open(temp_file & file, open::type openFlags=open::defaults, memory_size_type userDataSize=0);

    void close();

protected:
    void finish_requests(compressor_thread_lock & l);

    stream_size_type last_block_read_offset(compressor_thread_lock & l);

    stream_size_type current_file_size(compressor_thread_lock & l);

    bool use_compression() { return m_byteStreamAccessor.get_compressed(); }

    void uncache_read_writes() {
        m_cachedReads = m_cachedWrites = 0;
    }

public:
    bool is_open() const { return m_open; }

    stream_size_type size() const { return m_size; }

    stream_size_type file_size() const { return size(); }

    stream_size_type offset() const {
        switch (m_seekState) {
            case seek_state::none:
                return m_offset;
            case seek_state::beginning:
                return 0;
            case seek_state::end:
                return size();
            case seek_state::position:
                return m_nextPosition.offset();
        }
        tp_assert(false, "offset: Unreachable statement; m_seekState invalid");
        return 0; // suppress compiler warning
    }

protected:
    bool m_bufferDirty;
    memory_size_type m_blockItems;
    memory_size_type m_blockSize;
    bool m_canRead;
    bool m_canWrite;
    bool m_open;
    memory_size_type m_itemSize;
    memory_size_type m_cachedReads;
    memory_size_type m_cachedWrites;
    tpie::unique_ptr<temp_file> m_ownedTempFile;
    temp_file * m_tempFile;
    file_accessor::byte_stream_accessor<default_raw_file_accessor> m_byteStreamAccessor;
    stream_size_type m_size;
    stream_buffers m_buffers;
    buffer_t m_buffer;

    stream_size_type m_streamBlocks;

    stream_size_type m_lastBlockReadOffset;
    stream_size_type m_currentFileSize;

    compressor_response m_response;

    bool m_updateReadOffsetFromWrite = false;
    stream_size_type m_lastWriteBlockNumber;

    seek_state::type m_seekState;

    stream_size_type m_readOffset;

    stream_size_type m_offset;

    stream_position m_nextPosition;

    stream_size_type m_nextReadOffset;
};

class close_on_fail_guard {
public:
    close_on_fail_guard(compressed_stream_base * s)
        : m_committed(false)
        , m_stream(s)
    {
    }

    ~close_on_fail_guard() {
        if (!m_committed) m_stream->close();
    }

    void commit() {
        m_committed = true;
    }

private:
    bool m_committed;
    compressed_stream_base * m_stream;
};

template <typename T>
class file_stream : public compressed_stream_base {
    static_assert(is_stream_writable<T>::value, "file_stream item type must be trivially copyable");

    using compressed_stream_base::seek_state;

public:
    static const file_stream_base::offset_type beginning = file_stream_base::beginning;
    static const file_stream_base::offset_type end = file_stream_base::end;
    static const file_stream_base::offset_type current = file_stream_base::current;

    typedef T item_type;
    typedef file_stream_base::offset_type offset_type;

    file_stream(double blockFactor=1.0)
        : compressed_stream_base(sizeof(T), blockFactor)
        , m_bufferBegin(0)
        , m_bufferEnd(0)
        , m_nextItem(0)
    {
    }

    // This destructor would not need to be virtual if we could use
    // final (non-subclassable) classes, but that is a C++11 feature.
    virtual ~file_stream() {
        try {
            close();
        } catch (std::exception & e) {
            log_error() << "Someone threw an error in file_stream::~file_stream: " << e.what() << std::endl;
            abort();
        }
    }

    static memory_size_type memory_usage(double blockFactor=1.0) {
        // m_buffer is included in m_buffers memory usage
        return sizeof(file_stream)
            + sizeof(temp_file) // m_ownedTempFile
            + stream_buffers::memory_usage(block_size(blockFactor)) // m_buffers
            ;
    }

    void describe(std::ostream & out) {
        if (!this->is_open()) {
            out << "[Closed stream]";
            return;
        }

        out << "[(" << m_byteStreamAccessor.path() << ") item " << offset()
            << " of " << size();
        out << " (block " << block_number()
            << " @ byte " << m_readOffset
            << ", item " << block_item_index()
            << ")";

        if (use_compression()) {
            out << ", compressed";
        } else {
            out << ", uncompressed";
        }

        switch (m_seekState) {
            case seek_state::none:
                break;
            case seek_state::beginning:
                out << ", seeking to beginning";
                break;
            case seek_state::end:
                out << ", seeking to end";
                break;
            case seek_state::position:
                out << ", seeking to position " << m_nextPosition.offset();
                out << " (block " << block_number(m_nextPosition.offset())
                    << " @ byte " << m_nextPosition.read_offset()
                    << ", item " << block_item_index(m_nextPosition.offset())
                    << ")";
                break;
        }

        if (m_bufferDirty)
            out << " dirty";

        if (m_seekState == seek_state::none) {
            if (can_read()) out << ", can read";
            else out << ", cannot read";
        }

        out << ", " << m_streamBlocks << " blocks";
        if (m_lastBlockReadOffset != std::numeric_limits<stream_size_type>::max())
            out << ", last block at " << m_lastBlockReadOffset;
        if (m_currentFileSize != std::numeric_limits<stream_size_type>::max())
            out << ", current file size " << m_currentFileSize;

        out << ']';
    }

    std::string describe() {
        std::stringstream ss;
        describe(ss);
        return ss.str();
    }

    virtual void post_open() override {
        seek(0);
    }

    void seek(stream_offset_type offset, offset_type whence=beginning) {
        tp_assert(is_open(), "seek: !is_open");
        uncache_read_writes();
        m_updateReadOffsetFromWrite = false;
        if (!use_compression()) {
            // Handle uncompressed case by delegating to set_position.
            switch (whence) {
            case beginning:
                break;
            case end:
                offset += size();
                break;
            case current:
                offset += this->offset();
                break;
            }
            set_position(stream_position(0, offset));
            return;
        }
        // Otherwise, we are in a compressed stream.
        if (offset != 0) throw stream_exception("Random seeks are not supported");
        switch (whence) {
        case beginning:
            if (m_buffer.get() != 0 && buffer_block_number() == 0) {
                // We are already reading or writing the first block.
                m_nextItem = m_bufferBegin;
                m_offset = m_readOffset = 0;
                m_seekState = seek_state::none;
            } else {
                // We need to load the first block on the next I/O.
                m_seekState = seek_state::beginning;
            }
            return;
        case end:
            if (m_buffer.get() == 0) {
                m_seekState = seek_state::end;
            } else if (m_offset == size()) {
                // no-op
                m_seekState = seek_state::none;
            } else if (// We are in the last block, and it has NOT YET been written to disk, or
                       buffer_block_number() == m_streamBlocks ||
                       // we are in the last block, and it has ALREADY been written to disk.
                       buffer_block_number()+1 == m_streamBlocks)
            {
                // If the last block is full,
                // block_item_index() reports 0 when it should report m_blockItems.
                // Compute blockItemIndex manually to handle this edge case.
                stream_size_type blockItemIndex =
                    size() - buffer_block_number() * m_blockItems;
                memory_size_type cast = static_cast<memory_size_type>(blockItemIndex);
                tp_assert(blockItemIndex == cast, "seek: blockItemIndex out of bounds");
                m_nextItem = m_bufferBegin + cast;

                m_offset = size();
                m_seekState = seek_state::none;
            } else {
                m_seekState = seek_state::end;
            }
            return;
        case current:
            return;
        }
        tp_assert(false, "seek: Unknown whence");
    }

    void truncate(stream_size_type offset) {
        tp_assert(is_open(), "truncate: !is_open");
        uncache_read_writes();
        if (offset == size())
            return;
        else if (offset == 0)
            truncate_zero();
        else if (!use_compression())
            truncate_uncompressed(offset);
        else
            throw stream_exception("Arbitrary truncate is not supported");

        if (m_tempFile) m_tempFile->update_recorded_size(m_size);
    }

    void truncate(const stream_position & pos) {
        tp_assert(is_open(), "truncate: !is_open");
        uncache_read_writes();
        if (pos.offset() == size())
            return;
        else if (pos.offset() == 0)
            truncate_zero();
        else if (!use_compression())
            truncate_uncompressed(pos.offset());
        else
            truncate_compressed(pos);

        if (m_tempFile) m_tempFile->update_recorded_size(m_size);
    }

private:
    void truncate_zero() {
        // No need to flush block
        m_buffer.reset();
        m_response.clear_block_info();
        m_updateReadOffsetFromWrite = false;
        compressor_thread_lock l(compressor());
        finish_requests(l);
        get_buffer(l, 0);
        m_size = 0;
        m_streamBlocks = 0;
        m_byteStreamAccessor.truncate(0);

        m_readOffset = 0;
        m_offset = 0;
        m_nextItem = m_bufferBegin;
        m_seekState = seek_state::none;
        uncache_read_writes();
    }

    void truncate_uncompressed(stream_size_type offset) {
        tp_assert(!use_compression(), "truncate_uncompressed called on compressed stream");

        stream_size_type currentOffset = this->offset();
        if (m_buffer.get() != 0
            && block_number(offset) == buffer_block_number()
            && buffer_block_number() == m_streamBlocks)
        {
            // We are truncating a final block that has not been written yet.
            m_size = offset;
            if (offset < m_offset) {
                m_offset = offset;
                memory_size_type blockItemIndex =
                    static_cast<memory_size_type>(offset - m_streamBlocks * m_blockItems);
                m_nextItem = m_bufferBegin + blockItemIndex;
            }
            m_bufferDirty = true;
            m_seekState = seek_state::none;
            // No need to update m_streamBlocks
        } else {
            // We need to do a truncate on the file accessor.
            // Get rid of the current block first.
            compressor_thread_lock l(compressor());
            if (offset < buffer_block_number() * m_blockItems) {
                // No need to flush current block, since we are truncating it away.
            } else {
                // Changes to the current block may still be visible after the truncate.
                if (m_bufferDirty) {
                    m_updateReadOffsetFromWrite = false;
                    flush_block(l);
                }
            }
            m_buffer.reset();
            m_bufferDirty = false;
            finish_requests(l);
            m_byteStreamAccessor.truncate(offset);
            m_size = offset;
            m_streamBlocks = (offset + m_blockItems - 1) / m_blockItems;
        }
        seek(std::min(currentOffset, offset));
    }

    void truncate_compressed(const stream_position & pos) {
        tp_assert(use_compression(), "truncate_compressed called on uncompressed stream");

        stream_size_type offset = pos.offset();
        stream_position finalDestination = (offset < this->offset()) ? pos : get_position();

        if (m_buffer.get() == 0 || block_number(offset) != buffer_block_number()) {
            set_position(pos);
            perform_seek();
        }

        // We are truncating into the currently loaded block.
        if (buffer_block_number() < m_streamBlocks) {
            m_streamBlocks = buffer_block_number() + 1;
            m_lastBlockReadOffset = pos.read_offset();
            m_currentFileSize = std::numeric_limits<stream_size_type>::max();
            compressor_thread_lock l(compressor());
            m_response.clear_block_info();
            m_updateReadOffsetFromWrite = false;
        }
        m_size = offset;
        if (offset < m_offset) {
            m_offset = offset;
            memory_size_type blockItemIndex =
                static_cast<memory_size_type>(offset - m_streamBlocks * m_blockItems);
            m_nextItem = m_bufferBegin + blockItemIndex;
        }
        m_bufferDirty = true;

        set_position(finalDestination);
    }

public:
    stream_position get_position() {
        tp_assert(is_open(), "get_position: !is_open");
        if (!use_compression()) return stream_position(0, offset());
        switch (m_seekState) {
            case seek_state::position:
                // We just set_position, so we can just return what we got.
                return m_nextPosition;
            case seek_state::beginning:
                return stream_position(0, 0);
            case seek_state::none:
                if (buffer_block_number() != m_streamBlocks) {
                    if (m_nextItem == m_bufferEnd)
                        return stream_position(m_nextReadOffset, m_offset);
                    else
                        return stream_position(m_readOffset, m_offset);
                }
                // We are in a new block at the end of the stream.
                if (m_nextItem == m_bufferEnd) {
                    tp_assert(m_bufferDirty, "At end of buffer, but bufferDirty is false?");
                    // Make sure the position we get is not at the end of a block
                    compressor_thread_lock lock(compressor());
                    m_updateReadOffsetFromWrite = false;
                    flush_block(lock);
                    get_buffer(lock, m_streamBlocks);
                    m_nextItem = m_bufferBegin;
                }
                break;
            case seek_state::end:
                // Figure out the size of the file below.
                break;
        }

        stream_size_type readOffset;
        stream_size_type blockNumber = block_number(offset());
        compressor_thread_lock l(compressor());
        if (size() % m_blockItems == 0)
            readOffset = current_file_size(l);
        else if (blockNumber == m_streamBlocks)
            readOffset = current_file_size(l);
        else if (blockNumber == m_streamBlocks - 1)
            readOffset = last_block_read_offset(l);
        else {
            tp_assert(false, "get_position: Invalid block_number");
            readOffset = 1111111111111111111ull; // avoid compiler warning
        }
        return stream_position(readOffset, offset());
    }

    void set_position(const stream_position & pos) {
        m_updateReadOffsetFromWrite = false;

        // If the code is correct, short circuiting is not necessary;
        // if the code is not correct, short circuiting might mask faults.
        /*
        if (pos == m_position) {
            m_seekState = seek_state::none;
            return;
        }
        */

        if (pos == stream_position::end()) {
            seek(0, end);
            return;
        }

        if (!use_compression() && pos.read_offset() != 0)
            throw stream_exception("set_position: Invalid position, read_offset != 0");

        if (pos.offset() > size())
            throw stream_exception("set_position: Invalid position, offset > size");

        if (m_buffer.get() != 0
            && block_number(pos.offset()) == buffer_block_number())
        {
            if (pos.read_offset() != m_readOffset) {
                // We don't always know the read offset of the current block
                // in m_readOffset, so let's assume that
                // pos.read_offset() is correct.
            }

            m_readOffset = pos.read_offset();
            m_offset = pos.offset();
            m_nextItem = m_bufferBegin + block_item_index();
            m_seekState = seek_state::none;
            return;
        }

        m_nextPosition = pos;
        m_seekState = seek_state::position;
        uncache_read_writes();
    }

    const T & read() {
        if (m_cachedReads > 0) {
            --m_cachedReads;
            ++m_offset;
            return *m_nextItem++;
        }
        const T & res = peek();
        ++m_offset;
        ++m_nextItem;
        cache_read_writes();
        return res;
    }

    const T & peek() {
        if (m_cachedReads > 0) {
            return *m_nextItem;
        }
        if (m_seekState != seek_state::none) perform_seek();
        if (m_offset == m_size) throw end_of_stream_exception();
        if (m_nextItem == m_bufferEnd) {
            compressor_thread_lock l(compressor());
            if (this->m_bufferDirty) {
                m_updateReadOffsetFromWrite = false;
                flush_block(l);
            }
            // At this point, block_number() == buffer_block_number() + 1
            read_next_block(l, block_number());
        }
        return *m_nextItem;
    }

    void skip() {
        read();
    }

    void skip_back() {
        read_back();
    }

    template <typename IT>
    void read(IT const a, IT const b) {
        for (IT i = a; i != b; ++i) *i = read();
    }

    bool can_read() {
        if (m_cachedReads > 0)
            return true;

        if (!this->m_open)
            return false;

        return offset() < size();
    }

    bool can_read_back() {
        if (!this->m_open)
            return false;

        return offset() > 0;
    }

    const T & read_back() {
        if (m_seekState != seek_state::none) {
            if (offset() == 0) throw end_of_stream_exception();
            perform_seek(read_direction::backward);
        }
        if (m_nextItem == m_bufferBegin) {
            if (m_offset == 0) throw end_of_stream_exception();
            uncache_read_writes();
            compressor_thread_lock l(compressor());
            if (this->m_bufferDirty) {
                m_updateReadOffsetFromWrite = true;
                flush_block(l);
            }
            if (use_compression()) {
                read_previous_block(l, block_number() - 1);
            } else {
                read_next_block(l, block_number() - 1);
                m_nextItem = m_bufferEnd;
            }
        }
        ++m_cachedReads;
        --m_offset;
        return *--m_nextItem;
    }

    void write(const T & item) {
        if (m_cachedWrites > 0) {
            *m_nextItem++ = item;
            ++m_size;
            ++m_offset;
            --m_cachedWrites;
            return;
        }

        if (m_seekState != seek_state::none) perform_seek();

        if (!use_compression()) {
            if (m_nextItem == m_bufferEnd) {
                compressor_thread_lock lock(compressor());
                if (m_bufferDirty) {
                    m_updateReadOffsetFromWrite = true;
                    flush_block(lock);
                }
                if (offset() == size()) {
                    get_buffer(lock, m_streamBlocks);
                    m_nextItem = m_bufferBegin;
                } else {
                    read_next_block(lock, block_number());
                }
            }
            if (offset() == m_size) ++m_size;
            *m_nextItem++ = item;
            this->m_bufferDirty = true;
            ++m_offset;
            cache_read_writes();
            return;
        }

        if (m_offset != size())
            throw stream_exception("Non-appending write attempted");

        if (m_nextItem == m_bufferEnd) {
            compressor_thread_lock l(compressor());
            if (m_bufferDirty) {
                m_updateReadOffsetFromWrite = true;
                flush_block(l);
            }
            get_buffer(l, m_streamBlocks);
            m_nextItem = m_bufferBegin;
        }

        *m_nextItem++ = item;
        this->m_bufferDirty = true;
        ++m_size;
        ++m_offset;

        cache_read_writes();
    }

    template <typename IT>
    void write(IT const a, IT const b) {
        for (IT i = a; i != b; ++i) write(*i);
    }

private:
    void perform_seek(read_direction::type dir=read_direction::forward) {
        if (!is_open()) throw stream_exception("Stream is not open");
        // This must be initialized before the compressor lock below,
        // so that it is destructed after we free the lock.
        close_on_fail_guard closeOnFail(this);

        tp_assert(!(m_seekState == seek_state::none), "perform_seek when seekState is none");

        uncache_read_writes();

        compressor_thread_lock l(compressor());
        
        m_updateReadOffsetFromWrite = false;

        if (this->m_bufferDirty)
            flush_block(l);

        m_buffer.reset();
        finish_requests(l);

        // Ensure that seek state beginning will take us to a read-only state
        if (m_seekState == seek_state::beginning && size() == 0) {
            m_seekState = seek_state::end;
        }

        // Ensure that seek state position will take us to a read-only state
        if (m_seekState == seek_state::position
            && m_nextPosition.offset() == size())
        {
            m_seekState = seek_state::end;
        }

        if (m_seekState == seek_state::beginning) {
            // The (seek beginning && size() == 0) case is handled
            // by changing seekState to end.
            // Thus, we know for sure that size() != 0, and so the
            // read_next_block will not yield an end_of_stream_exception.
            tp_assert(!(size() == 0), "Seek beginning when size is zero");
            if (use_compression()) {
                m_nextReadOffset = 0;
            }
            read_next_block(l, 0);
            m_offset = 0;
            tp_assert(m_readOffset == 0, "perform_seek: Bad readOffset after reading first block");
        } else if (m_seekState == seek_state::position) {
            stream_size_type blockNumber = block_number(m_nextPosition.offset());
            memory_size_type blockItemIndex = block_item_index(m_nextPosition.offset());

            // This cannot happen in practice due to the implementation of
            // block_number and block_item_index, but it is an important
            // assumption in the following code.
            tp_assert(!(blockItemIndex >= m_blockItems), "perform_seek: Computed block item index >= blockItems");

            if (dir == read_direction::backward && blockItemIndex == 0 && blockNumber > 0) {
                if (use_compression()) {
                    m_readOffset = m_nextPosition.read_offset();
                    read_previous_block(l, blockNumber - 1);
                    // sets m_nextItem = m_bufferEnd
                } else {
                    read_next_block(l, blockNumber - 1);
                    m_nextItem = m_bufferEnd;
                }
            } else {
                if (use_compression()) {
                    m_nextReadOffset = m_nextPosition.read_offset();
                }
                read_next_block(l, blockNumber);
                m_nextItem = m_bufferBegin + blockItemIndex;
            }

            m_offset = m_nextPosition.offset();
        } else if (m_seekState == seek_state::end) {
            if (m_streamBlocks * m_blockItems == size() && dir == read_direction::forward) {
                // The last block in the stream is full,
                // so we can safely start a new empty one.
                get_buffer(l, m_streamBlocks);
                m_nextItem = m_bufferBegin;
                if (use_compression()) {
                    m_readOffset = current_file_size(l);
                } else {
                    m_readOffset = 0;
                }
                m_offset = size();
            } else {
                // The last block in the stream is non-full, or we are going to read_back.
                if (m_streamBlocks == 0) {
                    // This cannot happen in practice,
                    // since we short-circuit seek(end) when streamBlocks == 0.
                    throw exception("Attempted seek to end when no blocks have been written");
                }
                memory_size_type blockItemIndex =
                    static_cast<memory_size_type>(size() - (m_streamBlocks - 1) * m_blockItems);
                if (use_compression()) {
                    m_nextReadOffset = last_block_read_offset(l);
                }
                read_next_block(l, m_streamBlocks - 1);
                m_nextItem = m_bufferBegin + blockItemIndex;
                m_offset = size();
            }
        } else {
            log_debug() << "Unknown seek state " << m_seekState << std::endl;
            tp_assert(false, "perform_seek: Unknown seek state");
        }

        m_seekState = seek_state::none;

        closeOnFail.commit();
    }

    void get_buffer(compressor_thread_lock & l, stream_size_type blockNumber) {
        uncache_read_writes();
        buffer_t().swap(m_buffer);
        m_buffer = this->m_buffers.get_buffer(l, blockNumber);
        while (m_buffer->is_busy()) compressor().wait_for_request_done(l);
        m_bufferBegin = reinterpret_cast<T *>(m_buffer->get());
        m_bufferEnd = m_bufferBegin + block_items();
        this->m_bufferDirty = false;
    }

    virtual void flush_block(compressor_thread_lock & lock) override {
        uncache_read_writes();
        stream_size_type blockNumber = buffer_block_number();
        stream_size_type writeOffset;
        if (!use_compression()) {
            // Uncompressed case
            writeOffset = blockNumber * m_blockSize;
        } else {
            // Compressed case
            if (blockNumber == m_streamBlocks) {
                // New block; no truncate
                writeOffset = std::numeric_limits<stream_size_type>::max();
            } else if (blockNumber == m_streamBlocks - 1) {
                // Block rewrite; truncate
                writeOffset = last_block_read_offset(lock);
                m_response.clear_block_info();
            } else {
                throw exception("flush_block: blockNumber not at end of stream");
            }
        }

        m_lastBlockReadOffset = std::numeric_limits<stream_size_type>::max();
        m_currentFileSize = std::numeric_limits<stream_size_type>::max();

        if (m_nextItem == NULL) throw exception("m_nextItem is NULL");
        if (m_bufferBegin == NULL) throw exception("m_bufferBegin is NULL");
        memory_size_type blockItems = m_blockItems;
        if (blockItems + blockNumber * m_blockItems > size()) {
            blockItems =
                static_cast<memory_size_type>(size() - blockNumber * m_blockItems);
        }
        m_buffer->set_size(blockItems * sizeof(T));
        m_buffer->set_state(compressor_buffer_state::writing);
        compressor_request r;
        r.set_write_request(m_buffer,
                            &m_byteStreamAccessor,
                            m_tempFile,
                            writeOffset,
                            blockItems,
                            blockNumber,
                            &m_response);
        compressor().request(r);
        m_bufferDirty = false;

        if (m_updateReadOffsetFromWrite) {
            m_lastWriteBlockNumber = blockNumber;
        }

        if (blockNumber == m_streamBlocks) {
            ++m_streamBlocks;
        }
    }

    void maybe_update_read_offset(compressor_thread_lock & lock) {
        if (m_updateReadOffsetFromWrite && use_compression()) {
            while (!m_response.done()) {
                m_response.wait(lock);
            }
            if (m_response.has_block_info(m_lastWriteBlockNumber)) {
                m_readOffset = m_response.get_read_offset(m_lastWriteBlockNumber);
                m_nextReadOffset = m_readOffset + m_response.get_block_size(m_lastWriteBlockNumber);
            }
            m_updateReadOffsetFromWrite = false;
        }
    }

    void read_next_block(compressor_thread_lock & lock, stream_size_type blockNumber) {
        uncache_read_writes();
        get_buffer(lock, blockNumber);

        maybe_update_read_offset(lock);

        stream_size_type readOffset;
        if (m_buffer->get_state() == compressor_buffer_state::clean) {
            m_readOffset = m_buffer->get_read_offset();
            if (use_compression()) {
                tp_assert(m_readOffset == m_nextReadOffset,
                          "read_next_block: Buffer has wrong read offset");
                m_nextReadOffset = m_readOffset + m_buffer->get_block_size();
            }
        } else {
            if (use_compression()) {
                readOffset = m_nextReadOffset;
            } else {
                stream_size_type itemOffset = blockNumber * m_blockItems;
                readOffset = blockNumber * m_blockSize;
                memory_size_type blockSize =
                    std::min(m_blockSize,
                             static_cast<memory_size_type>((size() - itemOffset) * m_itemSize));
                m_buffer->set_size(blockSize);
            }

            read_block(lock, readOffset, read_direction::forward);
            size_t blockItems = m_blockItems;
            if (size() - blockNumber * m_blockItems < blockItems) {
                blockItems = static_cast<size_t>(size() - blockNumber * m_blockItems);
            }
            size_t usableBlockSize = m_buffer->size() / sizeof(T) * sizeof(T);
            size_t expectedBlockSize = blockItems * sizeof(T);
            if (usableBlockSize != expectedBlockSize) {
                log_error() << "Expected " << expectedBlockSize << " (" << blockItems
                    << " items), got " << m_buffer->size() << " (rounded to "
                    << usableBlockSize << ')' << std::endl;
                throw exception("read_next_block: Bad buffer->get_size");
            }

            // Update m_readOffset, m_nextReadOffset
            if (use_compression()) {
                m_readOffset = readOffset;
                m_nextReadOffset = m_response.next_read_offset();
                if (m_readOffset != m_buffer->get_read_offset())
                    throw exception("read_next_block: bad get_read_offset");
                if (m_nextReadOffset != m_readOffset + m_buffer->get_block_size())
                    throw exception("read_next_block: bad get_block_size");
            } else {
                // Uncompressed case. The following is a no-op:
                //m_readOffset = 0;
                // nextReadOffset is not used.
            }
        }

        m_nextItem = m_bufferBegin;
    }

    void read_previous_block(compressor_thread_lock & lock, stream_size_type blockNumber) {
        uncache_read_writes();
        tp_assert(use_compression(), "read_previous_block: !use_compression");
        get_buffer(lock, blockNumber);

        maybe_update_read_offset(lock);

        if (m_buffer->get_state() == compressor_buffer_state::clean) {
            m_readOffset = m_buffer->get_read_offset();
            m_nextReadOffset = m_readOffset + m_buffer->get_block_size();
        } else {
            read_block(lock, m_readOffset, read_direction::backward);

            // This is backwards since we are reading backwards.
            // Confusing, I know.
            m_nextReadOffset = m_readOffset;
            m_readOffset = m_response.next_read_offset();

            if (m_readOffset != m_buffer->get_read_offset())
                throw exception("Bad buffer get_read_offset");
            if (m_nextReadOffset != m_readOffset + m_buffer->get_block_size())
                throw exception("Bad buffer get_block_size");
        }

        m_nextItem = m_bufferEnd;
    }

    void read_block(compressor_thread_lock & lock,
                    stream_size_type readOffset,
                    read_direction::type readDirection)
    {
        compressor_request r;
        r.set_read_request(m_buffer,
                           &m_byteStreamAccessor,
                           readOffset,
                           readDirection,
                           &m_response);
        m_buffer->transition_state(compressor_buffer_state::dirty,
                                   compressor_buffer_state::reading);
        compressor().request(r);
        while (!m_response.done()) {
            m_response.wait(lock);
        }
    }

    stream_size_type block_number(stream_size_type offset) {
        return offset / m_blockItems;
    }

    stream_size_type block_number() {
        return block_number(m_offset);
    }

    stream_size_type buffer_block_number() {
        stream_size_type blockNumber = block_number();
        if (m_nextItem == m_bufferEnd)
            return blockNumber - 1;
        else
            return blockNumber;
    }

    memory_size_type block_item_index(stream_size_type offset) {
        stream_size_type i = offset % m_blockItems;
        memory_size_type cast = static_cast<memory_size_type>(i);
        tp_assert(!(i != cast), "Block item index out of bounds");
        return cast;
    }

    memory_size_type block_item_index() {
        return block_item_index(m_offset);
    }

    void cache_read_writes() {
        if (m_buffer.get() == 0 || m_seekState != seek_state::none) {
            m_cachedWrites = 0;
            m_cachedReads = 0;
        } else if (offset() == size()) {
            m_cachedWrites = m_bufferDirty ? m_bufferEnd - m_nextItem : 0;
            m_cachedReads = 0;
        } else {
            m_cachedWrites = 0;
            m_cachedReads = m_bufferEnd - m_nextItem;
            if (offset() + m_cachedReads > size()) {
                m_cachedReads =
                    static_cast<memory_size_type>(size() - offset());
            }
        }
    }

private:
    T * m_bufferBegin;
    T * m_bufferEnd;

    T * m_nextItem;
};

} // namespace tpie

#endif // TPIE_COMPRESSED_STREAM_H