mabain/src/iterator.cpp

   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 #include "db.h"
  20 #include "dict.h"
  21 #include "integer_4b_5b.h"
  22 #include "mbt_base.h"
  23
  24 namespace mabain {
  25
  26 typedef struct _iterator_node
  27 {
  28     std::string *key;
  29     uint8_t     *data;
  30     int          data_len;
  31     uint16_t     bucket_index;
  32 } iterator_node;
  33
  34 static void free_iterator_node(void *n)
  35 {
  36     if(n == NULL)
  37         return;
  38
  39     iterator_node *inode = (iterator_node *) n;
  40     if(inode->key != NULL)
  41         delete inode->key;
  42     if(inode->data != NULL)
  43         free(inode->data);
  44
  45     free(inode);
  46 }
  47
  48 static iterator_node* new_iterator_node(const std::string &key, MBData *mbdata)
  49 {
  50     iterator_node *inode = (iterator_node *) malloc(sizeof(*inode));
  51     if(inode == NULL)
  52         throw (int) MBError::NO_MEMORY;
  53
  54     inode->key = new std::string(key);
  55     if(mbdata != NULL)
  56     {
  57         inode->bucket_index = mbdata->bucket_index;
  58         mbdata->TransferValueTo(inode->data, inode->data_len);
  59         if(inode->data == NULL || inode->data_len <= 0)
  60         {
  61             free_iterator_node(inode);
  62             inode = NULL;
  63         }
  64     }
  65     else
  66     {
  67         inode->data = NULL;
  68         inode->data_len = 0;
  69     }
  70
  71     return inode;
  72 }
  73
  74 /////////////////////////////////////////////////////////////////////
  75 // DB iterator
  76 // Example to use DB iterator
  77 // for(DB::iterator iter = db.begin(); iter != db.end(); ++iter) {
  78 //     std::cout << iter.key << "\n";
  79 // }
  80 /////////////////////////////////////////////////////////////////////
  81
  82 const DB::iterator DB::begin(bool check_async_mode, bool rc_mode) const
  83 {
  84     DB::iterator iter = iterator(*this, DB_ITER_STATE_INIT);
  85     if(rc_mode) iter.value.options |= CONSTS::OPTION_RC_MODE;
  86     iter.init(check_async_mode);
  87
  88     return iter;
  89 }
  90
  91 const DB::iterator DB::end() const
  92 {
  93     return iterator(*this, DB_ITER_STATE_DONE);
  94 }
  95
  96 void DB::iterator::iter_obj_init()
  97 {
  98     node_stack = NULL;
  99     kv_per_node = NULL;
 100     lfree = NULL;
 101
 102     if(!(db_ref.GetDBOptions() & CONSTS::ACCESS_MODE_WRITER))
 103     {
 104 #ifdef __LOCK_FREE__
 105         lfree = db_ref.dict->GetLockFreePtr();
 106 #endif
 107     }
 108
 109     if(state == DB_ITER_STATE_INIT)
 110         state = DB_ITER_STATE_MORE;
 111 }
 112
 113 DB::iterator::iterator(const DB &db, int iter_state)
 114                      : db_ref(db), state(iter_state)
 115 {
 116     iter_obj_init();
 117 }
 118
 119 DB::iterator::iterator(const iterator &rhs)
 120                      : db_ref(rhs.db_ref), state(rhs.state)
 121 {
 122     iter_obj_init();
 123 }
 124
 125 DB::iterator::~iterator()
 126 {
 127     if(node_stack != NULL)
 128         delete node_stack;
 129     if(kv_per_node != NULL)
 130         delete kv_per_node;
 131 }
 132
 133 // Initialize the iterator, get the very first key-value pair.
 134 void DB::iterator::init(bool check_async_mode)
 135 {
 136     // Writer in async mode cannot be used for lookup
 137     if(check_async_mode && (db_ref.options & CONSTS::ASYNC_WRITER_MODE))
 138     {
 139         state = DB_ITER_STATE_DONE;
 140         return;
 141     }
 142
 143     node_stack = new MBlsq(free_iterator_node);
 144     kv_per_node = new MBlsq(free_iterator_node);
 145
 146     load_kv_for_node("");
 147     if(next() == NULL)
 148         state = DB_ITER_STATE_DONE;
 149 }
 150
 151 // Initialize the iterator, but do not get the first key-value pair.
 152 // This is used for resource collection.
 153 int DB::iterator::init_no_next()
 154 {
 155     node_stack = new MBlsq(NULL);
 156     kv_per_node = NULL;
 157
 158     int rval = db_ref.dict->ReadRootNode(node_buff, edge_ptrs, match, value);
 159     if(rval != MBError::SUCCESS)
 160         state = DB_ITER_STATE_DONE;
 161     return rval;
 162 }
 163
 164 const DB::iterator& DB::iterator::operator++()
 165 {
 166     if(next() == NULL)
 167         state = DB_ITER_STATE_DONE;
 168
 169     return *this;
 170 }
 171
 172 // This overloaded operator should only be used for iterator state check.
 173 bool DB::iterator::operator!=(const iterator &rhs)
 174 {
 175     return state != rhs.state;
 176 }
 177
 178 int DB::iterator::get_node_offset(const std::string &node_key,
 179                                   size_t &parent_edge_off,
 180                                   size_t &node_offset)
 181 {
 182     int rval;
 183
 184     node_offset = 0;
 185     value.options |= CONSTS::OPTION_FIND_AND_STORE_PARENT;
 186     while(true)
 187     {
 188         rval = db_ref.dict->Find((const uint8_t *)node_key.data(),
 189                                  node_key.size(), value);
 190         if(rval != MBError::TRY_AGAIN)
 191             break;
 192         nanosleep((const struct timespec[]){{0, 10L}}, NULL);
 193     }
 194
 195     if(rval == MBError::IN_DICT)
 196     {
 197         parent_edge_off = edge_ptrs.parent_offset;
 198         node_offset = Get6BInteger(value.edge_ptrs.offset_ptr);
 199         rval = MBError::SUCCESS;
 200     }
 201     return rval;
 202 }
 203
 204 int DB::iterator::load_kvs(const std::string &curr_node_key,
 205                            MBlsq *child_node_list)
 206 {
 207     int rval;
 208     size_t child_node_off;
 209     std::string match_str;
 210     iterator_node *inode;
 211
 212     while(true)
 213     {
 214         if(lfree == NULL)
 215         {
 216             rval = db_ref.dict->ReadNextEdge(node_buff, edge_ptrs, match, value,
 217                                 match_str, child_node_off);
 218         }
 219         else
 220         {
 221 #ifdef __LOCK_FREE__
 222             LockFreeData snapshot;
 223             int lf_ret;
 224             size_t edge_off_prev = edge_ptrs.offset;
 225             lfree->ReaderLockFreeStart(snapshot);
 226 #endif
 227             rval = db_ref.dict->ReadNextEdge(node_buff, edge_ptrs, match, value,
 228                                 match_str, child_node_off);
 229 #ifdef __LOCK_FREE__
 230             lf_ret = lfree->ReaderLockFreeStop(snapshot, edge_off_prev, value);
 231             if(lf_ret == MBError::TRY_AGAIN)
 232                 return lf_ret;
 233 #endif
 234         }
 235
 236         if(rval != MBError::SUCCESS)
 237             break;
 238
 239         match_str = curr_node_key + match_str;
 240         if(child_node_off > 0)
 241         {
 242             inode = new_iterator_node(match_str, NULL);
 243             if(inode != NULL)
 244             {
 245                 rval = child_node_list->AddToTail(inode);
 246                 if(rval != MBError::SUCCESS)
 247                 {
 248                     free_iterator_node(inode);
 249                     return rval;
 250                 }
 251             }
 252         }
 253
 254         if(match != MATCH_NONE)
 255         {
 256             inode = new_iterator_node(match_str, &value);
 257             if(inode != NULL)
 258             {
 259                 rval = kv_per_node->AddToTail(inode);
 260                 if(rval != MBError::SUCCESS)
 261                 {
 262                     free_iterator_node(inode);
 263                     return rval;
 264                 }
 265             }
 266         }
 267     }
 268
 269     if(rval == MBError::OUT_OF_BOUND)
 270         rval = MBError::SUCCESS;
 271     return rval;
 272 }
 273
 274 int DB::iterator::load_node(const std::string &curr_node_key,
 275                             size_t &parent_edge_off)
 276 {
 277     int rval;
 278
 279     if(curr_node_key.size() == 0)
 280     {
 281          rval = db_ref.dict->ReadRootNode(node_buff, edge_ptrs, match, value);
 282     }
 283     else
 284     {
 285         size_t node_offset;
 286         rval = get_node_offset(curr_node_key, parent_edge_off, node_offset);
 287         if(rval != MBError::SUCCESS)
 288             return rval;
 289         rval = db_ref.dict->ReadNode(node_offset, node_buff, edge_ptrs, match,
 290                                      value, false);
 291     }
 292
 293     return rval;
 294 }
 295
 296 int DB::iterator::load_kv_for_node(const std::string &curr_node_key)
 297 {
 298     int rval;
 299     MBlsq child_node_list(free_iterator_node);
 300     size_t parent_edge_off;
 301
 302     if(lfree == NULL)
 303     {
 304         rval = load_node(curr_node_key, parent_edge_off);
 305         if(rval == MBError::SUCCESS)
 306             rval = load_kvs(curr_node_key, &child_node_list);
 307     }
 308     else
 309     {
 310 #ifdef __LOCK_FREE__
 311         LockFreeData snapshot;
 312         int lf_ret;
 313 #endif
 314         while(true)
 315         {
 316 #ifdef __LOCK_FREE__
 317             lfree->ReaderLockFreeStart(snapshot);
 318 #endif
 319             rval = load_node(curr_node_key, parent_edge_off);
 320             if(rval == MBError::SUCCESS)
 321             {
 322                 rval = load_kvs(curr_node_key, &child_node_list);
 323 #ifdef __LOCK_FREE__
 324                 if(rval == MBError::TRY_AGAIN)
 325                 {
 326                     kv_per_node->Clear();
 327                     child_node_list.Clear();
 328                     continue;
 329                 }
 330 #endif
 331             }
 332 #ifdef __LOCK_FREE__
 333             lf_ret = lfree->ReaderLockFreeStop(snapshot, parent_edge_off, value);
 334             if(lf_ret == MBError::TRY_AGAIN)
 335             {
 336                 kv_per_node->Clear();
 337                 child_node_list.Clear();
 338                 continue;
 339             }
 340 #endif
 341             break;
 342         }
 343     }
 344
 345     if(rval == MBError::SUCCESS)
 346     {
 347         iterator_node *inode;
 348         while((inode = (iterator_node *) child_node_list.RemoveFromHead()))
 349         {
 350             node_stack->AddToHead(inode);
 351         }
 352     }
 353     else
 354     {
 355         std::cerr << "failed to run ietrator: " << MBError::get_error_str(rval) << "\n";
 356         kv_per_node->Clear();
 357         child_node_list.Clear();
 358     }
 359     return rval;
 360 }
 361
 362 // Find next iterator match
 363 DB::iterator* DB::iterator::next()
 364 {
 365     iterator_node *inode;
 366
 367     while(kv_per_node->Count() == 0)
 368     {
 369         inode = (iterator_node *) node_stack->RemoveFromHead();
 370         if(inode == NULL)
 371             return NULL;
 372
 373         int rval = load_kv_for_node(*inode->key);
 374         free_iterator_node(inode);
 375         if(rval != MBError::SUCCESS)
 376             return NULL;
 377     }
 378
 379     if(kv_per_node->Count() > 0)
 380     {
 381         inode = (iterator_node *) kv_per_node->RemoveFromHead();
 382         match = MATCH_NODE_OR_EDGE;
 383         key = *inode->key;
 384         value.TransferValueFrom(inode->data, inode->data_len);
 385         value.bucket_index = inode->bucket_index;
 386         free_iterator_node(inode);
 387         return this;
 388     }
 389
 390     return NULL;
 391 }
 392
 393 // There is no need to perform lock-free check in next_dbt_buffer
 394 // since it can only be called by writer.
 395 bool DB::iterator::next_dbt_buffer(struct _DBTraverseNode *dbt_n)
 396 {
 397     int rval;
 398     size_t node_off;
 399     size_t curr_edge_off;
 400     std::string match_str;
 401
 402     memset(dbt_n, 0, sizeof(*dbt_n));
 403     do {
 404         curr_edge_off = edge_ptrs.offset;
 405         while((rval = db_ref.dict->ReadNextEdge(node_buff, edge_ptrs, match,
 406                       value, match_str, node_off, false)) == MBError::SUCCESS)
 407         {
 408             if(edge_ptrs.len_ptr[0] > LOCAL_EDGE_LEN)
 409             {
 410                 dbt_n->edgestr_offset       = Get5BInteger(edge_ptrs.ptr);
 411                 dbt_n->edgestr_size         = edge_ptrs.len_ptr[0] - 1;
 412                 dbt_n->edgestr_link_offset  = curr_edge_off;
 413                 dbt_n->buffer_type         |= BUFFER_TYPE_EDGE_STR;
 414             }
 415
 416             if(node_off > 0)
 417             {
 418                 dbt_n->node_offset         = node_off;
 419                 dbt_n->node_link_offset    = curr_edge_off + EDGE_NODE_LEADING_POS;
 420                 dbt_n->buffer_type        |= BUFFER_TYPE_NODE;
 421                 db_ref.dict->ReadNodeHeader(node_off, dbt_n->node_size, match, dbt_n->data_offset,
 422                                             dbt_n->data_link_offset);
 423                 if(match == MATCH_NODE)
 424                     dbt_n->buffer_type |= BUFFER_TYPE_DATA;
 425             }
 426             else if(match == MATCH_EDGE)
 427             {
 428                 dbt_n->data_offset      = Get6BInteger(edge_ptrs.offset_ptr);
 429                 dbt_n->data_link_offset = curr_edge_off + EDGE_NODE_LEADING_POS;
 430                 dbt_n->buffer_type     |= BUFFER_TYPE_DATA;
 431             }
 432
 433             if(dbt_n->buffer_type != BUFFER_TYPE_NONE)
 434             {
 435                 dbt_n->edge_offset = curr_edge_off;
 436                 return true;
 437             }
 438
 439             curr_edge_off = edge_ptrs.offset;
 440         }
 441
 442         if(rval == MBError::OUT_OF_BOUND)
 443         {
 444             node_off = (size_t ) node_stack->RemoveIntFromHead();
 445             if(node_off == 0)
 446                 break;
 447             rval = db_ref.dict->ReadNode(node_off, node_buff, edge_ptrs, match,
 448                                          value, false);
 449             if(rval != MBError::SUCCESS)
 450                 throw rval;
 451         }
 452         else
 453         {
 454             throw rval;
 455         }
 456     } while(true);
 457
 458     return false;
 459 }
 460
 461 // Add an node offset to the iterator queue
 462 void DB::iterator::add_node_offset(size_t node_offset)
 463 {
 464     int rval = node_stack->AddIntToHead(node_offset);
 465     if(rval != MBError::SUCCESS)
 466         throw rval;
 467 }
 468
 469 }