/home/mhx/git/github/libmoost/include/moost/io/block_store.hpp

Go to the documentation of this file.

/* vim:set ts=3 sw=3 sts=3 et: */
#ifndef MOOST_IO_BLOCK_STORE_HPP__
#define MOOST_IO_BLOCK_STORE_HPP__

#include <algorithm>
#include <string>
#include <fstream>
#include <stdexcept>

#include <boost/filesystem/path.hpp>
#include <boost/filesystem/operations.hpp>

#include <boost/scoped_array.hpp>
#include <boost/thread/mutex.hpp>

#include "../container/resource_stack.hpp"

namespace moost { namespace io {

class block_store
{
private:
   moost::container::resource_stack< std::fstream > m_rstreams;
   size_t                                           m_block_size;
   size_t                                           m_allocated;
   std::vector<size_t>                              m_free_list;
   boost::mutex                                     m_mutex;

   void free(size_t index)
   {
      boost::mutex::scoped_lock lock(m_mutex);
      if (index >= m_allocated)
         return; // weird
      if (index == m_allocated - 1)
         --m_allocated;
      else
      {
         std::vector<size_t>::iterator it = std::lower_bound(m_free_list.begin(), m_free_list.end(), index);
         if (it == m_free_list.end() || *it != index)
            m_free_list.insert(it, index);
      }
   }

   size_t alloc()
   {
      size_t ret_val;
      boost::mutex::scoped_lock lock(m_mutex);
      if (m_free_list.empty())
         ret_val = m_allocated++;
      else
      {
         ret_val = m_free_list.back();
         m_free_list.pop_back();
      }
      return ret_val;
   }

   std::streampos getpos(size_t index)
   {
      return static_cast<std::streampos>(sizeof(size_t)) + static_cast<std::streampos>( m_block_size ) * static_cast<std::streampos>( index );
   }

public:

   class scoped_block
   {
   private:
      block_store &                                                     m_block_store;
      moost::container::resource_stack< std::fstream >::scoped_resource m_rstream;
      size_t                                                            m_index;
      bool                                                              m_free;
   public:
      scoped_block(block_store & block_store_)
         : m_block_store(block_store_),
           m_rstream(block_store_.m_rstreams),
           m_free(false)
      {
         m_index = m_block_store.alloc();
         std::streampos newPos = m_block_store.getpos(m_index);
         m_rstream->seekg( newPos );
         m_rstream->seekp( newPos );
      }
      scoped_block(block_store & block_store_, size_t index)
         : m_block_store(block_store_),
           m_rstream(block_store_.m_rstreams),
           m_index(index),
           m_free(false)
      {
         std::streampos newPos = m_block_store.getpos(m_index);
         m_rstream->seekg( newPos );
         m_rstream->seekp( newPos );
      }
      ~scoped_block()
      {
         if (m_free)
            m_block_store.free(m_index);
         else
            m_rstream->rdbuf()->pubsync();
      }
      void free()                  { m_free = true; }
      std::fstream & operator * () { return m_rstream.operator* (); }
      std::fstream * operator ->() { return m_rstream.operator->(); }
      size_t index()               { return m_index; }
      size_t block_size()          { return m_block_store.block_size(); }
   };

   block_store(const std::string & path,
      size_t block_size,
      size_t num_streams = 8  /* the number of concurrent streams accessing this blocks store */ )
      : m_block_size(block_size),
        m_allocated(0)
   {
      if (!boost::filesystem::exists(boost::filesystem::path(path)))
         std::fstream(path.c_str(), std::ios::binary | std::ios::out | std::ios::app); // poor man's touch
      else
      {
         std::fstream in(path.c_str(), std::ios::binary | std::ios::out | std::ios::in);
         in.read(reinterpret_cast<char *>(&m_allocated), sizeof(size_t));
         in.seekg(getpos(m_allocated));
         size_t free_list_size = 0;
         in.read(reinterpret_cast<char *>(&free_list_size), sizeof(size_t));
         m_free_list.resize(free_list_size);
         if (free_list_size > 0)
            in.read(reinterpret_cast<char *>(&m_free_list[0]), free_list_size * sizeof(size_t));
      }
      while (num_streams-- != 0)
      {
         boost::shared_ptr<std::fstream> pTmpStream( new std::fstream(
                  path.c_str(),
                  std::ios::binary | std::ios::out | std::ios::in));

         if ( ! *pTmpStream )
         {
            throw std::runtime_error("Cannot open output stream for <" + path + ">!");
         }

         m_rstreams.add_resource(pTmpStream);
      }
   }

   ~block_store()
   {
      moost::container::resource_stack< std::fstream >::scoped_resource rstream(m_rstreams);
      rstream->seekp(0);
      rstream->write(reinterpret_cast<const char *>(&m_allocated), sizeof(size_t));
      rstream->seekp(getpos(m_allocated));
      size_t free_list_size = m_free_list.size();
      rstream->write(reinterpret_cast<const char *>(&free_list_size), sizeof(size_t));
      if (free_list_size > 0)
         rstream->write(reinterpret_cast<const char *>(&m_free_list[0]), free_list_size * sizeof(size_t));
      rstream->rdbuf()->pubsync();
   }

   size_t allocated()
   {
      return m_allocated;
   }

   size_t block_size()
   {
      return m_block_size;
   }

};

}} // moost::io

#endif // MOOST_IO_BLOCK_STORE_HPP__