qdp_map_obj_disk.h

Go to the documentation of this file.

// -*- C++ -*-
 
 
#ifndef __qdp_map_obj_disk_h__
#define __qdp_map_obj_disk_h__
 
#include "qdp_map_obj.h"
#include <limits>
#include <unordered_map>
 
namespace QDP
{
 
  namespace MapObjDiskEnv { 
    typedef unsigned int file_version_t;
   
    std::string getFileMagic();

    std::string getMetaData(const std::string& filename);

    bool checkForNewFile(const std::string& filename, std::ios_base::openmode mode);
  };
 
 
 
 
  //----------------------------------------------------------------------------
  template<typename K, typename V>
  class MapObjectDisk : public MapObject<K,V>
  {
  public:
    MapObjectDisk() : file_version(1), state(INIT), level(0) {}

    ~MapObjectDisk();

    void setDebug(int level);

    int getDebug() const {return level;}

    void open(const std::string& file, std::ios_base::openmode mode = std::ios_base::in | std::ios_base::out);

    bool fileExists(const std::string& file) const {
      return (! MapObjDiskEnv::checkForNewFile(file, std::ios_base::in));
    }

    void close();

    int insert(const K& key, const V& val);

    int get(const K& key, V& val) const;
 

    void flush();
 

    bool exist(const K& key) const;

    unsigned int size() const {return static_cast<unsigned long>(src_map.size());}

    void keys(std::vector<K>& keys_) const;
    
    int insertUserdata(const std::string& user_data);
    
    int getUserdata(std::string& user_data) const;
 
  public:
    typedef std::iostream::pos_type pos_type;  // position in buffer
    typedef std::iostream::off_type off_type;  // offset in buffer
 
  private:

    typedef std::array<uint64_t, 2> priv_pos_type_t;

    typedef std::unordered_map<std::string, priv_pos_type_t> MapType_t;

    enum State {INIT, UNCHANGED, MODIFIED};

    State state;

    int level;

    MapObjDiskEnv::file_version_t file_version;
    
    mutable MapType_t src_map;
    
    std::string filename;
    
    std::string user_data;

    mutable BinaryFileReaderWriter streamer;

    priv_pos_type_t convertToPrivate(const pos_type& input) const;

    pos_type convertFromPrivate(const priv_pos_type_t& input) const;

    void openWrite(const std::string& file, std::ios_base::openmode mode);

    void openRead(const std::string& file, std::ios_base::openmode mode);
 
    // Internal Utility: Create/Skip past header
    void writeSkipHeader(void);
    
    priv_pos_type_t readCheckHeader(void);
    
    void writeMapBinary(void);  
    
    void readMapBinary(const priv_pos_type_t& md_start);
    
    void closeWrite(void);
    
    void errorState(const std::string err) const {
      throw err;
    }
  };
 
 
  
  /* ****************** IMPLEMENTATIONS ********************** */
  
  template<typename K, typename V>
  void 
  MapObjectDisk<K,V>::setDebug(int level_)
  {
    level = level_;
  }
 

  template<typename K, typename V>
  typename MapObjectDisk<K,V>::priv_pos_type_t
  MapObjectDisk<K,V>::convertToPrivate(const pos_type& input) const
  {
    const pos_type max_uint64 = pos_type(std::numeric_limits<uint64_t>::max()) + pos_type(1u);
    uint64_t less = uint64_t(input % max_uint64);
    uint64_t more = input / max_uint64;
    // NOTE comment on priv_pos_type_t
    return (QDPUtil::big_endian() ? priv_pos_type_t{more, less} : priv_pos_type_t{less, more});
  }

  template<typename K, typename V>
  typename MapObjectDisk<K,V>::pos_type 
  MapObjectDisk<K,V>::convertFromPrivate(const priv_pos_type_t& input) const
  {
    const pos_type max_uint64 = pos_type(std::numeric_limits<uint64_t>::max()) + pos_type(1u);
    // NOTE comment on priv_pos_type_t
    if (!QDPUtil::big_endian())
      return pos_type(input[0]) + pos_type(input[1]) * max_uint64;
    else
      return pos_type(input[1]) + pos_type(input[0]) * max_uint64;
  }
 

