Two more fixes for anticache:

1. Need to check whether a tuple has already been unevicted.
2. BerkeleyDB does not check the validation of blockID. This will lead to duplicated read an incorrect stats.

Author: linmagit

src/ee/anticache/AntiCacheDB.cpp

@@ -139,9 +139,9 @@ void AntiCacheDB::setStatsSource() {
     //m_stats = new AntiCacheStats(NULL, this);
 }
 
-void AntiCacheDB::removeSingleTupleStats(uint16_t blockId) {
-    m_bytesUnevicted += static_cast<int32_t>( blockSize[blockId] / tupleInBlock[blockId]);
-    evictedTupleInBlock[blockId]--;
+void AntiCacheDB::removeSingleTupleStats(uint16_t blockId, int32_t sign) {
+    m_bytesUnevicted += static_cast<int32_t>( blockSize[blockId] / tupleInBlock[blockId]) * sign;
+    evictedTupleInBlock[blockId] -= sign;
 }
 
 }

src/ee/anticache/AntiCacheDB.h

@@ -213,7 +213,7 @@ class AntiCacheDB {
          * Because of the structure of AnticacheDB we have to have this another
          * function.
          */
-        void removeSingleTupleStats(uint16_t blockId);
+        void removeSingleTupleStats(uint16_t blockId, int32_t sign);
 
         /**
          * Return the AntiCacheID number.

src/ee/anticache/AntiCacheEvictionManager.cpp

@@ -1210,8 +1210,9 @@ bool AntiCacheEvictionManager::readEvictedBlock(PersistentTable *table, int32_t
 
         table->insertUnevictedBlock(table->getUnevictedBlocks(already_unevicted - 1));
         table->insertTupleOffset(tuple_offset);
+        table->insertBlockID(block_id);
 
-        antiCacheDB->removeSingleTupleStats(_block_id);
+        antiCacheDB->removeSingleTupleStats(_block_id, 1);
 
         VOLT_DEBUG("BLOCK %u TUPLE %d - unevicted blocks size is %d",
                 block_id, tuple_offset, (int)table->unevictedBlocksSize());
@@ -1271,6 +1272,7 @@ bool AntiCacheEvictionManager::readEvictedBlock(PersistentTable *table, int32_t
 
         table->insertUnevictedBlock(unevicted_tuples);
         table->insertTupleOffset(tuple_offset);
+        table->insertBlockID(block_id);
 
 
         table->insertUnevictedBlockID(std::pair<int32_t,int32_t>(block_id, table->unevictedBlocksSize()));
@@ -1613,6 +1615,13 @@ bool AntiCacheEvictionManager::mergeUnevictedTuples(PersistentTable *table) {
             VOLT_TRACE("%s", tableNames[j].c_str());
         }
 
+        // Get ACDB for this tuple. That is used for correct stats tuple-merge strategy
+        int32_t block_id = table->getBlockID(i);
+        uint16_t _block_id = (int16_t)(block_id & 0x0000FFFF);
+        int16_t ACID = (int16_t)((block_id & 0x00070000) >> 16);
+        VOLT_DEBUG("block_id: %8x ACID: %d _block_id: %d blocking: %d\n", block_id, ACID, _block_id, (int)blocking);
+        AntiCacheDB* antiCacheDB = m_db_lookup[ACID]; 
+
         int count = 0;
         for (std::vector<std::string>::iterator it = tableNames.begin() ; it != tableNames.end(); ++it){
             PersistentTable *tableInBlock = dynamic_cast<PersistentTable*>(m_engine->getTable(*it));
@@ -1623,7 +1632,12 @@ bool AntiCacheEvictionManager::mergeUnevictedTuples(PersistentTable *table) {
             int64_t bytes_unevicted = 0;
             int tuplesRead = 0;
             if(!table->mergeStrategy()) {
-                bytes_unevicted += tableInBlock->unevictTuple(&in, merge_tuple_offset, merge_tuple_offset, (bool)table->mergeStrategy());
+                int64_t current_unevicted = tableInBlock->unevictTuple(&in, merge_tuple_offset, merge_tuple_offset, (bool)table->mergeStrategy());
+                bytes_unevicted += current_unevicted;
+                if (current_unevicted == 0) {
+                    antiCacheDB->removeSingleTupleStats(_block_id, -1);
+                    //printf("Add back: %u %u\n", ACID, _block_id);
+                }
             } else {
                 for (int j = 0; j < num_tuples_in_block; j++)
                 {
@@ -1707,6 +1721,7 @@ bool AntiCacheEvictionManager::mergeUnevictedTuples(PersistentTable *table) {
     }
     table->clearUnevictedBlocks();
     table->clearMergeTupleOffsets();
+    table->clearBlockIDs();
     VOLT_DEBUG("unevicted blockIDs size %d", static_cast<int>(table->getUnevictedBlockIDs().size()));
     VOLT_DEBUG("unevicted blocks size %d", static_cast<int>(table->unevictedBlocksSize()));
 

src/ee/anticache/BerkeleyAntiCacheDB.cpp

@@ -179,12 +179,16 @@ void BerkeleyAntiCacheDB::writeBlock(const std::string tableName,
 
 
     pushBlockLRU(blockId);
+    m_blockSet.insert(blockId);
 
     delete [] databuf_;
 }
 
 bool BerkeleyAntiCacheDB::validateBlock(uint16_t blockId) {
-    return true;
+    
+    //if (m_blockSet.find(blockId) == m_blockSet.end())
+    //    printf("Berkeley block already unevicted!\n");
+    return m_blockSet.find(blockId) != m_blockSet.end();
 }
 
 AntiCacheBlock* BerkeleyAntiCacheDB::readBlock(uint16_t blockId, bool isMigrate) {
@@ -228,9 +232,11 @@ AntiCacheBlock* BerkeleyAntiCacheDB::readBlock(uint16_t blockId, bool isMigrate)
     }
 
     removeBlockLRU(blockId);
+    m_blockSet.erase(blockId);
 
-    if (!(this->isBlockMerge())) { //i.e. tuple merge
+    if (!(this->isBlockMerge()) && !isMigrate) { //i.e. tuple merge
         pushBlockLRU(blockId); // update block LRU
+        m_blockSet.insert(blockId);
     }
 
     /*uint16_t rm_block = removeBlockLRU(blockId);

src/ee/anticache/BerkeleyAntiCacheDB.h

@@ -8,6 +8,7 @@
 
 #include <map>
 #include <vector>
+#include <set>
 
 using namespace std;
 
@@ -106,6 +107,7 @@ class BerkeleyAntiCacheDB : public AntiCacheDB {
         DbEnv* m_dbEnv;
         Db* m_db;
         Dbt m_value;
+        std::set <uint16_t> m_blockSet;
 };
 
 }

src/ee/storage/persistenttable.cpp

@@ -363,6 +363,12 @@ int32_t PersistentTable::getMergeTupleOffset(int i)
 {
     return m_mergeTupleOffset[i];
 }
+
+int32_t PersistentTable::getBlockID(int i)
+{
+    return m_blockIDs[i];
+}
+
 char* PersistentTable::getUnevictedBlocks(int i)
 {
     return m_unevictedBlocks[i];
@@ -377,6 +383,11 @@ void PersistentTable::insertTupleOffset(int32_t tuple_offset)
     m_mergeTupleOffset.push_back(tuple_offset);
 }
 
+void PersistentTable::insertBlockID(int32_t ACID)
+{
+    m_blockIDs.push_back(ACID);
+}
+
 int32_t PersistentTable::getTuplesRead()
 {
     return m_tuplesRead;
@@ -414,6 +425,11 @@ void PersistentTable::clearMergeTupleOffsets()
     m_mergeTupleOffset.clear();
 }
 
+void PersistentTable::clearBlockIDs()
+{
+    m_blockIDs.clear();
+}
+
 int64_t PersistentTable::unevictTuple(ReferenceSerializeInput * in, int j, int merge_tuple_offset, bool blockMerge){
     TableTuple evicted_tuple = m_evictedTable->tempTuple();
     // get a free tuple and increment the count of tuples current used
@@ -435,6 +451,11 @@ int64_t PersistentTable::unevictTuple(ReferenceSerializeInput * in, int j, int m
     // Note, this goal of the section below is to get a tuple that points to the tuple in the EvictedTable and has the
     // schema of the evicted tuple. However, the lookup has to be done using the schema of the original (unevicted) version
     m_tmpTarget2 = lookupTuple(m_tmpTarget1);       // lookup the tuple in the table
+    //printf("%d\n", m_tmpTarget2.isEvicted());
+    if (!m_tmpTarget2.isEvicted()) {
+        deleteTupleStorage(m_tmpTarget1);
+        return 0;
+    }
     evicted_tuple.move(m_tmpTarget2.address());
     static_cast<EvictedTable*>(m_evictedTable)->deleteEvictedTuple(evicted_tuple);             // delete the EvictedTable tuple
 

src/ee/storage/persistenttable.h

@@ -282,6 +282,7 @@ class PersistentTable : public Table {
     std::map<int32_t, int32_t> getUnevictedBlockIDs();
     std::vector<char*> getUnevictedBlocks();
     int32_t getMergeTupleOffset(int);
+    int32_t getBlockID(int);
     bool mergeStrategy();
     int32_t getTuplesEvicted();
     void setTuplesEvicted(int32_t tuplesEvicted);
@@ -300,6 +301,7 @@ class PersistentTable : public Table {
     bool removeUnevictedBlockID(int32_t blockId);
     void insertUnevictedBlock(char* unevicted_tuples);
     void insertTupleOffset(int32_t tuple_offset);
+    void insertBlockID(int32_t);
     int isAlreadyUnEvicted(int32_t blockId);
     int32_t getTuplesRead();
     void setTuplesRead(int32_t tuplesRead);
@@ -310,6 +312,7 @@ class PersistentTable : public Table {
     int64_t unevictTuple(ReferenceSerializeInput * in, int j, int merge_tuple_offset, bool blockMerge);
     void clearUnevictedBlocks(int i);
     void clearUnevictedBlockIDs();
+    void clearBlockIDs();
     char* getUnevictedBlocks(int i);
     int unevictedBlocksSize();
     std::vector<AntiCacheDB*> allACDBs() const;
@@ -384,6 +387,7 @@ class PersistentTable : public Table {
 //    std::vector<int16_t> m_unevictedBlockIDs;
     std::vector<char*> m_unevictedBlocks;
     std::vector<int32_t> m_mergeTupleOffset; 
+    std::vector<int32_t> m_blockIDs;
 
     std::map<int, int> m_unevictedTuplesPerBlocks;