serialization_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_SERIALIZATION_STREAM_H
#define TPIE_SERIALIZATION_STREAM_H


#include <tpie/serialization2.h>
#include <tpie/file_accessor/file_accessor.h>
#include <memory>
#include <tpie/access_type.h>
#include <tpie/array.h>
#include <tpie/tempname.h>

namespace tpie {

namespace bits {

class serialization_writer_base {
public:
    static memory_size_type block_size() {
        return 2*1024*1024;
    }

private:
    file_accessor::raw_file_accessor m_fileAccessor;
    stream_size_type m_blocksWritten;
    stream_size_type m_size;
    bool m_open;

    temp_file * m_tempFile;

protected:
    serialization_writer_base();

    void open(std::string path, bool reverse);
    void open(temp_file & tempFile, bool reverse);

private:
    void open_inner(std::string path, bool reverse);

protected:
    void write_block(const char * const s, const memory_size_type n);

    void close(bool reverse);

public:
    static memory_size_type memory_usage() { return block_size(); }

    stream_size_type file_size();
};

} // namespace bits

class serialization_writer : public bits::serialization_writer_base {
private:
    typedef bits::serialization_writer_base p_t;

    tpie::array<char> m_block;
    memory_size_type m_index;

    void write_block();

    class serializer {
        serialization_writer & wr;

    public:
        serializer(serialization_writer & wr) : wr(wr) {}

        void write(const char * const s, const memory_size_type n) {
            const char * i = s;
            memory_size_type written = 0;
            while (written != n) {
                if (wr.m_index >= wr.block_size()) wr.write_block();

                memory_size_type remaining = n - written;
                memory_size_type blockRemaining = wr.block_size() - wr.m_index;

                memory_size_type writeSize = std::min(remaining, blockRemaining);

                std::copy(i, i + writeSize, &wr.m_block[wr.m_index]);
                i += writeSize;
                written += writeSize;
                wr.m_index += writeSize;
            }
        }
    };

    friend class serializer;

public:
    void open(std::string path);
    void open(temp_file & tempFile);

    void close();

    template <typename T>
    void serialize(const T & v) {
        using tpie::serialize;
        serializer s(*this);
        serialize(s, v);
    }

    template <typename IT>
    void serialize(IT a, IT b) {
        using tpie::serialize;
        serializer s(*this);
        serialize(s, a, b);
    }
};

class serialization_reverse_writer : public bits::serialization_writer_base {
    typedef bits::serialization_writer_base p_t;

    tpie::array<char> m_block;
    memory_size_type m_index;
    std::vector<char> m_serializationBuffer;

    void write_block();

    class serializer {
        serialization_reverse_writer & wr;

    public:
        serializer(serialization_reverse_writer & wr) : wr(wr) {}

        void write(const char * const s, const memory_size_type n) {
            std::vector<char> & data = wr.m_serializationBuffer;
            memory_size_type offs = data.size();
            data.resize(data.size() + n);
            std::copy(s, s + n, &data[offs]);
        }

        ~serializer() {
            std::vector<char> & data = wr.m_serializationBuffer;
            const memory_size_type n = data.size();
            const char * const s = &data[0];
            if (wr.m_index + n <= wr.block_size()) {
                std::copy(s, s + n, &wr.m_block[block_size() - wr.m_index - n]);
                wr.m_index += n;
            } else {
                const char * i = s + n;
                memory_size_type written = 0;
                while (written != n) {
                    if (wr.m_index >= wr.block_size()) wr.write_block();

                    memory_size_type remaining = n - written;
                    memory_size_type blockRemaining = wr.block_size() - wr.m_index;

                    memory_size_type writeSize = std::min(remaining, blockRemaining);

                    std::copy(i - writeSize, i, &wr.m_block[block_size() - wr.m_index - writeSize]);
                    i -= writeSize;
                    written += writeSize;
                    wr.m_index += writeSize;
                }
            }
            data.resize(0);
        }
    };

    friend class serializer;

public:
    void open(std::string path);
    void open(temp_file & tempFile);

    void close();

    template <typename T>
    void serialize(const T & v) {
        using tpie::serialize;
        serializer s(*this);
        serialize(s, v);
    }

    template <typename IT>
    void serialize(IT a, IT b) {
        using tpie::serialize;
        serializer s(*this);
        serialize(s, a, b);
    }
};

namespace bits {

class serialization_reader_base {
public:
    static memory_size_type block_size() {
        return serialization_writer_base::block_size();
    }

private:
    file_accessor::raw_file_accessor m_fileAccessor;
    bool m_open;

protected:
    tpie::array<char> m_block;
    stream_size_type m_size;
    memory_size_type m_index;
    memory_size_type m_blockSize;

    serialization_reader_base();

    void open(std::string path, bool reverse);

    void read_block(const stream_size_type blk);

    // Check if EOF is reached, call read_block(blk) to reset m_index/m_blockSize.
    virtual void next_block() = 0;

public:
    void close();

    void read(char * const s, const memory_size_type n) {
        // TODO: inline some of this
        char * i = s;
        memory_size_type written = 0;
        while (written != n) {
            if (m_index >= m_blockSize) {
                // virtual invocation
                next_block();
            }

            memory_size_type remaining = n - written;
            memory_size_type blockRemaining = m_blockSize - m_index;

            memory_size_type readSize = std::min(remaining, blockRemaining);

            i = std::copy(m_block.get() + m_index,
                          m_block.get() + (m_index + readSize),
                          i);

            written += readSize;
            m_index += readSize;
        }
    }

    template <typename T>
    void unserialize(T & v) {
        using tpie::unserialize;
        unserialize(*this, v);
    }

    template <typename IT>
    void unserialize(IT a, IT b) {
        using tpie::unserialize;
        unserialize(*this, a, b);
    }

    static memory_size_type memory_usage() { return block_size(); }

    stream_size_type file_size();

    stream_size_type size();
};

} // namespace bits

class serialization_reader : public bits::serialization_reader_base {
    typedef bits::serialization_reader_base p_t;
    stream_size_type m_blockNumber;

protected:
    virtual void next_block() /*override*/;

public:
    serialization_reader();

    void open(std::string path);
    void open(temp_file & tempFile);

    bool can_read() {
        if (m_index < m_blockSize) return true;
        return m_blockNumber * (stream_size_type)block_size() + m_index < m_size;
    }

    stream_size_type offset();
};

class serialization_reverse_reader : public bits::serialization_reader_base {
    typedef bits::serialization_reader_base p_t;
    stream_size_type m_blockNumber;

protected:
    virtual void next_block() /*override*/;

public:
    serialization_reverse_reader();

    void open(std::string path);
    void open(temp_file & tempFile);

    bool can_read() {
        if (m_index < m_blockSize) return true;
        return m_blockNumber > 0;
    }

    stream_size_type offset();
};

}

#endif // TPIE_SERIALIZATION_STREAM_H