  template<typename K, typename V>
  void 
  MapObjectDisk<K,V>::open(const std::string& file, std::ios_base::openmode mode)
  {
    if ( MapObjDiskEnv::checkForNewFile(file, mode) )
    {
      openWrite(file, mode);
    }
    else
    {
      openRead(file, mode);
    }
  }
 

  template<typename K, typename V>
  void
  MapObjectDisk<K,V>::openWrite(const std::string& file, std::ios_base::openmode mode)
  {
    switch(state) { 
    case INIT: 
    {
      filename = file;
 
      QDPIO::cout << "MapObjectDisk: opening file " << filename
                  << " for writing" << std::endl;
      
      streamer.open(filename, mode);
          
      if (level >= 2) {
        QDPIO::cout << "sizeof(unsigned char) = " << sizeof(unsigned char) << std::endl;
        QDPIO::cout << "sizeof(int) = " << sizeof(int) << std::endl;
        QDPIO::cout << "sizeof(pos_type) = " << sizeof(pos_type) << std::endl;
        QDPIO::cout << "sizeof(priv_pos_type_t) = " << sizeof(priv_pos_type_t) << std::endl;
        QDPIO::cout << "sizeof(file_version)t) = " << sizeof(MapObjDiskEnv::file_version_t) << std::endl;
      }
 
      if (level >= 2) {
        QDPIO::cout << "Writing file magic: len= " << MapObjDiskEnv::getFileMagic().length() << std::endl;
      }
 
      // Write string
      streamer.writeDesc(MapObjDiskEnv::getFileMagic());
      
      if (level >= 2) {
        QDPIO::cout << "Wrote magic. Current Position: " << streamer.currentPosition() << std::endl;
      }
      
      write(streamer, (MapObjDiskEnv::file_version_t)file_version);
    
      if (level >= 2) {
        QDPIO::cout << "Wrote Version. Current Position is: " << streamer.currentPosition() << std::endl;
      }
      
      if (level >= 2) {
        QDPIO::cout << "Writing User Data string=" << user_data << std::endl;
      }
      writeDesc(streamer, user_data);
      
      if (level >= 2) {
        QDPIO::cout << "Wrote User Data string. Current Position is: " << streamer.currentPosition() << std::endl;
      }
      
      priv_pos_type_t dummypos = convertToPrivate(streamer.currentPosition());
    
      if (level >= 2) {
        int user_len = user_data.length();
        QDPInternal::broadcast(user_len);
 
        QDPIO::cout << "Sanity Check 1" << std::endl; ;
        pos_type cur_pos = streamer.currentPosition();
        pos_type exp_pos = 
          MapObjDiskEnv::getFileMagic().length()+sizeof(int)
          +user_len+sizeof(int)
          +sizeof(MapObjDiskEnv::file_version_t);
 
        QDPIO::cout << "cur pos=" << (size_t)(cur_pos) << " expected " << (size_t)(exp_pos) << std::endl;
 
        if ( cur_pos != exp_pos ) {
          QDPIO::cout << "ERROR: Sanity Check 1 failed." << std::endl;
          QDPIO::cout << "cur pos=" << (size_t)(cur_pos) << " expected " << (size_t)(exp_pos) << std::endl;
          QDP_abort(1);
        }
      }
    
      /* Write a dummy link - make room for it */
      streamer.writeArray((char *)&dummypos, sizeof(priv_pos_type_t), 1);
      
      if (level >= 2) {
        QDPIO::cout << "Wrote dummy link: Current Position " << streamer.currentPosition() << std::endl;
        int user_len = user_data.length();
        QDPInternal::broadcast(user_len);
 
        QDPIO::cout << "Sanity Check 2" << std::endl;
        pos_type cur_pos = streamer.currentPosition();
        pos_type exp_pos = 
          MapObjDiskEnv::getFileMagic().length()+sizeof(int)
          +user_len+sizeof(int)
          +sizeof(MapObjDiskEnv::file_version_t)
          +sizeof(priv_pos_type_t);
 
        if ( cur_pos != exp_pos ) {
          QDPIO::cout << "Cur pos = " << (size_t)(cur_pos) << std::endl;
          QDPIO::cout << "Expected: " << (size_t)(exp_pos) << std::endl;
          QDPIO::cout << "ERROR: Sanity Check 2 failed." << std::endl;
          QDP_abort(1);
        }
        QDPIO::cout << "Finished sanity Check 2" << std::endl;
      }
      
      // Advance state machine state
      state = MODIFIED;
      break;      
    }
 
    default:
      errorState("MapOjectDisk: openWrite called from invalid state");
      break;
    }
  
    return;
  }
 
 

