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/tpie_export.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/compressed/stream_position.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; }
};
 
 
class compressed_stream_base_p;
 
class TPIE_EXPORT compressed_stream_base {
public:
    friend class compressed_stream_base_p;
 
    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 file_stream_base::offset_type offset_type;
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();
 
 
    void cache_read_writes();
 
    void peak_unlikely();
    
    void read_back_unlikely();
    
    void write_unlikely(const char * item);
 
    static memory_size_type memory_usage(double blockFactor=1.0) noexcept;
    
public:
    bool is_readable() const noexcept;
 
    bool is_writable() const noexcept;
 
    static memory_size_type block_size(double blockFactor) noexcept;
 
    static double calculate_block_factor(memory_size_type blockSize) noexcept;
 
    static memory_size_type block_memory_usage(double blockFactor) noexcept;
 
    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);
 
    void open(memory_size_type userDataSize,
              cache_hint cacheHint=access_sequential,
              compression_flags compressionFlags=compression_none);
 
    void open(temp_file & file,
              access_type accessType,
              memory_size_type userDataSize = 0,
              cache_hint cacheHint=access_sequential,
              compression_flags compressionFlags=compression_none);
 
    void open(const std::string & path, compression_flags compressionFlags);
 
    void open(compression_flags compressionFlags);
 
    void open(temp_file & file, compression_flags compressionFlags);
 
    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();
 
    bool is_open() const noexcept;
    
    stream_size_type offset() const;
 
    void describe(std::ostream & out);
    
    std::string describe();
 
 
    void seek(stream_offset_type offset, offset_type whence=beginning);
    
    void truncate(stream_size_type offset);
    
    void truncate(const stream_position & pos);
    
    stream_position get_position();
    
    void set_position(const stream_position & pos);
 
    stream_size_type size() const { return m_size; }
 
    stream_size_type file_size() const { return size(); }
 
    
    bool can_read() {
        if (m_cachedReads > 0)  return true;
        return offset() < size();
    }
 
    bool can_read_back() {
        return offset() > 0;
    }
protected:
    memory_size_type m_cachedReads;
    memory_size_type m_cachedWrites;
    
    stream_size_type m_size;
    seek_state::type m_seekState;
    
    stream_size_type m_offset;
 
    char * m_bufferBegin;
 
    char * m_bufferEnd;
        
    char * m_nextItem;
 
    compressed_stream_base_p * m_p;
};
 
 
 
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");
public:
    typedef T item_type;
    
    file_stream(double blockFactor=1.0)
        : compressed_stream_base(sizeof(T), blockFactor) {}
    
    static memory_size_type memory_usage(double blockFactor=1.0) noexcept {
        // m_buffer is included in m_buffers memory usage
        return sizeof(file_stream) + compressed_stream_base::memory_usage(blockFactor);
    }
 
    const T & read() {
        if (m_cachedReads == 0) {
            peak_unlikely();
            const T & res = *reinterpret_cast<const T*>(m_nextItem);
            ++m_offset;
            m_nextItem += sizeof(T);
            cache_read_writes();
            return res;
        }
        --m_cachedReads;
        ++m_offset;
        const T & res = *reinterpret_cast<const T*>(m_nextItem);
        m_nextItem += sizeof(T);
        return res;
    }
    
    const T & peek() {
        if (m_cachedReads == 0) peak_unlikely();
        return *reinterpret_cast<const T*>(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();
    }
    
    const T & read_back() {
        if (m_seekState != seek_state::none || m_nextItem == m_bufferBegin) read_back_unlikely();
        ++m_cachedReads;
        --m_offset;
        m_nextItem -= sizeof(T);
        return *reinterpret_cast<const T *>(m_nextItem);
    }
    
    void write(const T & item) {
        if (m_cachedWrites == 0) {
            write_unlikely(reinterpret_cast<const char*>(&item));
            return;
        }
        memcpy(m_nextItem, &item, sizeof(T));
        m_nextItem += sizeof(T);
        ++m_size;
        ++m_offset;
        --m_cachedWrites;
        return;
    }
 
    template <typename IT>
    void write(IT const a, IT const b) {
        for (IT i = a; i != b; ++i) write(*i);
    }
};
 
} // namespace tpie
 
#endif // TPIE_COMPRESSED_STREAM_H