/home/mhx/git/github/libmoost/include/moost/container/memory_mapped_dataset/dataset.hpp

Go to the documentation of this file.

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

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

#include <boost/archive/text_iarchive.hpp>
#include <boost/archive/text_oarchive.hpp>
#include <boost/serialization/string.hpp>
#include <boost/serialization/map.hpp>
#include <boost/iostreams/device/mapped_file.hpp>
#include <boost/type_traits/is_pod.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/noncopyable.hpp>
#include <boost/cstdint.hpp>

#include "config.hpp"

namespace moost { namespace container {

class memory_mapped_dataset : public boost::noncopyable
{
private:
   static const boost::uint32_t MMD_MAGIC = 0x7473614C;    
   static const boost::uint32_t MMD_VERSION = 1;           

   static const size_t MAP_PAGE_SIZE = 4096;               

   struct mmd_header // must be POD
   {
      boost::uint32_t mmd_magic;              
      boost::uint32_t mmd_version;            
      boost::uint64_t index_offset;           
      boost::uint64_t index_length;           
   };

public:
   class section_info
   {
   private:
      typedef std::map<std::string, std::string> attribute_map_type;

   public:
      section_info()
         : m_offset(0)
         , m_alignment(0)
      {
      }

      section_info(const std::string& type, size_t alignment)
         : m_type(type)
         , m_offset(0)
         , m_alignment(alignment)
      {
      }

      boost::uint64_t offset() const
      {
         return m_offset;
      }

      size_t alignment() const
      {
         return m_alignment;
      }

      void set_offset(boost::uint64_t offset)
      {
         m_offset = offset;
      }

      template <typename T>
      void setattr(const std::string& name, const T& value)
      {
         m_attributes[name] = boost::lexical_cast<std::string>(value);
      }

      template <typename T>
      const T getattr(const std::string& name) const
      {
         attribute_map_type::const_iterator it = m_attributes.find(name);

         if (it == m_attributes.end())
         {
            throw std::runtime_error("no such attribute " + name);
         }

         return boost::lexical_cast<T>(it->second);
      }

      const std::string& type() const
      {
         return m_type;
      }

   private:
      friend class boost::serialization::access;

      template <class Archive>
      void serialize(Archive & ar, const unsigned int /* version */)
      {
         ar & m_type & m_offset & m_attributes;
      }

      std::string m_type;
      boost::uint64_t m_offset;
      const size_t m_alignment;
      attribute_map_type m_attributes;
   };

   typedef std::map<std::string, section_info> section_map_type;

   class writer
   {
   public:
      writer(const std::string& map_file_name, const std::string& dataset_name, boost::uint32_t format_version)
         : m_ofs(map_file_name.c_str(), std::ios::binary | std::ios::trunc)
         , m_dataset_name(dataset_name)
         , m_format_version(format_version)
      {
         if (!m_ofs)
         {
            throw std::runtime_error("failed to open file " + map_file_name);
         }

         if (dataset_name.empty())
         {
            throw std::runtime_error("empty dataset name");
         }

         m_header.mmd_magic = MMD_MAGIC;
         m_header.mmd_version = MMD_VERSION;
         m_header.index_offset = 0;
         m_header.index_length = 0;

         write(m_header);
      }

      ~writer()
      {
         try
         {
            close();
         }
         catch (...)
         {
         }
      }

      void close()
      {
         if (m_ofs.is_open())
         {
            m_header.index_offset = m_ofs.tellp();
            boost::archive::text_oarchive oa(m_ofs);
            oa << m_dataset_name << m_format_version << m_section_map;
            m_header.index_length = static_cast<boost::uint64_t>(m_ofs.tellp()) - m_header.index_offset;
            m_ofs.seekp(0);
            write(m_header);
            m_ofs.close();
         }
      }

      void create_section(const std::string& name, const std::string& type, size_t alignment)
      {
         if (name.empty())
         {
            throw std::runtime_error("invalid empty section name");
         }

         if (type.empty())
         {
            throw std::runtime_error("invalid empty section type");
         }

         if (alignment == 0 || (alignment & (alignment - 1)) != 0)
         {
            throw std::runtime_error("alignment must be a power of 2");
         }

         std::pair<section_map_type::iterator, bool> rv = m_section_map.insert(std::make_pair(name, section_info(type, alignment)));

         if (!rv.second)
         {
            throw std::runtime_error("attempt to create duplicate section " + name);
         }
      }

      void uncreate_section(const std::string& name)
      {
         if (name.empty())
         {
            throw std::runtime_error("invalid empty section name");
         }

         section_map_type::iterator it = m_section_map.find(name);

         if (it == m_section_map.end())
         {
            throw std::runtime_error("attempt to uncreate non-existent section " + name);
         }

         if (it->second.offset())
         {
            throw std::runtime_error("cannot uncreate section " + name + " that has already been written to");
         }

         m_section_map.erase(it);
      }

      void write(const std::string& section, const char *data, size_t size)
      {
         set_active_section(section);
         m_ofs.write(data, size);
      }

      void commit_section(const std::string& section)
      {
         set_active_section(section);
      }

