stream.h

Go to the documentation of this file.

// -*- mode: c++; tab-width: 4; indent-tabs-mode: t; c-file-style: "stroustrup"; -*-
// vi:set ts=4 sts=4 sw=4 noet :
// Copyright 2008, 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_AMI_STREAM_H
#define _TPIE_AMI_STREAM_H


// Include the configuration header.
#include <tpie/config.h>

// Get definitions for working with Unix and Windows
#include <tpie/portability.h>

// Get the error codes.
#include <tpie/err.h>
#include <tpie/persist.h>

#include <tpie/tempname.h>
#include <tpie/file_stream.h>
#include <tpie/file_count.h>

#include <tpie/tpie_log.h>

#include <tpie/stream_usage.h>

#include <tpie/tpie_assert.h>

namespace tpie {

    namespace ami {

    enum stream_type {
        READ_STREAM = 1,    // Open existing stream for reading
        WRITE_STREAM,       // Open for writing.  Create if non-existent
        APPEND_STREAM,      // Open for writing at end.  Create if needed.
        READ_WRITE_STREAM   // Open to read and write.
    };
    
    enum stream_status {
        STREAM_STATUS_VALID = 0,
        STREAM_STATUS_INVALID
    };

    }  //  ami namespace

}  //  tpie namespace


namespace tpie {

    namespace ami {

template<class T > 
class stream {
    
public:
    typedef T item_type;
    
    // We have a variety of constructors for different uses.

    stream();
    
    stream(const std::string& path_name, 
       stream_type st = READ_WRITE_STREAM);

    err new_substream(stream_type     st, 
              stream_offset_type  sub_begin, 
              stream_offset_type  sub_end,
              stream<T>       **sub_stream);
  
    inline stream_status status() const { 
        return m_status; 
    }
    
    inline bool is_valid() const { 
        return m_status == STREAM_STATUS_VALID; 
    }

    inline bool operator!() const { 
        return !is_valid(); 
    }

    inline err read_item(T **elt);
    
    inline err write_item(const T &elt);
  
    err read_array(T *mm_space, stream_offset_type *len);

    err read_array(T *mm_space, memory_size_type & len);
    
    err write_array(const T *mm_space, memory_size_type len);
    
    inline stream_offset_type stream_len(void) const { 
        return m_stream.size();
    }
  
    inline std::string name() const;
  
    inline err seek(stream_offset_type offset);
    
    inline stream_offset_type tell() const { 
        return m_stream.offset(); 
    }

    inline err truncate(stream_offset_type offset);
    
    err main_memory_usage(size_type *usage,
              stream_usage usage_type) const;



    static memory_size_type memory_usage(memory_size_type count);
  
    // This method used to return a statistics object, but this is no longer
    // supported.
    //struct stats_stream & stats() const { 
    //  tp_assert(0, "stream::stats() is no longer supported");
    //  return *((stats_stream *)0);
    //}

    // This method used to return a statistics object, but this is no longer
    // supported.
    //static stats_stream & gstats() {
    //  tp_assert(0, "stream::stats() is no longer supported");
    //  return *((stats_stream *)0);
    //}

    size_t available_streams(void) {
        return available_files();
    }
    
    memory_size_type chunk_size(void) const { 
        return file_base::block_size(1.0) / sizeof(T);
    }
    
    void persist(persistence p) {
        m_temp.set_persistent(p == PERSIST_PERSISTENT);
    }
    
    persistence persist() const { 
        return m_temp.is_persistent() ? PERSIST_PERSISTENT : PERSIST_DELETE;
    }

    // This function gives easy access to the stream's file name and its length.
    std::string& sprint();

    file_stream<T>& underlying_stream() {
        return m_stream;
    }
    
private:

    stream(const stream<T>& other);
    stream<T>& operator=(const stream<T>& other);
    
    temp_file m_temp;
    file_stream<T> m_stream;
    
    //bool m_destructBTEStream;
    stream_status m_status;