  template<typename K, typename V>
  void
  MapObjectDisk<K,V>::openRead(const std::string& file, std::ios_base::openmode mode)
  {  
    switch (state) { 
    case INIT:
    {
      filename = file;
 
      QDPIO::cout << "MapObjectDisk: opening file " << filename
                  << " for reading" << std::endl;
      
      // Open the reader
      streamer.open(filename, mode);
        
      QDPIO::cout << "MapObjectDisk: reading and checking header" << std::endl;
 
      priv_pos_type_t md_start = readCheckHeader();
        
      // Seek to metadata
      QDPIO::cout << "MapObjectDisk: reading key/fileposition data" << std::endl;
        
      /* Read the map in (metadata) */
      readMapBinary(md_start);
        
      /* And we are done */
      state = UNCHANGED;
    }
    break;
    default:
      errorState("MapObjectDisk: openRead() called from invalid state");
      break;
    }
    
    return;
  }
 
 
  
  template<typename K, typename V>
  void
  MapObjectDisk<K,V>::close() 
  {
    switch(state) { 
    case UNCHANGED:
      if( streamer.is_open() ) { 
        streamer.close();
      }
      break;
    case MODIFIED:
      closeWrite(); // This finalizes files for us
      if( streamer.is_open() ) { 
        streamer.close();
      }
      break;
    case INIT:
      break;
    default:
      errorState("close: destructor called from invalid state");
      break;
    }
 
    state = INIT;
  }
  

  template<typename K, typename V>
  MapObjectDisk<K,V>::~MapObjectDisk() 
  {
    close();
  }
  

  template<typename K, typename V>
  void
  MapObjectDisk<K,V>::flush() 
  {
    switch(state) { 
    case MODIFIED:
      closeWrite();  // not optimal
      state = UNCHANGED;
      break;
    case UNCHANGED:
      break;
    case INIT:
      break;
    default:
      break;
    }
  }
  

  template<typename K, typename V>
  void
  MapObjectDisk<K,V>::keys(std::vector<K>& keys_) const 
  {
    if( streamer.is_open() ) 
    {
      typename MapType_t::const_iterator iter;
      for(iter  = src_map.begin();
          iter != src_map.end();
          ++iter) 
      { 
        BinaryBufferReader bin(iter->first);
        K key;
        read(bin, key);
        keys_.push_back(key);
      }
    }
  }
    

  template<typename K, typename V>
  int 
  MapObjectDisk<K,V>::insertUserdata(const std::string& _user_data)
  {
    int ret = 0;
    switch(state) { 
    case INIT:
      user_data = _user_data;
      break;
 
    case UNCHANGED:
    case MODIFIED:
      ret = 1;
      break;
 
    default:
      errorState("MapObjectDisk::insertUserdata() called from invalid state");
      break;
    }
 
    return ret;
  }
 
    
  template<typename K, typename V>
  int
  MapObjectDisk<K,V>::getUserdata(std::string& _user_data) const
  {
    int ret = 0;
 
    switch(state) { 
    case INIT:
    {
      ret = 1;
      break;
    }
    case UNCHANGED:
    case MODIFIED:
    {
      _user_data = user_data;
      break;
    }
    default:
      errorState("MapObjectDisk::getUserdata called from unknown state");
      break;
    }
 
    return ret;
  }
 

