cachedDBM.h

Go to the documentation of this file.

/*
  @copyright Russell Standish 2000-2013
  @author Russell Standish
  This file is part of EcoLab

  Open source licensed under the MIT license. See LICENSE for details.
*/

#ifndef CACHEDBM_H
#define  CACHEDBM_H

#include "pack_base.h"
#include "classdesc_access.h"
//#include "pack_stl.h"
//#include "error.h"
#include "TCL_obj_base.h"
#include "TCL_obj_stl.h"
#include "omp_rw_lock.h"

#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include <utility>
#include <limits>

namespace ecolab
{
  class Datum: public classdesc::xdr_pack
  {
  public:
    //  void assign_ptr(void *d, size_t sz) {Realloc(data,0); data=(char*)d; size=sz;}
    void copy_ptr(void *d, size_t sz) {packraw((char*)d,sz);}
    Datum(): classdesc::xdr_pack() {}
    Datum(const Datum& x) {packraw(x.data(),x.size());}
    Datum& operator<<(const Datum& x) {packraw(x.data(),x.size()); return *this;}
    const Datum& operator>>(Datum& x) {x.packraw(data(),size()); return *this;}
    template <class T> 
    Datum& operator<<(const T& x) {pack(*this,"",const_cast<T&>(x)); return *this;}
    template <class T> 
    Datum& operator>>(T& x) {unpack(*this,"",x); return *this;}
    Datum& operator=(const Datum& x) {reseti(); packraw(x.data(),x.size()); return *this;}
    template <class T> Datum& operator=(const T& x){reseti(); pack(*this,"",const_cast<T&>(x)); return *this;}
    template <class T> operator T() {T x;  reseto(); unpack(*this,"",x); return x;}
  };

}

// must be in global namespace due to a classdesc bug
class Db
{
  void *db;   // opaque pointer to database object (be it ndbm or Berkley)
  void *dbc;  // opaque pointer to database cursor (BDB)
  Db(const Db& x);
  void operator=(const Db& x);
public:
  enum rw {read,write}; 
  
  void open(const char* filename,rw readwrite);
  void close();
  void flush(); 
  bool fetch(const ecolab::Datum& key, ecolab::Datum& val) const;
  void store(const ecolab::Datum& key, const ecolab::Datum& val);
  void del(const ecolab::Datum& key);
  bool firstkey(ecolab::Datum& key, ecolab::Datum& val) const;
  bool nextkey(ecolab::Datum& key, ecolab::Datum& val) const;
  bool opened() const {return db!=NULL;}
  Db(const char* f,rw rdwr): db(NULL), dbc(NULL) {open(f,rdwr);}
  ~Db() {close();}
};

namespace ecolab
{
  template <class key, class val> struct base_map: public std::map<key,val>
  {
    void set(const key& k, Datum& v) 
    {v>>std::map<key,val>::operator[](k);}
    val& get(const key& k) {return std::map<key,val>::operator[](k);}
    //virtual val& operator[] (const key& k)=0; //used to access cachedDBM::operator[] in TCL_obj()
    virtual ~base_map() {}
  };

  template <class U, class V>
  struct sortPair
  {
    bool operator()(const std::pair<U,V>& x, const std::pair<U,V>& y) {return x.second<y.second;}
  };

  class limited_set: public std::set<int>
  {
    size_t limit;
    CLASSDESC_ACCESS(limited_set);
  public:
    limited_set(size_t limit): limit(limit) {}
    void insert(int x) {
      if (size() < limit || x>*begin()) std::set<int>::insert(x);
      if (size() >= limit) erase(begin());
    }
  };

  /* make the main class a base class in order to derive a special case for 
     strings */
  template<class key, class val>
  class cachedDBM_base : protected base_map<key,val>
  {
    typedef base_map<key,val> Base;
    shared_ptr<Db> db;
    bool readonly;
    classdesc::string filename;
    typedef std::map<key,size_t> TSMap;
    TSMap timestamp;
    size_t ts;
    RWlock rwl;
    mutable bool last;
  public:
    size_t max_elem;   /* limit number of elements to this value */
    cachedDBM_base(): ts(1), max_elem(std::numeric_limits<int>::max()), 
                      keys(*this) {}
    void init(const char *fname, char mode='w')
    {
      db.reset(new Db((char*)fname, (mode=='w')? Db::write: Db::read)); 
      readonly=mode=='r';
      if (!db->opened()) throw error("DBM file %s open failed",fname);
      filename=fname;
    }
    void Init(TCL_args args) {
      char *fname=args, *mode=args;
      init(fname,mode[0]);
    }
    //  void init(const char *fname, char *mode="w") {init(fname,mode[0]);}
    void close() {if (db) {commit(); db->close(); db.reset();} clear();}
    ~cachedDBM_base() {close();}
    bool opened() const {return bool(db);}
    bool load(const key& k)
    {
      Datum dk, vv; 
      dk<<k;
      write_lock w(rwl);
      if (db && !db->fetch(dk,vv)) 
        {
          this->set(k,vv);
          timestamp[k] = ts++;
          return true;
        }
      return false;
    }

