mabain/src/dict.h

   1 /**
   2  * Copyright (C) 2017 Cisco Inc.
   3  *
   4  * This program is free software: you can redistribute it and/or  modify
   5  * it under the terms of the GNU General Public License, version 2,
   6  * as published by the Free Software Foundation.
   7  *
   8  * This program is distributed in the hope that it will be useful,
   9  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  10  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11  * GNU General Public License for more details.
  12  *
  13  * You should have received a copy of the GNU General Public License
  14  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  15  */
  16
  17 // @author Changxue Deng <chadeng@cisco.com>
  18
  19 #ifndef __DICT_H__
  20 #define __DICT_H__
  21
  22 #include <stdint.h>
  23 #include <string>
  24
  25 #include "drm_base.h"
  26 #include "dict_mem.h"
  27 #include "rollable_file.h"
  28 #include "mb_data.h"
  29 #include "lock_free.h"
  30
  31 namespace mabain {
  32
  33 #ifdef __SHM_QUEUE__
  34 struct _AsyncNode;
  35 typedef struct _AsyncNode AsyncNode;
  36 #endif
  37
  38 // dictionary class
  39 // This is the work horse class for basic db operations (add, find and remove).
  40 class Dict : public DRMBase
  41 {
  42 public:
  43     Dict(const std::string &mbdir, bool init_header, int datasize,
  44          int db_options, size_t memsize_index, size_t memsize_data,
  45          uint32_t block_sz_index, uint32_t block_sz_data,
  46          int max_num_index_blk, int max_num_data_blk,
  47          int64_t entry_per_bucket, uint32_t queue_size);
  48     virtual ~Dict();
  49     void Destroy();
  50
  51     // Called by writer only
  52     int Init(uint32_t id);
  53     // Add key-value pair
  54     int Add(const uint8_t *key, int len, MBData &data, bool overwrite);
  55     // Find value by key
  56     int Find(const uint8_t *key, int len, MBData &data);
  57     // Find value by key using prefix match
  58     int FindPrefix(const uint8_t *key, int len, MBData &data);
  59     // Delete entry by key
  60     int Remove(const uint8_t *key, int len);
  61     // Delete entry by key
  62     int Remove(const uint8_t *key, int len, MBData &data);
  63
  64     // Delete all entries
  65     int RemoveAll();
  66
  67 #ifdef __SHM_QUEUE__
  68     // multiple-process updates using shared memory queue
  69     int  SHMQ_Add(const char *key, int key_len, const char *data, int data_len,
  70                   bool overwrite);
  71     int  SHMQ_Remove(const char *key, int len);
  72     int  SHMQ_RemoveAll();
  73     int  SHMQ_Backup(const char *backup_dir);
  74     int  SHMQ_CollectResource(int64_t m_index_rc_size, int64_t m_data_rc_size,
  75                               int64_t max_dbsz, int64_t max_dbcnt);
  76     bool SHMQ_Busy() const;
  77 #endif
  78
  79     void ReserveData(const uint8_t* buff, int size, size_t &offset);
  80     void WriteData(const uint8_t *buff, unsigned len, size_t offset) const;
  81
  82     // Print dictinary stats
  83     void PrintStats(std::ostream *out_stream) const;
  84     void PrintStats(std::ostream &out_stream) const;
  85     int Status() const;
  86     int64_t Count() const;
  87     size_t GetRootOffset() const;
  88     size_t GetStartDataOffset() const;
  89
  90     DictMem *GetMM() const;
  91
  92     LockFree* GetLockFreePtr();
  93
  94     // Used for DB iterator
  95     int  ReadNextEdge(const uint8_t *node_buff, EdgePtrs &edge_ptrs, int &match,
  96                  MBData &data, std::string &match_str, size_t &node_off,
  97                  bool rd_kv = true) const;
  98     int  ReadNode(size_t node_off, uint8_t *node_buff, EdgePtrs &edge_ptrs,
  99                  int &match, MBData &data, bool rd_kv = true) const;
 100     void ReadNodeHeader(size_t node_off, int &node_size, int &match,
 101                  size_t &data_offset, size_t &data_link_offset);
 102     int  ReadRootNode(uint8_t *node_buff, EdgePtrs &edge_ptrs, int &match,
 103                  MBData &data) const;
 104
 105     // Shared memory objects
 106     int InitShmObjects();
 107     pthread_rwlock_t* GetShmLockPtrs() const;
 108
 109     void UpdateNumReader(int delta) const;
 110     int  UpdateNumWriter(int delta) const;
 111
 112     void ResetSlidingWindow() const;
 113     void Flush() const;
 114     int  ExceptionRecovery();
 115
 116 private:
 117     int Find_Internal(size_t root_off, const uint8_t *key, int len, MBData &data);
 118     int FindPrefix_Internal(size_t root_off, const uint8_t *key, int len, MBData &data);
 119     int ReleaseBuffer(size_t offset);
 120     int UpdateDataBuffer(EdgePtrs &edge_ptrs, bool overwrite, const uint8_t *buff,
 121                          int len, bool &inc_count);
 122     int ReadDataFromEdge(MBData &data, const EdgePtrs &edge_ptrs) const;
 123     int ReadDataFromNode(MBData &data, const uint8_t *node_ptr) const;
 124     int DeleteDataFromEdge(MBData &data, EdgePtrs &edge_ptrs);
 125     int ReadNodeMatch(size_t node_off, int &match, MBData &data) const;
 126 #ifdef __SHM_QUEUE__
 127     int SHMQ_PrepareSlot(AsyncNode *node_ptr) const;
 128     AsyncNode* SHMQ_AcquireSlot() const;
 129 #endif
 130
 131     // DB access permission
 132     int options;
 133     // Memory management
 134     DictMem mm;
 135
 136     // dict status
 137     int status;
 138
 139     LockFree lfree;
 140
 141     size_t reader_rc_off;
 142 #ifdef __SHM_QUEUE__
 143     AsyncNode *queue;
 144 #endif
 145 };
 146
 147 }
 148
 149 #endif