  template<typename K, typename V>
  int 
  MapObjectDisk<K,V>::insert(const K& key, const V& val) 
  {
    int ret = 0;
 
    switch (state)  { 
    case MODIFIED :
    case UNCHANGED : {
      //  Find key
      BinaryBufferWriter bin;
      write(bin, key);
      typename MapType_t::const_iterator key_ptr = src_map.find(bin.str());
 
      if (key_ptr != src_map.end()) { 
        // Key does exist
        pos_type wpos = convertFromPrivate(key_ptr->second);
        if (level >= 2) {
          QDPIO::cout << "Found key to update. Position is " << wpos << std::endl;
        }
        
        streamer.seek(wpos);
        
        if (level >= 2) {
          QDPIO::cout << "Sought write position. Current Position: " << streamer.currentPosition() << std::endl;
        }
        streamer.resetChecksum(); // Reset checksum. It gets calculated on write.
        write(streamer, val);
        
        if (level >= 2) {       
          QDPIO::cout << "Wrote value to disk. Current Position: " << streamer.currentPosition() << std::endl;
        }
        write(streamer, streamer.getChecksum()); // Write Checksum
        streamer.flush(); // Sync the file
        
        if (level >= 2) {
          QDPIO::cout << "Wrote checksum " << streamer.getChecksum() << " to disk. Current Position: " << streamer.currentPosition() << std::endl;
        }
 
        // Done
        state = MODIFIED;
      }
      else {
        // Key does not exist
 
        // Make note of current writer position
        pos_type pos = streamer.currentPosition();
      
        // Insert pos into map
        src_map.insert(std::make_pair(bin.str(), convertToPrivate(pos)));
 
        streamer.resetChecksum();
 
        // Add position to the map
        StopWatch swatch;
        swatch.reset();
        swatch.start();
 
        write(streamer, val); // DO write
        swatch.stop();
     
        // Get diagnostics.
        if (level >= 1) {
          pos_type end_pos = streamer.currentPosition();
          double MiBWritten = (double)(end_pos - pos)/(double)(1024*1024);
          double time = swatch.getTimeInSeconds();
 
          QDPIO::cout << " wrote: " << MiBWritten << " MiB. Time: " << time << " sec. Write Bandwidth: " << MiBWritten/time<<std::endl;
        }
 
        if (level >= 2) {
          QDPIO::cout << "Wrote value to disk. Current Position: " << streamer.currentPosition() << std::endl;
        }
 
        write(streamer, streamer.getChecksum()); // Write Checksum
        streamer.flush();
        
        if (level >= 2) {
          QDPIO::cout << "Wrote checksum " << streamer.getChecksum() << " to disk. Current Position: " << streamer.currentPosition() << std::endl;
        }
 
        // Done
        state = MODIFIED;
      }
      break;
    }
    default:
      ret = 1;
      break;
    }
 
    return ret;
  }
 
 

  template<typename K, typename V>
  int 
  MapObjectDisk<K,V>::get(const K& key, V& val) const
  { 
    int ret = 0;
 
    switch(state) { 
    case UNCHANGED: // Deliberate fallthrough
    case MODIFIED: {
      BinaryBufferWriter bin;
      write(bin, key);
      typename MapType_t::const_iterator key_ptr = src_map.find(bin.str());
 
      if (key_ptr != src_map.end())
      {
        // If key exists find file offset
        pos_type pos = convertFromPrivate(key_ptr->second);
 
        // Do the seek and time it 
        StopWatch swatch;
 
        swatch.reset();
        swatch.start();
        streamer.seek(pos);
        swatch.stop();
        double seek_time = swatch.getTimeInSeconds();
 
        // Reset the checkums
        streamer.resetChecksum();
 
        // Grab start pos: We've just seeked it
        pos_type start_pos = pos;
 
        // Time the read
        swatch.reset();
        swatch.start();
        read(streamer, val);
        swatch.stop();
 
        double read_time = swatch.getTimeInSeconds();
        pos_type end_pos = streamer.currentPosition();
 
        // Print data
        if (level >= 1) { 
          double MiBRead = (double)(end_pos - start_pos)/(double)(1024*1024);
          QDPIO::cout << " seek time: " << seek_time 
                      << " sec. read time: " << read_time 
                      << "  " << MiBRead <<" MiB, " << MiBRead/read_time << " MiB/sec" << std::endl;
        }
 
 
        if (level >= 2) { 
          QDPIO::cout << "Read record. Current position: " << streamer.currentPosition() << std::endl;
        }
 
        QDPUtil::n_uint32_t calc_checksum=streamer.getChecksum();
        QDPUtil::n_uint32_t read_checksum;
        read(streamer, read_checksum);
 
        if (level >= 2) {
          QDPIO::cout << " Record checksum: " << read_checksum << "  Current Position: " << streamer.currentPosition() << std::endl;
        }
 
        if( read_checksum != calc_checksum ) { 
          QDPIO::cout << "Mismatched Checksums: Expected: " << calc_checksum << " but read " << read_checksum << std::endl;
          QDP_abort(1);
        }
 
        if (level >= 2) {
          QDPIO::cout << "  Checksum OK!" << std::endl;
        }
      }
      else {
        ret = 1;
      }
      break;
    }
    default:
      ret = 1;
      break;
    }
 
    return ret;
  }
  
  
  template<typename K, typename V>
  bool 
  MapObjectDisk<K,V>::exist(const K& key) const 
  {
    BinaryBufferWriter bin;
    write(bin, key);
    return (src_map.find(bin.str()) == src_map.end()) ? false : true;
  }
  
  
 