    bool key_exists(const key& k) const
    {
      if (this->count(k)) return true;
      if (db) 
        return const_cast<cachedDBM_base*>(this)->load(k);
      return false;
    }
    val& operator[] (const key& k) 
    {
      read_lock r(rwl);
      if (Base::size()>=max_elem) {commit();} /* do a simple purge of database */
      if (!key_exists(k)) //read data into memory if it exists
        {
          write_lock w(rwl);
          Base::insert(std::make_pair(k,val()));
          timestamp[k]=ts++;
        }
#ifdef _OPENMP
#pragma omp atomic
#endif
      ts++;
      timestamp[k]=ts; //not so important if we pick up a simultaneous timestamp
      return Base::get(k);
    }

    val elem(TCL_args args)
    {
      if (args.count>=2)
        return operator[](args[0].get<key>())=args[1].get<val>();
      else if (args.count==1)
        return operator[](args[0].get<key>());
      else
        return val();
    }

    size_t cacheSize() const {return Base::size();}

    void clear() {Base::clear();}

    void commit()
    {
      write_lock w(rwl);
      typename Base::iterator i, j;
      Datum k, v;
      size_t sum_ts=0;
      for (typename TSMap::const_iterator t=timestamp.begin(); t!=timestamp.end(); ++t)
        sum_ts+=t->second;
      size_t cut_time=sum_ts/2;
      limited_set ls(cut_time);
      for (i=Base::begin(); i!=Base::end(); ) 
        {
          k=i->first; v=i->second;
          if (db)
            {
              if (readonly && db->fetch(k,v))
                {
                  ++i; 
                  continue; // do not erase element not in database
                }
              else if (!readonly  && v.data()) 
                db->store(k,v);
            }
          
          typename TSMap::iterator ts=timestamp.find(i->first);
          j=i; i++; //save current iter for later erase
          // erase map element if timestamp earlier than cut_time.
          if (ts->second<cut_time) 
            {
              Base::erase(j);
              timestamp.erase(ts);
            }
        }
      if (db) db->flush();
      // rebase timestamps to cut_time
      ts-=cut_time;
      for (typename TSMap::iterator t=timestamp.begin(); t!=timestamp.end(); ++t)
        t->second-=cut_time;

    }

    void del(key k)   
    {
      if (db && db->opened())
        {
          write_lock w(rwl);
          Datum dk; 
          dk=k; 
          db->del(dk);
        }
      Base::erase(k); 
    }
    key firstkey() const
    {
      key k; 
      if (db)
        {
          Datum kk,vv;
          if (!(last=db->firstkey(kk,vv))) 
            kk>>k;
        }
      return k;
    }
    key nextkey() const
    {
      key k; 
      if (db)
        {
          Datum kk,vv;
          if (!(last=db->nextkey(kk,vv))) 
            kk>>k; 
        }
      return k;
    }
    bool eof() const {return last || !db;}
    void pack(classdesc::pack_t& b) {
      if (opened()) {
        commit();
        ::pack(b,"",filename);
        ::pack(b,"",readonly);
        ::pack(b,"",static_cast<Base&>(*this));
      } else {
        classdesc::string nullstring;
        ::pack(b,"",nullstring);
        ::pack(b,"",static_cast<Base&>(*this));
      }
    }
    void unpack(classdesc::pack_t& b) {
      close();
      classdesc::string fname;
      bool readonly;
      ::unpack(b,"",filename);
      if (filename!="") {
        ::unpack(b,"",readonly);
        init(filename.c_str(),readonly? 'r': 'w');
      } 
      ::unpack(b,"",static_cast<Base&>(*this));
    }

    class KeyValueIterator
    {
      shared_ptr<Db> db;
      std::pair<key,val> keyValue; 
      void getKV(bool (Db::*op)(Datum&, Datum&) const)
      {
        Datum k,v;
        if ((db.get()->*op)(k,v))
          db.reset();
        else
          {
            k>>keyValue.first;
            v>>keyValue.second;
          }
      }
    public:
      KeyValueIterator() {}
      KeyValueIterator(const string& fname): db(new Db(fname.c_str(), Db::read)) 
      {getKV(&Db::firstkey);}
      KeyValueIterator& operator++() {getKV(&Db::nextkey); return *this;}
      // iterator comparison is undefined when referring to different
      // databases, and keys are unique within a given database, so we
      // can use comparisons of keys
      bool operator==(const KeyValueIterator& x) const
      // TODO check whether operator< should be used here???
      {return (!db && !x.db) || (db && x.db && keyValue.first==x.keyValue.first);}
      bool operator!=(const KeyValueIterator& x) const
      {return !operator==(x);}
      const std::pair<key, val> operator*() const {return keyValue;}
      const std::pair<key, val>* operator->() const {return &keyValue;}
    };