      template <typename T>
      void setattr(const std::string& section, const std::string& attr, const T& value)
      {
         find(section).setattr(attr, value);
      }

   private:
      section_info& find(const std::string& section)
      {
         section_map_type::iterator it = m_section_map.find(section);

         if (it == m_section_map.end())
         {
            throw std::runtime_error("no such section " + section);
         }

         return it->second;
      }

      template <typename T>
      void write(const T& data)
      {
         m_ofs.write(reinterpret_cast<const char *>(&data), sizeof(data));
      }

      void set_active_section(const std::string& section)
      {
         // Yeah, this is slightly inefficient, but it's only being used at
         // dataset creation time.
         if (section != m_active_section)
         {
            section_info& sec = find(section);

            // We have to ensure that only one section is written at a time.
            if (sec.offset() > 0)
            {
               throw std::runtime_error("interleaved write access to section " + section);
            }

            align_stream(sec.alignment());
            sec.set_offset(m_ofs.tellp());
            m_active_section = section;
         }
      }

      void align_stream(size_t alignment)
      {
         while (m_ofs.tellp() % alignment)
         {
            m_ofs.put(0);
         }
      }

      section_map_type m_section_map;
      std::string m_active_section;
      std::ofstream m_ofs;
      const std::string m_dataset_name;
      const boost::uint32_t m_format_version;
      mmd_header m_header;
   };

   memory_mapped_dataset(const std::string& map_file_name,
                         const std::string& dataset_name,
                         boost::uint32_t format_version)
      : m_file(map_file_name)
      , m_format(dataset_name)
   {
      try
      {
         m_map.open(m_file, boost::iostreams::mapped_file::readonly);
      }
      catch (const BOOST_IOSTREAMS_FAILURE& fail)
      {
         // otherwise it's a real pain to figure out which file it's actually complaining about
         throw BOOST_IOSTREAMS_FAILURE(m_file + ": " + fail.what());
      }

      const mmd_header *hdr = data<mmd_header>();

      if (hdr->mmd_magic != MMD_MAGIC)
      {
         throw std::runtime_error(m_file + ": invalid magic");
      }

      if (hdr->mmd_version != MMD_VERSION)
      {
         throw std::runtime_error(m_file + ": unsupported version");
      }

      if (hdr->index_offset == 0 || hdr->index_length == 0)
      {
         throw std::runtime_error(m_file + ": corrupted file");
      }

      std::string indexstr(data<char>(hdr->index_offset, hdr->index_length), hdr->index_length);
      std::istringstream iss(indexstr);

      std::string dset_name;
      boost::uint32_t fmt_version;

      boost::archive::text_iarchive ia(iss);
      ia >> dset_name >> fmt_version >> m_section_map;

      if (dset_name != dataset_name)
      {
         throw std::runtime_error(m_file + ": unexpected format name: " + dset_name + " (expected " + dataset_name + ")");
      }

      if (fmt_version != format_version)
      {
         std::ostringstream oss;
         oss << m_file << ": unsupported format version: " << fmt_version << " (expected " << format_version << ")";
         throw std::runtime_error(oss.str());
      }
   }

   std::string description() const
   {
      return m_format + " (" + m_file + ")";
   }

   const section_info& find(const std::string& section, const std::string& type) const
   {
      section_map_type::const_iterator it = m_section_map.find(section);

      if (it == m_section_map.end())
      {
         throw std::runtime_error(m_file + ": no such section " + section);
      }

      if (it->second.offset() == 0)
      {
         throw std::runtime_error(m_file + ": corrupt section " + section);
      }

      if (it->second.type() != type)
      {
         throw std::runtime_error(m_file + ": invalid section type " + it->second.type() + " (expected " + type + ")");
      }

      return it->second;
   }

   template <typename T>
   const T *data(size_t offset = 0, size_t count = 1) const
   {
      BOOST_STATIC_ASSERT_MSG(boost::is_pod<T>::value, "data<>() called on non-POD type");

      if (offset + count*sizeof(T) > m_map.size())
      {
         throw std::runtime_error(m_file + ": potential attempt to access data beyond end of mapping");
      }

      return reinterpret_cast<const T *>(m_map.const_data() + offset);
   }

   const std::string& filename() const
   {
      return m_file;
   }

   static void warm_cache(const void *beg, const void *end)
   {
      const char *b = reinterpret_cast<const char *>(beg);
      const char *e = reinterpret_cast<const char *>(end);
      char buf[MAP_PAGE_SIZE];
      size_t page_off = static_cast<size_t>(b - static_cast<const char *>(0))%sizeof(buf);

      if (page_off)
      {
         size_t len = std::min(sizeof(buf) - page_off, static_cast<size_t>(e - b));
         std::memcpy(buf, beg, len);
         b += len;
      }

      while (b < e)
      {
         size_t len = std::min(sizeof(buf), static_cast<size_t>(e - b));
         std::memcpy(buf, b, len);
         b += len;
      }
   }

private:
   const std::string m_file;
   const std::string m_format;
   boost::iostreams::mapped_file m_map;
   section_map_type m_section_map;
};

}}

#endif