    static inline float block_factor() {
#ifndef STREAM_UFS_BLOCK_FACTOR
        return 1.0;
#else
#   ifdef WIN32
        return static_cast<float>(STREAM_UFS_BLOCK_FACTOR)/32;
#   else
        return static_cast<float>(STREAM_UFS_BLOCK_FACTOR)/512;
#   endif
#endif
    }
};

// Create a temporary AMI stream on one of the devices in the default
// device description. Persistence is PERSIST_DELETE by default. We
// are given the index of the string describing the desired device.
    template<class T>
    stream<T>::stream(): m_stream(block_factor()), m_status(STREAM_STATUS_INVALID)
    {
        TP_LOG_DEBUG_ID("Temporary stream in file: ");
        TP_LOG_DEBUG_ID( m_temp.path() );
        try {
            m_stream.open( m_temp );
        } catch(const stream_exception &e) {
            TP_LOG_FATAL_ID("Open failed: " << e.what());
            return;
        }
        //  Set status to VALID.
        m_status = STREAM_STATUS_VALID;
    };


// A stream created with this constructor will persist on disk at the
// location specified by the path name.
    template<class T>
    stream<T>::stream(const std::string& path_name, stream_type st) :
        m_temp(path_name, true), m_stream(block_factor()), m_status(STREAM_STATUS_INVALID) {
        try {
            m_stream.open(m_temp, st==READ_STREAM ? access_read: access_read_write);
            if (st == APPEND_STREAM) m_stream.seek(0, file_stream_base::end);
        } catch(const stream_exception &e) {
            TP_LOG_FATAL_ID("Open failed: " <<  e.what());
            return;
        }
        m_status = STREAM_STATUS_VALID;
    };

    
    // *stream::new_substream* //
    template<class T>
    err stream<T>::new_substream(stream_type     st,
                     stream_offset_type  sub_begin,
                     stream_offset_type  sub_end,
                                 stream<T>       **sub_stream)
    {
        unused(st);
        unused(sub_begin);
        unused(sub_end);
        unused(sub_stream);
        return BTE_ERROR;
    }


    template<class T>
    inline std::string stream<T>::name() const {
        return m_stream.path();
    }

// Move to a specific offset.
    template<class T>
    inline err stream<T>::seek(stream_offset_type offset) {
        try {
            m_stream.seek(offset);
        } catch(const stream_exception &e) {
            TP_LOG_WARNING_ID("BTE error - seek failed: " << e.what());
            return BTE_ERROR;
        }
        return NO_ERROR;
    }

// Truncate
    template<class T>
    inline err stream<T>::truncate(stream_offset_type offset) {
        try {
            m_stream.truncate(offset);
        } catch(const stream_exception & e) {
            TP_LOG_WARNING_ID("BTE error - truncate failed: " << e.what());
            return BTE_ERROR;
        }
        return NO_ERROR;
    }

template<class T>
memory_size_type stream<T>::memory_usage(memory_size_type count) {
    return count*(file_stream<T>::memory_usage(block_factor()) + sizeof(stream<T>));
}
    

// Query memory usage
    template<class T>
    err stream<T>::main_memory_usage(memory_size_type *usage,
                     stream_usage usage_type) const {

        switch (usage_type) {
        case STREAM_USAGE_OVERHEAD:
            *usage = sizeof(*this) + file_stream<T>::memory_usage(0.0);
            return NO_ERROR;
        case STREAM_USAGE_CURRENT:
        case STREAM_USAGE_MAXIMUM:
        case STREAM_USAGE_SUBSTREAM:
            *usage =  memory_usage(1);
            return NO_ERROR;
        case STREAM_USAGE_BUFFER:
            *usage = file_stream<T>::memory_usage(block_factor()) - file_stream<T>::memory_usage(0.0); 
            return NO_ERROR;
        }
        return BTE_ERROR;
    }

    template<class T>
    inline err stream<T>::read_item(T **elt) {
        if (!m_stream.can_read())
        return END_OF_STREAM;

        *elt = &(const_cast<T &>(m_stream.read()));
        return NO_ERROR;
    }

    template<class T>
    inline err stream<T>::write_item(const T &elt) {
        m_stream.write(elt);
        return NO_ERROR;
    }

    template<class T>
    err stream<T>::read_array(T *mm_space, stream_offset_type *len) {
        size_type l=(size_t)*len;
        err e = read_array(mm_space, l);
        *len = l;
        return e;
    }

    template<class T>
    err stream<T>::read_array(T *mm_space, memory_size_type & len) {
        size_type l = static_cast<size_type>(std::min(
            static_cast<stream_size_type>(len), 
            static_cast<stream_size_type>(m_stream.size() - m_stream.offset())));
        m_stream.read(mm_space, mm_space+l);
        return (l == len)?NO_ERROR:END_OF_STREAM;
    }

    template<class T>
    err stream<T>::write_array(const T *mm_space, size_type len) {
        m_stream.write(mm_space, mm_space+len);
        return NO_ERROR;
    }

    template<class T>
    std::string& stream<T>::sprint() {
        static std::string buf;
        std::stringstream ss;
        ss << "STREAM " << name() <<  " " << static_cast<long>(stream_len());
        ss >> buf;
        return buf;
    }

    }  //  ami namespace

}  //  tpie namespace

#endif // _TPIE_AMI_STREAM_H