mabain/src/lock_free.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 <iostream>
  20
  21 #include "lock_free.h"
  22 #include "mabain_consts.h"
  23 #include "error.h"
  24 #include "integer_4b_5b.h"
  25 #include "drm_base.h"
  26 #include "dict_mem.h"
  27
  28 namespace mabain {
  29
  30 LockFree::LockFree()
  31 {
  32     shm_data_ptr = NULL;
  33 }
  34
  35 LockFree::~LockFree()
  36 {
  37 }
  38
  39 void LockFree::LockFreeInit(LockFreeShmData *lock_free_ptr, IndexHeader *hdr, int mode)
  40 {
  41     shm_data_ptr = lock_free_ptr;
  42     header = hdr;
  43     if(mode & CONSTS::ACCESS_MODE_WRITER)
  44     {
  45         // Clear the lock free data
  46         shm_data_ptr->counter.store(0, MEMORY_ORDER_WRITER);
  47         shm_data_ptr->offset.store(MAX_6B_OFFSET, MEMORY_ORDER_WRITER);
  48     }
  49 }
  50
  51 //////////////////////////////////////////////////
  52 // DO NOT CHANGE THE STORE ORDER IN THIS FUNCTION.
  53 //////////////////////////////////////////////////
  54 void LockFree::WriterLockFreeStop()
  55 {
  56     int index = shm_data_ptr->counter % MAX_OFFSET_CACHE;
  57     shm_data_ptr->offset_cache[index].store(shm_data_ptr->offset, MEMORY_ORDER_WRITER);
  58
  59     shm_data_ptr->counter.fetch_add(1, MEMORY_ORDER_WRITER);
  60     shm_data_ptr->offset.store(MAX_6B_OFFSET, MEMORY_ORDER_WRITER);
  61 }
  62
  63 //////////////////////////////////////////////////
  64 // DO NOT CHANGE THE LOAD ORDER IN THIS FUNCTION.
  65 //////////////////////////////////////////////////
  66 int LockFree::ReaderLockFreeStop(const LockFreeData &snapshot, size_t reader_offset,
  67                                  MBData &mbdata)
  68 {
  69     LockFreeData curr;
  70     curr.offset = shm_data_ptr->offset.load(MEMORY_ORDER_READER);
  71     curr.counter = shm_data_ptr->counter.load(MEMORY_ORDER_READER);
  72
  73     if(curr.offset == reader_offset)
  74     {
  75         if(mbdata.options & CONSTS::OPTION_READ_SAVED_EDGE)
  76         {
  77             if(reader_offset == mbdata.edge_ptrs.offset)
  78             {
  79                 mbdata.options &= ~CONSTS::OPTION_READ_SAVED_EDGE;
  80                 return MBError::SUCCESS;
  81             }
  82         }
  83         else
  84         {
  85             mbdata.options |= CONSTS::OPTION_READ_SAVED_EDGE;
  86         }
  87
  88         // Save the edge
  89         switch(header->excep_updating_status)
  90         {
  91             case EXCEP_STATUS_ADD_EDGE:
  92             case EXCEP_STATUS_ADD_DATA_OFF:
  93             case EXCEP_STATUS_ADD_NODE:
  94             case EXCEP_STATUS_REMOVE_EDGE:
  95             case EXCEP_STATUS_RC_NODE:
  96             case EXCEP_STATUS_RC_EDGE_STR:
  97             case EXCEP_STATUS_RC_DATA:
  98                 memcpy(mbdata.edge_ptrs.edge_buff, header->excep_buff, EDGE_SIZE);
  99                 mbdata.edge_ptrs.offset = shm_data_ptr->offset.load(MEMORY_ORDER_READER);
 100                 break;
 101             case EXCEP_STATUS_CLEAR_EDGE:
 102             default:
 103                 memset(mbdata.edge_ptrs.edge_buff, 0, EDGE_SIZE);
 104                 mbdata.edge_ptrs.offset = shm_data_ptr->offset.load(MEMORY_ORDER_READER);
 105                 break;
 106         }
 107         if(mbdata.edge_ptrs.offset == reader_offset)
 108         {
 109             InitTempEdgePtrs(mbdata.edge_ptrs);
 110         }
 111         else
 112         {
 113             mbdata.options &= ~CONSTS::OPTION_READ_SAVED_EDGE;
 114             mbdata.edge_ptrs.offset = MAX_6B_OFFSET;
 115         }
 116         return MBError::TRY_AGAIN;
 117     }
 118
 119     if(mbdata.options & CONSTS::OPTION_READ_SAVED_EDGE)
 120         mbdata.options &= ~CONSTS::OPTION_READ_SAVED_EDGE;
 121
 122     // Note it is expected that count_diff can overflow.
 123     uint32_t count_diff = curr.counter - snapshot.counter;
 124     if(count_diff == 0)
 125         return MBError::SUCCESS; // Writer was doing nothing. Reader can proceed.
 126     if(count_diff >= MAX_OFFSET_CACHE)
 127         return MBError::TRY_AGAIN; // Cache is overwritten. Have to retry.
 128
 129     for(unsigned i = 0; i < count_diff; i++)
 130     {
 131         int index = (snapshot.counter + i) % MAX_OFFSET_CACHE;
 132         if(reader_offset == shm_data_ptr->offset_cache[index].load(MEMORY_ORDER_READER))
 133             return MBError::TRY_AGAIN;
 134     }
 135
 136     // Need to recheck the counter difference
 137     count_diff = shm_data_ptr->counter.load(MEMORY_ORDER_READER) - snapshot.counter;
 138     if(count_diff >= MAX_OFFSET_CACHE)
 139         return MBError::TRY_AGAIN;
 140
 141     // Writer was modifying different edges. It is safe to for the reader to proceed.
 142     return MBError::SUCCESS;
 143 }
 144
 145 }