Merge branch 'cache-accounting' into develop

CMakeLists.txt

@@ -89,6 +89,7 @@ set(storagemanager_SRCS
 )
 
 option(TRACE "Enable some tracing output" OFF)
+option(LOG_TO_STDERR "Make the logging system also print to stderr" OFF)
 if (TRACE)
     add_definitions(-DSM_TRACE)
 endif()
@@ -98,6 +99,10 @@ if (CMAKE_BUILD_TYPE STREQUAL Debug)
     set(S3_CONFIGURE_OPT --enable-debug)
 endif()
 
+if (LOG_TO_STDERR)
+    add_definitions(-DDEBUG)
+endif()
+
 set(S3API_DIR ${CMAKE_SOURCE_DIR}/libmarias3)
 
 include(ExternalProject)

src/Cache.cpp

@@ -156,6 +156,30 @@ void Cache::populate()
         ++dir;
     }
 }
+// be careful using this!  SM should be idle.  No ongoing reads or writes.
+void Cache::validateCacheSize()
+{
+    boost::unique_lock<boost::mutex> s(lru_mutex);
+    
+    if (!doNotEvict.empty() || !toBeDeleted.empty())
+    {
+        cout << "Not safe to use validateCacheSize() at the moment." << endl;
+        return;
+    }
+    
+    size_t oldSize = currentCacheSize;
+    currentCacheSize = 0;
+    m_lru.clear();
+    lru.clear();
+    populate();
+    
+    if (oldSize != currentCacheSize)
+        logger->log(LOG_DEBUG, "Cache::validateCacheSize(): found a discrepancy.  Actual size is %lld, had %lld.",
+            currentCacheSize, oldSize);
+    else
+        logger->log(LOG_DEBUG, "Cache::validateCacheSize(): Cache size accounting agrees with reality for now.");
+}
+    
 
 #if 0
 /*  Need to simplify this, we keep running into sneaky problems, and I just spotted a couple more.
@@ -250,24 +274,27 @@ void Cache::read(const vector<string> &keys)
         if (mit != m_lru.end())
         {
             addToDNE(mit->lit);
-            lru.splice(lru.end(), lru, mit->lit);
+            lru.splice(lru.end(), lru, mit->lit);   // move them to the back so they are last to pick for eviction
-            // if it's about to be deleted, stop that
-            TBD_t::iterator tbd_it = toBeDeleted.find(mit->lit);
-            if (tbd_it != toBeDeleted.end())
-            {
-                //cout << "Saved one from being deleted" << endl;
-                toBeDeleted.erase(tbd_it);
-            }
         }
         else
-            keysToFetch.push_back(&key);
+        {
+            // There's window where the file has been downloaded but is not yet
+            // added to the lru structs.  However it is in the DNE.  If it is in the DNE, then it is also
+            // in Downloader's map.  So, this thread needs to start the download if it's not in the 
+            // DNE or if there's an existing download that hasn't finished yet.  Starting the download
+            // includes waiting for an existing download to finish, which from this class's pov is the
+            // same thing.
+            if (doNotEvict.find(key) == doNotEvict.end() || downloader.inProgress(key))
+                keysToFetch.push_back(&key);
+            else
+                cout << "Cache: detected and stopped a racey download" << endl;
+            addToDNE(key);
+        }
     }
     if (keysToFetch.empty())
         return;
     
-    assert(s.owns_lock());
     downloader->download(keysToFetch, &dlErrnos, &dlSizes);
-    assert(s.owns_lock());
     
     size_t sum_sizes = 0;
     for (uint i = 0; i < keysToFetch.size(); ++i)
@@ -275,14 +302,22 @@ void Cache::read(const vector<string> &keys)
         // downloads with size 0 didn't actually happen, either because it
         // was a preexisting download (another read() call owns it), or because
         // there was an error downloading it.  Use size == 0 as an indication of
-        // what to add to the cache
+        // what to add to the cache.  Also needs to verify that the file was not deleted,
+        // indicated by existence in doNotEvict.
         if (dlSizes[i] != 0)
         {
-            sum_sizes += dlSizes[i];
+            if (doNotEvict.find(*keysToFetch[i]) != doNotEvict.end())
-            lru.push_back(*keysToFetch[i]);
+            {
-            LRU_t::iterator lit = lru.end();
+                sum_sizes += dlSizes[i];
-            m_lru.insert(--lit);  // I dislike this way of grabbing the last iterator in a list.
+                lru.push_back(*keysToFetch[i]);
-            addToDNE(lit);
+                LRU_t::iterator lit = lru.end();
+                m_lru.insert(--lit);  // I dislike this way of grabbing the last iterator in a list.
+            }
+            else    // it was downloaded, but a deletion happened so we have to toss it
+            {
+                cout << "removing a file that was deleted by another thread during download" << endl;
+                bf::remove(prefix / (*keysToFetch[i]));
+            }
         }
     }
     
@@ -290,7 +325,7 @@ void Cache::read(const vector<string> &keys)
     for (const string &key : keys)
     {
         mit = m_lru.find(key);
-        if (mit != m_lru.end())   // it could have been deleted by deletedObject() or ifExistsThenDelete()
+        if (mit != m_lru.end())    // all of the files exist, just not all of them are 'owned by' this thread.
             lru.splice(lru.end(), lru, mit->lit);
     }
     
@@ -304,9 +339,14 @@ void Cache::doneReading(const vector<string> &keys)
     boost::unique_lock<boost::mutex> s(lru_mutex);
     for (const string &key : keys)
     {
-        const auto &it = m_lru.find(key);
+        removeFromDNE(key);
-        if (it != m_lru.end())
+        // most should be in the map.
-            removeFromDNE(it->lit);
+        // debateable whether it's faster to look up the list iterator and use it
+        // or whether it's faster to bypass that and use strings only.
+        
+        //const auto &it = m_lru.find(key);
+        //if (it != m_lru.end())
+        //    removeFromDNE(it->lit);
     }
     _makeSpace(0);
 }
@@ -319,24 +359,25 @@ void Cache::doneWriting()
 Cache::DNEElement::DNEElement(const LRU_t::iterator &k) : key(k), refCount(1)
 {
 }
-        
+Cache::DNEElement::DNEElement(const string &k) : sKey(k), refCount(1)
-void Cache::addToDNE(const LRU_t::iterator &key)
 {
-    DNEElement e(key);
+}
-    DNE_t::iterator it = doNotEvict.find(e);
+
+void Cache::addToDNE(const DNEElement &key)
+{
+    DNE_t::iterator it = doNotEvict.find(key);
     if (it != doNotEvict.end())
     {
         DNEElement &dnee = const_cast<DNEElement &>(*it);
         ++(dnee.refCount);
     }
     else
-        doNotEvict.insert(e);
+        doNotEvict.insert(key);
 }
         
-void Cache::removeFromDNE(const LRU_t::iterator &key)
+void Cache::removeFromDNE(const DNEElement &key)
 {
-    DNEElement e(key);
+    DNE_t::iterator it = doNotEvict.find(key);
-    DNE_t::iterator it = doNotEvict.find(e);
     if (it == doNotEvict.end())
         return;
     DNEElement &dnee = const_cast<DNEElement &>(*it);
@@ -398,9 +439,9 @@ void Cache::deletedObject(const string &key, size_t size)
     boost::unique_lock<boost::mutex> s(lru_mutex);
     assert(currentCacheSize >= size);
     M_LRU_t::iterator mit = m_lru.find(key);
-    assert(mit != m_lru.end());          // TODO: 5/16/19 - got this assertion using S3 by running test000, then test000 again.
+    assert(mit != m_lru.end());
     
-    // if it's being flushed, let it do the deleting
+    // if it's being flushed, let makeSpace() do the deleting
     if (toBeDeleted.find(mit->lit) == toBeDeleted.end())
     {
         doNotEvict.erase(mit->lit);
@@ -456,9 +497,9 @@ void Cache::_makeSpace(size_t size)
             return;
         }
         
-        
+        if (!bf::exists(prefix / *it))
-        bf::path cachedFile = prefix / *it;
+            cout << prefix / *it << " doesn't exist, WTF?" << endl;  // ran into this a couple times, still happens as of commit 948ee1aa5
-        assert(bf::exists(cachedFile));
+        assert(bf::exists(prefix / *it));
         /* 
             tell Synchronizer that this key will be evicted
             delete the file
@@ -469,10 +510,12 @@ void Cache::_makeSpace(size_t size)
         //logger->log(LOG_WARNING, "Cache:  flushing!");
         toBeDeleted.insert(it);
 
+        string key = *it;    // need to make a copy; it could get changed after unlocking.
+        
         lru_mutex.unlock();
         try 
         {
-            Synchronizer::get()->flushObject(*it);
+            Synchronizer::get()->flushObject(key);
         }
         catch (...)
         {
@@ -483,15 +526,12 @@ void Cache::_makeSpace(size_t size)
         }
         lru_mutex.lock();
         
-        TBD_t::iterator tbd_it = toBeDeleted.find(it);
+        // check doNotEvict again in case this object is now being read
-        if (tbd_it != toBeDeleted.end())
+        if (doNotEvict.find(it) == doNotEvict.end())
         {
-            // if it's still in toBeDeleted then it is safe to delete.
+            bf::path cachedFile = prefix / *it;
-            // if read() happened to access it while it was flushing, it will not 
-            // be in that set.
-            cachedFile = prefix / *it;
-            toBeDeleted.erase(tbd_it);
             m_lru.erase(*it);
+            toBeDeleted.erase(it);
             lru.erase(it);
             size_t newSize = bf::file_size(cachedFile);
             replicator->remove(cachedFile, Replicator::LOCAL_ONLY);
@@ -507,6 +547,8 @@ void Cache::_makeSpace(size_t size)
                 thisMuch = 0;
             }
         }
+        else
+            toBeDeleted.erase(it);
     }
 }
 
@@ -557,7 +599,6 @@ int Cache::ifExistsThenDelete(const string &key)
     boost::unique_lock<boost::mutex> s(lru_mutex);
     bool objectExists = false;
    
-    
     auto it = m_lru.find(key);
     if (it != m_lru.end())
     {
@@ -572,7 +613,7 @@ int Cache::ifExistsThenDelete(const string &key)
             return 0;
     }
     bool journalExists = bf::exists(journalPath);
-    assert(objectExists == bf::exists(cachedPath));
+    //assert(objectExists == bf::exists(cachedPath));
     
     size_t objectSize = (objectExists ? bf::file_size(cachedPath) : 0);
     //size_t objectSize = (objectExists ? MetadataFile::getLengthFromKey(key) : 0);
@@ -643,12 +684,16 @@ inline bool Cache::KeyEquals::operator()(const M_LRU_element_t &l1, const M_LRU_
 
 inline size_t Cache::DNEHasher::operator()(const DNEElement &l) const
 {
-    return hash<string>()(*(l.key));
+    return (l.sKey.empty() ? hash<string>()(*(l.key)) : hash<string>()(l.sKey));
 }
 
 inline bool Cache::DNEEquals::operator()(const DNEElement &l1, const DNEElement &l2) const
 {
-    return (*(l1.key) == *(l2.key));
+    const string *s1, *s2;
+    s1 = l1.sKey.empty() ? &(*(l1.key)) : &(l1.sKey);
+    s2 = l2.sKey.empty() ? &(*(l2.key)) : &(l2.sKey);
+    
+    return (*s1 == *s2);
 }
 
 inline bool Cache::TBDLess::operator()(const LRU_t::iterator &i1, const LRU_t::iterator &i2) const

src/Cache.h

@@ -53,13 +53,14 @@ class Cache : public boost::noncopyable
         size_t getCurrentCacheSize() const;
         size_t getCurrentCacheElementCount() const;
         size_t getMaxCacheSize() const;
+        void shutdown();
 
         // test helpers
         const boost::filesystem::path &getCachePath();
         const boost::filesystem::path &getJournalPath();
         // this will delete everything in the cache and journal paths, and empty all Cache structures.
         void reset();
-        void shutdown();
+        void validateCacheSize();
         
     private:
         Cache();
@@ -106,7 +107,9 @@ class Cache : public boost::noncopyable
         struct DNEElement
         {
             DNEElement(const LRU_t::iterator &);
+            DNEElement(const std::string &);
             LRU_t::iterator key;
+            std::string sKey;
             uint refCount;
         };
         
@@ -122,8 +125,8 @@ class Cache : public boost::noncopyable
         
         typedef std::unordered_set<DNEElement, DNEHasher, DNEEquals> DNE_t;
         DNE_t doNotEvict;
-        void addToDNE(const LRU_t::iterator &key);
+        void addToDNE(const DNEElement &);
-        void removeFromDNE(const LRU_t::iterator &key);
+        void removeFromDNE(const DNEElement &);
         
         // the to-be-deleted set.  Elements removed from the LRU but not yet deleted will be here.
         // Elements are inserted and removed by makeSpace().  If read() references a file that is in this,

src/Downloader.cpp

@@ -186,6 +186,15 @@ void Downloader::download(const vector<const string *> &keys, vector<int> *errno
     }
 }
 
+bool Downloader::inProgress(const string &key)
+{
+    boost::shared_ptr<Download> tmp(new Download(key));
+    auto it = downloads.find(tmp);
+    if (it != downloads.end())
+        return !(*it)->finished;
+    return false;
+}
+
 void Downloader::setDownloadPath(const string &path)
 {
     downloadPath = path;
@@ -197,6 +206,11 @@ Downloader::Download::Download(const string &source, const string &_dlPath, boos
 {
 }
 
+Downloader::Download::Download(const string &source) : 
+    key(source), dl_errno(0), size(0), lock(NULL), finished(false), itRan(false)
+{
+}
+
 Downloader::Download::~Download()
 {
     assert(!itRan || finished);

src/Downloader.h

@@ -28,6 +28,7 @@ class Downloader
         void download(const std::vector<const std::string *> &keys, std::vector<int> *errnos, std::vector<size_t> *sizes);
         void setDownloadPath(const std::string &path);
         void useThisLock(boost::mutex *);
+        bool inProgress(const std::string &);
         
     private:
         uint maxDownloads;
@@ -51,6 +52,7 @@ class Downloader
         struct Download : public ThreadPool::Job
         {
             Download(const std::string &source, const std::string &_dlPath, boost::mutex *_lock);
+            Download(const std::string &source);
             ~Download();
             void operator()();
             boost::filesystem::path dlPath;

src/IOCoordinator.cpp

@@ -150,7 +150,8 @@ ssize_t IOCoordinator::read(const char *_filename, uint8_t *data, off_t offset,
     vector<metadataObject> relevants = meta.metadataRead(offset, length);
     map<string, int> journalFDs, objectFDs;
     map<string, string> keyToJournalName, keyToObjectName;
-    vector<SharedCloser> fdMinders;
+    ScopedCloser fdMinders[relevants.size() * 2];
+    int mindersIndex = 0;
     char buf[80];
     
     // load them into the cache
@@ -174,7 +175,8 @@ ssize_t IOCoordinator::read(const char *_filename, uint8_t *data, off_t offset,
         {
             keyToJournalName[key] = filename;
             journalFDs[key] = fd;
-            fdMinders.push_back(SharedCloser(fd));
+            fdMinders[mindersIndex++].fd = fd;
+            //fdMinders.push_back(SharedCloser(fd));
         }
         else if (errno != ENOENT)
         {
@@ -202,7 +204,8 @@ ssize_t IOCoordinator::read(const char *_filename, uint8_t *data, off_t offset,
         }
         keyToObjectName[key] = filename;
         objectFDs[key] = fd;
-        fdMinders.push_back(SharedCloser(fd));
+        fdMinders[mindersIndex++].fd = fd;
+        //fdMinders.push_back(SharedCloser(fd));
     }
     //fileLock.unlock();
     
@@ -317,7 +320,7 @@ ssize_t IOCoordinator::_write(const char *filename, const uint8_t *data, off_t o
                 // There's another block below that looks similar.  Also similar blocks in append().
                 if ((count + objectOffset) > i->length)
                     metadata.updateEntryLength(i->offset, (count + objectOffset));
-                metadata.writeMetadata(filename);
+                replicator->updateMetadata(filename, metadata);
                 logger->log(LOG_ERR,"IOCoordinator::write(): object failed to complete write, %u of %u bytes written.",count,length);
                 return count;
             }
@@ -327,7 +330,7 @@ ssize_t IOCoordinator::_write(const char *filename, const uint8_t *data, off_t o
 
             cache->newJournalEntry(writeLength+JOURNAL_ENTRY_HEADER_SIZE);
 
-            synchronizer->newJournalEntry(i->key);
+            synchronizer->newJournalEntry(i->key, writeLength+JOURNAL_ENTRY_HEADER_SIZE);
             count += writeLength;
             dataRemaining -= writeLength;
         }
@@ -368,7 +371,7 @@ ssize_t IOCoordinator::_write(const char *filename, const uint8_t *data, off_t o
             // update metadataObject length to reflect what awas actually written
             if ((count + objectOffset) > newObject.length)
                 metadata.updateEntryLength(newObject.offset, (count + objectOffset));
-            metadata.writeMetadata(filename);
+            replicator->updateMetadata(filename, metadata);
             logger->log(LOG_ERR,"IOCoordinator::write(): newObject failed to complete write, %u of %u bytes written.",count,length);
             return count;
             //log error and abort
@@ -382,7 +385,7 @@ ssize_t IOCoordinator::_write(const char *filename, const uint8_t *data, off_t o
     }
     synchronizer->newObjects(newObjectKeys);
 
-    metadata.writeMetadata(filename);
+    replicator->updateMetadata(filename, metadata);
 
     return count;
 }
@@ -432,7 +435,7 @@ ssize_t IOCoordinator::append(const char *_filename, const uint8_t *data, size_t
             if (err <= 0)
             {
                 metadata.updateEntryLength(i->offset, (count + i->length));
-                metadata.writeMetadata(filename);
+                replicator->updateMetadata(filename, metadata);
                 logger->log(LOG_ERR,"IOCoordinator::append(): journal failed to complete write, %u of %u bytes written.",count,length);
                 goto out;
             }
@@ -440,7 +443,7 @@ ssize_t IOCoordinator::append(const char *_filename, const uint8_t *data, size_t
 
             cache->newJournalEntry(writeLength+JOURNAL_ENTRY_HEADER_SIZE);
 
-            synchronizer->newJournalEntry(i->key);
+            synchronizer->newJournalEntry(i->key, writeLength+JOURNAL_ENTRY_HEADER_SIZE);
             count += writeLength;
             dataRemaining -= writeLength;
         }
@@ -469,7 +472,7 @@ ssize_t IOCoordinator::append(const char *_filename, const uint8_t *data, size_t
         {
             // update metadataObject length to reflect what awas actually written
             metadata.updateEntryLength(newObject.offset, (count));
-            metadata.writeMetadata(filename);
+            replicator->updateMetadata(filename, metadata);
             logger->log(LOG_ERR,"IOCoordinator::append(): newObject failed to complete write, %u of %u bytes written.",count,length);
             goto out;
             //log error and abort
@@ -481,11 +484,11 @@ ssize_t IOCoordinator::append(const char *_filename, const uint8_t *data, size_t
         dataRemaining -= writeLength;
     }
     synchronizer->newObjects(newObjectKeys);
-    metadata.writeMetadata(filename);
+    replicator->updateMetadata(filename, metadata);
     
     // had to add this hack to prevent deadlock
 out:
-    // need to release the file lock before calling the cache fcns.
+    // need to release the file lock before telling Cache that we're done writing.
     lock.unlock();
     cache->doneWriting();
     
@@ -795,12 +798,18 @@ int IOCoordinator::copyFile(const char *_filename1, const char *_filename2)
         return -1;
     }
 
-    vector<string> newJournalEntries;
+    vector<pair<string, size_t> > newJournalEntries;
     ScopedReadLock lock(this, filename1);
+    ScopedWriteLock lock2(this, filename2);
     MetadataFile meta1(metaFile1);
-    MetadataFile meta2(metaFile2);
+    MetadataFile meta2(metaFile2.string().c_str(), MetadataFile::no_create_t());
     vector<metadataObject> objects = meta1.metadataRead(0, meta1.getLength());
     
+    if (meta2.exists()) {
+        cout << "copyFile: overwriting a file" << endl;
+        meta2.removeAllEntries();
+    }
+    
     // TODO.  I dislike large try-catch blocks, and large loops.  Maybe a little refactoring is in order.
     try 
     {
@@ -844,8 +853,9 @@ int IOCoordinator::copyFile(const char *_filename1, const char *_filename2)
                 try 
                 {
                     bf::copy_file(journalFile, newJournalFile);
-                    cache->newJournalEntry(bf::file_size(newJournalFile));
+                    size_t tmp = bf::file_size(newJournalFile);
-                    newJournalEntries.push_back(newObj.key);
+                    cache->newJournalEntry(tmp);
+                    newJournalEntries.push_back(pair<string, size_t>(newObj.key, tmp));
                 }
                 catch (bf::filesystem_error &e)
                 {
@@ -863,7 +873,7 @@ int IOCoordinator::copyFile(const char *_filename1, const char *_filename2)
             cs->deleteObject(newObject.key);
         for (auto &jEntry : newJournalEntries)
         {
-            bf::path fullJournalPath = journalPath/(jEntry + ".journal");
+            bf::path fullJournalPath = journalPath/(jEntry.first + ".journal");
             cache->deletedJournal(bf::file_size(fullJournalPath));
             bf::remove(fullJournalPath);
         }
@@ -871,10 +881,11 @@ int IOCoordinator::copyFile(const char *_filename1, const char *_filename2)
         return -1;
     }
     lock.unlock();
-    
     replicator->updateMetadata(filename2, meta2);
+    lock2.unlock();
+    
     for (auto &jEntry : newJournalEntries)
-        sync->newJournalEntry(jEntry);
+        sync->newJournalEntry(jEntry.first, jEntry.second);
     return 0;
 }
 
@@ -1080,12 +1091,36 @@ int IOCoordinator::mergeJournalInMem(boost::shared_array<uint8_t> &objData, size
     {
         uint64_t offlen[2];
         int err = ::read(journalFD, &offlen, 16);
-        if (err != 16)   // got EOF
+        if (err == 0)   // got EOF
             break;
+        else if (err < 16)
+        {
+            // punting on this
+            cout << "mergeJournalInMem: failed to read a journal entry header in one attempt.  fixme..." << endl;
+            errno = ENODATA;
+            return -1;
+        }
         
         uint64_t startReadingAt = offlen[0];
         uint64_t lengthOfRead = offlen[1];
 
+        // XXXPAT: Speculative change.  Got mem errors from writing past the end of objData.  The length
+        // in the metadata is shorter than this journal entry, and not because it got crazy values.
+        // I think the explanation is a truncation.
+        if (startReadingAt > len)
+        {
+            //logger->log(LOG_CRIT, "mergeJournalInMem: skipping a theoretically irrelevant journal entry in %s.  "
+            //    "jstart = %llu, max = %llu", journalPath, startReadingAt, len);
+            ::lseek(journalFD, offlen[1], SEEK_CUR);
+            continue;
+        }
+        
+        if (startReadingAt + lengthOfRead > len)
+        {
+            //logger->log(LOG_CRIT, "mergeJournalInMem: possibly bad journal entry in %s.  jStart = %llu, jEnd = %llu, max = %llu",
+            //    journalPath, startReadingAt, startReadingAt + lengthOfRead, len);
+            lengthOfRead = len - startReadingAt;
+        }
         uint count = 0;
         while (count < lengthOfRead)
         {
@@ -1106,6 +1141,8 @@ int IOCoordinator::mergeJournalInMem(boost::shared_array<uint8_t> &objData, size
             }
             count += err;
         }
+        if (lengthOfRead < offlen[1])
+            ::lseek(journalFD, offlen[1] - lengthOfRead, SEEK_CUR);
     }
     return 0;
 }

src/LocalStorage.cpp

@@ -5,6 +5,7 @@
 #include <sys/types.h>
 #include <sys/stat.h>
 #include <fcntl.h>
+#include <time.h>
 #include "LocalStorage.h"
 #include "Config.h"
 
@@ -30,6 +31,22 @@ LocalStorage::LocalStorage()
             throw e;
         }
     }
+    string stmp = Config::get()->getValue("LocalStorage", "fake_latency");
+    if (!stmp.empty() && (stmp[0] == 'Y' || stmp[0] == 'y'))
+    {
+        fakeLatency = true;
+        stmp = Config::get()->getValue("LocalStorage", "max_latency");
+        usecLatencyCap = strtoull(stmp.c_str(), NULL, 10);
+        if (usecLatencyCap == 0)
+        {
+            logger->log(LOG_CRIT, "LocalStorage:  bad value for max_latency");
+            throw runtime_error("LocalStorage:  bad value for max_latency");
+        }
+        r_seed = (uint) ::time(NULL);
+        logger->log(LOG_DEBUG, "LocalStorage:  Will simulate cloud latency of max %llu us", usecLatencyCap);
+    }
+    else
+        fakeLatency = false;
 }
 
 LocalStorage::~LocalStorage()
@@ -41,8 +58,19 @@ const bf::path & LocalStorage::getPrefix() const
     return prefix;
 }
 
+void LocalStorage::addLatency()
+{
+    if (fakeLatency)
+    {
+        uint64_t usec_delay = ((double) rand_r(&r_seed) / (double) RAND_MAX) * usecLatencyCap;
+        ::usleep(usec_delay);
+    }
+}
+
 int LocalStorage::copy(const bf::path &source, const bf::path &dest)
 {
+    addLatency();
+    
     boost::system::error_code err;
     bf::copy_file(source, dest, bf::copy_option::fail_if_exists, err);
     if (err)
@@ -72,6 +100,8 @@ int LocalStorage::getObject(const string &source, const string &dest, size_t *si
 
 int LocalStorage::getObject(const std::string &sourceKey, boost::shared_array<uint8_t> *data, size_t *size)
 {
+    addLatency();
+    
     bf::path source = prefix / sourceKey;
     const char *c_source = source.string().c_str();
     //char buf[80];
@@ -115,6 +145,8 @@ int LocalStorage::putObject(const string &source, const string &dest)
 
 int LocalStorage::putObject(boost::shared_array<uint8_t> data, size_t len, const string &dest)
 {
+    addLatency();
+    
     bf::path destPath = prefix / dest;
     const char *c_dest = destPath.string().c_str();
     //char buf[80];
@@ -155,14 +187,17 @@ int LocalStorage::copyObject(const string &source, const string &dest)
 
 int LocalStorage::deleteObject(const string &key)
 {
-    boost::system::error_code err;
+    addLatency();
     
+    boost::system::error_code err;
     bf::remove(prefix / key, err);
     return 0;
 }
 
 int LocalStorage::exists(const std::string &key, bool *out)
 {
+    addLatency();
+    
     *out = bf::exists(prefix / key);
     return 0;
 }

src/LocalStorage.h

@@ -27,8 +27,13 @@ class LocalStorage : public CloudStorage
 
     private:
         boost::filesystem::path prefix;
-        
         int copy(const boost::filesystem::path &sourceKey, const boost::filesystem::path &destKey);
+        
+        // stuff for faking the latency on cloud ops
+        bool fakeLatency;
+        uint64_t usecLatencyCap;
+        uint r_seed;
+        void addLatency();
 };
 
 }

src/MetadataFile.cpp

@@ -326,6 +326,11 @@ void MetadataFile::removeEntry(off_t offset)
     mObjects.erase(it);
 }
 
+void MetadataFile::removeAllEntries()
+{
+    mObjects.clear();
+}
+
 // There are more efficient ways to do it.  Optimize if necessary.
 void MetadataFile::breakout(const string &key, vector<string> &ret)
 {

src/MetadataFile.h

@@ -56,6 +56,7 @@ class MetadataFile
         metadataObject addMetadataObject(const char *filename, size_t length);
         bool getEntry(off_t offset, const metadataObject **out) const;
         void removeEntry(off_t offset);
+        void removeAllEntries();
         
         // TBD: this may have to go; there may be no use case where only the uuid needs to change.
         static std::string getNewKeyFromOldKey(const std::string &oldKey, size_t length=0);

src/SMLogging.cpp

@@ -47,7 +47,7 @@ void SMLogging::log(int priority,const char *format, ...)
     va_list args2;
     va_copy(args2, args);
     vfprintf(stderr, format, args2);
-    printf("\n");
+    fprintf(stderr, "\n");
     #endif
     vsyslog(priority, format, args);
 

src/Synchronizer.cpp

@@ -59,6 +59,8 @@ Synchronizer::Synchronizer() : maxUploads(0)
     threadPool.reset(new ThreadPool());
     threadPool->setMaxThreads(maxUploads);
     die = false;
+    uncommittedJournalSize = 0;
+    journalSizeThreshold = cache->getMaxCacheSize() / 2;
     syncThread = boost::thread([this] () { this->periodicSync(); });
 }
 
@@ -82,8 +84,9 @@ enum OpFlags
     NEW_OBJECT = 0x4,
 };
 
-void Synchronizer::_newJournalEntry(const string &key)
+void Synchronizer::_newJournalEntry(const string &key, size_t size)
 {
+    uncommittedJournalSize += size;
     auto it = pendingOps.find(key);
     if (it != pendingOps.end())
     {
@@ -94,17 +97,29 @@ void Synchronizer::_newJournalEntry(const string &key)
     pendingOps[key] = boost::shared_ptr<PendingOps>(new PendingOps(JOURNAL));
 }
     
-void Synchronizer::newJournalEntry(const string &key)
+void Synchronizer::newJournalEntry(const string &key, size_t size)
 {
     boost::unique_lock<boost::mutex> s(mutex);
-    _newJournalEntry(key);   
+    _newJournalEntry(key, size);
+    if (uncommittedJournalSize > journalSizeThreshold)
+    {
+        uncommittedJournalSize = 0;
+        s.unlock();
+        forceFlush();
+    }
 }
 
-void Synchronizer::newJournalEntries(const vector<string> &keys)
+void Synchronizer::newJournalEntries(const vector<pair<string, size_t> > &keys)
 {
     boost::unique_lock<boost::mutex> s(mutex);
-    for (const string &key : keys)
+    for (auto &keysize : keys)
-        _newJournalEntry(key);
+        _newJournalEntry(keysize.first, keysize.second);
+    if (uncommittedJournalSize > journalSizeThreshold)
+    {
+        uncommittedJournalSize = 0;
+        s.unlock();
+        forceFlush();
+    }
 }
 
 void Synchronizer::newObjects(const vector<string> &keys)
@@ -228,6 +243,7 @@ void Synchronizer::periodicSync()
         //    threadPool.currentQueueSize() << endl;
         for (auto &job : pendingOps)
             makeJob(job.first);
+        uncommittedJournalSize = 0;
     }
 }
 
@@ -235,7 +251,6 @@ void Synchronizer::forceFlush()
 {
     boost::unique_lock<boost::mutex> lock(mutex);
     syncThread.interrupt();
-    lock.unlock();
 }
 
 void Synchronizer::makeJob(const string &key)
@@ -359,12 +374,13 @@ void Synchronizer::synchronize(const string &sourceFile, list<string>::iterator
     
     const metadataObject *mdEntry;
     bool entryExists = md.getEntry(MetadataFile::getOffsetFromKey(key), &mdEntry);
-    if (!entryExists)
+    if (!entryExists || key != mdEntry->key)
     {
         logger->log(LOG_DEBUG, "synchronize(): %s does not exist in metadata for %s.  This suggests truncation.", key.c_str(), sourceFile.c_str());
         return;
     }
-    assert(key == mdEntry->key);
+    
+    //assert(key == mdEntry->key);  <-- This could fail b/c of truncation + a write/append before this job runs.
     
     err = cs->exists(key, &exists);
     if (err)
@@ -374,7 +390,7 @@ void Synchronizer::synchronize(const string &sourceFile, list<string>::iterator
         return;
 
     // TODO: should be safe to check with Cache instead of a file existence check
-    exists = bf::exists(cachePath / key);
+    exists = cache->exists(key);
     if (!exists)
     {
         logger->log(LOG_DEBUG, "synchronize(): was told to upload %s but it does not exist locally", key.c_str());
@@ -409,12 +425,12 @@ void Synchronizer::synchronizeWithJournal(const string &sourceFile, list<string>
     
     const metadataObject *mdEntry;
     bool metaExists = md.getEntry(MetadataFile::getOffsetFromKey(key), &mdEntry);
-    if (!metaExists)
+    if (!metaExists || key != mdEntry->key)
     {
         logger->log(LOG_DEBUG, "synchronizeWithJournal(): %s does not exist in metadata for %s.  This suggests truncation.", key.c_str(), sourceFile.c_str());
         return;
     }
-    assert(key == mdEntry->key);
+    //assert(key == mdEntry->key);   <--- I suspect this can happen in a truncate + write situation + a deep sync queue
     
     bf::path oldCachePath = cachePath / key;
     string journalName = (journalPath/ (key + ".journal")).string();
@@ -431,7 +447,7 @@ void Synchronizer::synchronizeWithJournal(const string &sourceFile, list<string>
             throw runtime_error(string("Synchronizer: cs->exists() failed: ") + strerror_r(errno, buf, 80));
         if (!existsOnCloud)
         {
-            if (bf::exists(oldCachePath))
+            if (cache->exists(key))
             {
                 logger->log(LOG_DEBUG, "synchronizeWithJournal(): %s has no journal and does not exist in the cloud, calling "
                     "synchronize() instead.  Need to explain how this happens.", key.c_str());

src/Synchronizer.h

@@ -30,8 +30,8 @@ class Synchronizer : public boost::noncopyable
         
         // these take keys as parameters, not full path names, ex, pass in '12345' not
         // 'cache/12345'.
-        void newJournalEntry(const std::string &key);
+        void newJournalEntry(const std::string &key, size_t len);
-        void newJournalEntries(const std::vector<std::string> &keys);
+        void newJournalEntries(const std::vector<std::pair<std::string, size_t> > &keys);
         void newObjects(const std::vector<std::string> &keys);
         void deletedObjects(const std::vector<std::string> &keys);        
         void flushObject(const std::string &key);
@@ -43,7 +43,7 @@ class Synchronizer : public boost::noncopyable
     private:
         Synchronizer();
         
-        void _newJournalEntry(const std::string &key);
+        void _newJournalEntry(const std::string &key, size_t len);
         void process(std::list<std::string>::iterator key);
         void synchronize(const std::string &sourceFile, std::list<std::string>::iterator &it);
         void synchronizeDelete(const std::string &sourceFile, std::list<std::string>::iterator &it);
@@ -88,6 +88,7 @@ class Synchronizer : public boost::noncopyable
         boost::thread syncThread;
         const boost::chrono::seconds syncInterval = boost::chrono::seconds(10);
         void periodicSync();
+        size_t uncommittedJournalSize, journalSizeThreshold;
         
         SMLogging *logger;
         Cache *cache;

src/Utilities.cpp

@@ -54,8 +54,11 @@ void ScopedWriteLock::unlock()
     }
 }
 
+ScopedCloser::ScopedCloser() : fd(-1) { }
 ScopedCloser::ScopedCloser(int f) : fd(f) { assert(f != -1); }
 ScopedCloser::~ScopedCloser() {
+    if (fd < 0)
+        return;
     int s_errno = errno;
     ::close(fd);
     errno = s_errno; 

src/Utilities.h

@@ -37,6 +37,7 @@ struct ScopedWriteLock
 
 struct ScopedCloser
 {
+    ScopedCloser();
     ScopedCloser(int f);
     ~ScopedCloser();
     int fd;

src/main.cpp

@@ -24,6 +24,7 @@ bool signalCaught = false;
 
 void printCacheUsage(int sig)
 {
+    Cache::get()->validateCacheSize();
     cout << "Current cache size = " << Cache::get()->getCurrentCacheSize() << endl;
     cout << "Cache element count = " << Cache::get()->getCurrentCacheElementCount() << endl;
 }

src/unit_tests.cpp

@@ -1041,7 +1041,7 @@ bool syncTest1()
     assert(!err);
     assert(exists);
     
-    sync->newJournalEntry(key);
+    sync->newJournalEntry(key, 0);
     sync->forceFlush();
     sleep(1);  // let it do what it does
     
@@ -1080,7 +1080,7 @@ bool syncTest1()
     // make the journal again, call sync->newJournalObject()
     makeTestJournal((journalPath / (newKey + ".journal")).string().c_str());
     cache->newJournalEntry(bf::file_size(journalPath / (newKey + ".journal")));
-    sync->newJournalEntry(newKey);
+    sync->newJournalEntry(newKey, 0);
     sync->forceFlush();
     sleep(1);
     

storagemanager.cnf

@@ -20,6 +20,14 @@ bucket = s3-cs-test2
 [LocalStorage]
 path = ${HOME}/storagemanager/fake-cloud
 
+# introduce latency to fake-cloud operations.  Useful for debugging.
+fake_latency = n
+
+# values are randomized between 1 and max_latency in microseconds.
+# values between 30000-50000 roughly simulate observed latency of S3
+# access from an EC2 node.
+max_latency = 50000
+
 [Cache]
 cache_size = 2g
 path = ${HOME}/storagemanager/cache