  /***************** UTILITY ******************/
 

  template<typename K, typename V>
  void 
  MapObjectDisk<K,V>::writeSkipHeader(void) 
  { 
    switch(state) { 
    case MODIFIED: {
      if ( streamer.is_open() ) 
      { 
        int user_len = user_data.length();
        QDPInternal::broadcast(user_len);
        
        streamer.seek( MapObjDiskEnv::getFileMagic().length() + sizeof(int)
                       + user_len + sizeof(int)
                       + sizeof(MapObjDiskEnv::file_version_t) );
      }
      else { 
        QDPIO::cerr << "Attempting writeSkipHeader, not in write mode" <<std::endl;
        QDP_abort(1);
      }
    }
    break;
    default:
      errorState("MapObjectDisk: writeSkipHeader() called not in MODIFIED state");
      break;
    }
  }
  
  template<typename K, typename V>
  typename MapObjectDisk<K,V>::priv_pos_type_t
  MapObjectDisk<K,V>::readCheckHeader(void) 
  {
    priv_pos_type_t md_position{0, 0};
 
    if( streamer.is_open() ) 
    {
      if (level >= 2) {
        QDPIO::cout << "Rewinding File" << std::endl;
      }
      
      streamer.rewind();
      
      std::string read_magic;
      streamer.readDesc(read_magic);
      
      // Check magic
      if (read_magic != MapObjDiskEnv::getFileMagic()) { 
        QDPIO::cerr << "Magic String Wrong: Expected: " << MapObjDiskEnv::getFileMagic() 
                    << " but read: " << read_magic << std::endl;
        QDP_abort(1);
      }
 
      if (level >= 2) {
        QDPIO::cout << "Read File Magic. Current Position: " << streamer.currentPosition() << std::endl;
      }
      
      MapObjDiskEnv::file_version_t read_version;
      read(streamer, read_version);
      
      if (level >= 2) {
        QDPIO::cout << "Read File Verion. Current Position: " << streamer.currentPosition() << std::endl;
      }
      
      // Check version
      QDPIO::cout << "MapObjectDisk: file has version: " << read_version << std::endl;
      
      QDP::readDesc(streamer, user_data);
      if (level >= 2) {
        QDPIO::cout << "User data. String=" << user_data << ". Current Position: " << streamer.currentPosition() << std::endl;
      }
      
      // Read MD location
      streamer.readArray((char *)&md_position, sizeof(priv_pos_type_t), 1);
 
      if (level >= 2) {
        QDPIO::cout << "Read MD Location. Current position: " << streamer.currentPosition() << std::endl;
      }
      
      if (level >= 2) {
        QDPIO::cout << "Metadata starts at position: " << convertFromPrivate(md_position) << std::endl;
      }
        
    }
    else { 
      QDPIO::cerr << "readCheckHeader needs reader mode to be opened. It is not" << std::endl;
      QDP_abort(1);
    }
 
    return md_position;
  }