    KeyValueIterator begin() const {return KeyValueIterator(filename);}
    KeyValueIterator begin() 
    {commit(); return KeyValueIterator(filename);}
    KeyValueIterator end() const {return KeyValueIterator();}

    class KeyIterator: public KeyValueIterator
    {
    public:
      KeyIterator() {}
      KeyIterator(const string& fname): KeyValueIterator(fname) {}
      const key operator*() const {return KeyValueIterator::operator*().first;}
      const key* operator->() const 
      {return &KeyValueIterator::operator*().first;}
    };
     
    struct Keys
    {
      cachedDBM_base& _this;
      Keys(cachedDBM_base& _this): _this(_this) {}
      KeyIterator begin() const {return KeyIterator(_this.filename);}
      KeyIterator begin()  
      {_this.commit(); return KeyIterator(_this.filename);}
      KeyIterator end() const {return KeyIterator();}
    };
    Keys keys;

  };

  template<class key, class val>
  class cachedDBM: public  cachedDBM_base<key,val>
  {
  public:
    cachedDBM(){}
    cachedDBM(const char* f, char mode='w'){this->init(f,mode);}
  };

  /*
    specialisations to handle char * cases
  */

  struct cachedDBM_string: public std::string
  {
    operator const char*() const {return c_str();}
    cachedDBM_string& operator=(const char* s) 
    {std::string::operator=(s); return *this;}
    cachedDBM_string() {}
    cachedDBM_string(const char*x): std::string(x) {}
  };

  template<class val>
  class cachedDBM<char *,val>: public cachedDBM_base<std::string,val>
  {
  public: 
    cachedDBM(){}
    cachedDBM(const char* f, char mode='w'){this->init(f,mode);}
  };

  template<class key>
  class cachedDBM<key,char *>: public cachedDBM_base<key,cachedDBM_string>
  {
  public: 
    cachedDBM(){}
    cachedDBM(const char* f, char mode='w'){this->init(f,mode);}
  };

  template <>
  class cachedDBM<char *,char *>: public cachedDBM_base<std::string,cachedDBM_string>
  {
  public: 
    cachedDBM(){}
    cachedDBM(const char* f, char mode='w'){init(f,mode);}
  };

#ifdef _CLASSDESC
#pragma omit pack ecolab::Datum
#pragma omit unpack ecolab::Datum
#pragma omit pack ecolab::cachedDBM
#pragma omit unpack ecolab::cachedDBM
#pragma omit pack ecolab::cachedDBM_base
#pragma omit unpack ecolab::cachedDBM_base

#pragma omit pack ecolab::Db
#pragma omit unpack ecolab::Db
#pragma omit pack Db
#pragma omit unpack Db

#endif
}

namespace classdesc_access
{
  namespace cd=classdesc;

  template <>
  struct access_pack<ecolab::Datum>
  {
    void operator()(cd::pack_t& t, const cd::string& d, const ecolab::Datum& x)
    {pack(t,"",x.size()); t.packraw(x.data(),x.size());}
  };

  template <>
  struct access_unpack<ecolab::Datum>
  {
    void operator()(cd::pack_t& t, const cd::string& d, ecolab::Datum& x)
    {
      size_t size; unpack(t,"",size);
      x.packraw(t.data()+t.pos(), size);
      t.seeko(size);
    }
    void operator()(classdesc::pack_t& t, const classdesc::string& d, const ecolab::Datum& x1)
    {
      ecolab::Datum x(x1);
      size_t size; unpack(t,"",size);
      x.packraw(t.data()+t.pos(), size);
      t.seeko(size);
    }
  };

  template <class K, class V>
  struct access_pack<ecolab::cachedDBM<K,V> >
  {
    template <class U>
    void operator()(classdesc::pack_t& b,const classdesc::string& d, U& a) 
    {a.pack(b);}
  };

  template <class K, class V>
  struct access_unpack<ecolab::cachedDBM<K,V> >
  {
    template <class U>
    void operator()(classdesc::unpack_t& b,const classdesc::string& d,U& a) 
    {a.unpack(b);}
  };
}

#ifdef _CLASSDESC
#pragma omit TCL_obj ecolab::base_map
#endif
#include "cachedDBM.cd"
#endif