  // Private utility function -- no one else should use.
  template<typename K, typename V>
  void
  MapObjectDisk<K,V>::writeMapBinary(void)  
  {
    unsigned int map_size = src_map.size();
 
    streamer.resetChecksum();
    write(streamer, map_size);
    if (level >= 2) {
      QDPIO::cout << "Wrote map size: " << map_size << " entries.  Current position : " << streamer.currentPosition() << std::endl;
    }
    
    typename MapType_t::const_iterator iter;
    for(iter  = src_map.begin();
        iter != src_map.end();
        ++iter) 
    { 
      priv_pos_type_t pos=iter->second;
      
      writeDesc(streamer, iter->first); 
      streamer.writeArray((char *)&pos,sizeof(priv_pos_type_t),1);
      
      if (level >= 2) {
        QDPIO::cout << "Wrote Key/Position pair:  Current Position: " << streamer.currentPosition() << std::endl;
      }
    }
    write(streamer, streamer.getChecksum());
    QDPIO::cout << "Wrote Checksum On Map: " << streamer.getChecksum() << std::endl;
    streamer.flush();
  }
  
  // assume positioned at start of map data
  // Private utility function -- no one else should use.
  template<typename K, typename V>
  void 
  MapObjectDisk<K,V>::readMapBinary(const priv_pos_type_t& md_start)
  {
    streamer.seek(convertFromPrivate(md_start));
    streamer.resetChecksum();
 
    if (level >= 2) {
      QDPIO::cout << "Sought start of metadata. Current position: " << streamer.currentPosition() << std::endl;
    }
    
    unsigned int num_records;
    read(streamer, num_records);
 
    if (level >= 2) {
      QDPIO::cout << "Read num of entries: " << num_records << " records. Current Position: " << streamer.currentPosition() << std::endl;
    }
    
    for(unsigned int i=0; i < num_records; i++) 
    { 
      priv_pos_type_t rpos;
      std::string key_str;
      readDesc(streamer, key_str);
      
      streamer.readArray((char *)&rpos, sizeof(priv_pos_type_t),1);
      
      if (level >= 2) {
        QDPIO::cout << "Read Key/Position pair. Current position: " << streamer.currentPosition() << std::endl;
      }
      // Add position to the map
      src_map.insert(std::make_pair(key_str,rpos));
    }
    QDPUtil::n_uint32_t calc_checksum = streamer.getChecksum();
    QDPUtil::n_uint32_t read_checksum;
    read(streamer, read_checksum);
 
    if (level >= 2) {
      QDPIO::cout << "Read Map checksum: " << read_checksum << "  Current Position: " << streamer.currentPosition();
    }
    if( read_checksum != calc_checksum ) { 
      QDPIO::cout << "Mismatched Checksums: Expected: " << calc_checksum << " but read " << read_checksum << std::endl;
      QDP_abort(1);
    }
 
    if (level >= 2) {
      QDPIO::cout << " Map Checksum OK!" << std::endl;
    }
  }
 
 

  template<typename K, typename V>
  void
  MapObjectDisk<K,V>::closeWrite(void) 
  {
    switch(state) { 
    case MODIFIED:
    {
      if (level >= 2) {
        QDPIO::cout << "Beginning closeWrite: current position: " << streamer.currentPosition() << std::endl;
      }
 
      // Go to end of file
      streamer.seekEnd(0);
 
      // Take note of current position
      priv_pos_type_t metadata_start = convertToPrivate(streamer.currentPosition());
        
      if (level >= 2) {
        QDPIO::cout << "CloseWrite: Metadata starts at position: " << convertFromPrivate(metadata_start) << std::endl;
      }
 
      // Dump metadata
      writeMapBinary();
        
      // Rewind and Skip header 
      streamer.rewind();
      if (level >= 2) {
        QDPIO::cout << "Rewound file. Current Position: " << streamer.currentPosition() << std::endl;
      }
      writeSkipHeader();
      if (level >= 2) {
        QDPIO::cout << "Skipped Header. Current Position: " << streamer.currentPosition() << std::endl;
      }
      // write start position of metadata
      streamer.writeArray((const char *)&metadata_start,sizeof(priv_pos_type_t),1);
        
      if (level >= 2) {
        QDPIO::cout << "Wrote link to metadata. Current Position: " << streamer.currentPosition() << std::endl;
      }
        
      // skip to end and close
      streamer.seekEnd(0);
      streamer.flush();
        
      QDPIO::cout << "MapObjectDisk: Closed file " << filename<< " for write access" <<  std::endl;
    }
    break;
    default:
      errorState("MapObjectDisk: closeWrite() called in an invalid state");
      break;
    }
  }
 
 
} // namespace Chroma
 
#endif