diff --git a/CMakeLists.txt b/CMakeLists.txt index d5740c475..f8139d16e 100755 --- a/CMakeLists.txt +++ b/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) diff --git a/src/Cache.cpp b/src/Cache.cpp index 24b2bd3ea..10bbbe631 100755 --- a/src/Cache.cpp +++ b/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 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 &keys) if (mit != m_lru.end()) { addToDNE(mit->lit); - lru.splice(lru.end(), lru, mit->lit); - // 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); - } + lru.splice(lru.end(), lru, mit->lit); // move them to the back so they are last to pick for eviction } 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 &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]; - lru.push_back(*keysToFetch[i]); - LRU_t::iterator lit = lru.end(); - m_lru.insert(--lit); // I dislike this way of grabbing the last iterator in a list. - addToDNE(lit); + if (doNotEvict.find(*keysToFetch[i]) != doNotEvict.end()) + { + sum_sizes += dlSizes[i]; + lru.push_back(*keysToFetch[i]); + 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 &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 &keys) boost::unique_lock s(lru_mutex); for (const string &key : keys) { - const auto &it = m_lru.find(key); - if (it != m_lru.end()) - removeFromDNE(it->lit); + removeFromDNE(key); + // most should be in the map. + // 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) { } - -void Cache::addToDNE(const LRU_t::iterator &key) +Cache::DNEElement::DNEElement(const string &k) : sKey(k), refCount(1) { - 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(*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(e); + DNE_t::iterator it = doNotEvict.find(key); if (it == doNotEvict.end()) return; DNEElement &dnee = const_cast(*it); @@ -398,9 +439,9 @@ void Cache::deletedObject(const string &key, size_t size) boost::unique_lock 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; } - - bf::path cachedFile = prefix / *it; - assert(bf::exists(cachedFile)); + if (!bf::exists(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(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); - if (tbd_it != toBeDeleted.end()) + // check doNotEvict again in case this object is now being read + if (doNotEvict.find(it) == doNotEvict.end()) { - // if it's still in toBeDeleted then it is safe to delete. - // if read() happened to access it while it was flushing, it will not - // be in that set. - cachedFile = prefix / *it; - toBeDeleted.erase(tbd_it); + bf::path cachedFile = prefix / *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 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()(*(l.key)); + return (l.sKey.empty() ? hash()(*(l.key)) : hash()(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 diff --git a/src/Cache.h b/src/Cache.h index 5227a6ece..7735c064a 100755 --- a/src/Cache.h +++ b/src/Cache.h @@ -53,14 +53,15 @@ 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 DNE_t; DNE_t doNotEvict; - void addToDNE(const LRU_t::iterator &key); - void removeFromDNE(const LRU_t::iterator &key); + void addToDNE(const DNEElement &); + 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, diff --git a/src/Downloader.cpp b/src/Downloader.cpp index ae988f0bb..6192c9fe5 100644 --- a/src/Downloader.cpp +++ b/src/Downloader.cpp @@ -186,6 +186,15 @@ void Downloader::download(const vector &keys, vector *errno } } +bool Downloader::inProgress(const string &key) +{ + boost::shared_ptr 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); diff --git a/src/Downloader.h b/src/Downloader.h index 07d82a739..518658de4 100644 --- a/src/Downloader.h +++ b/src/Downloader.h @@ -28,6 +28,7 @@ class Downloader void download(const std::vector &keys, std::vector *errnos, std::vector *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; diff --git a/src/IOCoordinator.cpp b/src/IOCoordinator.cpp index c2f3c6778..afc3ce1cb 100755 --- a/src/IOCoordinator.cpp +++ b/src/IOCoordinator.cpp @@ -150,7 +150,8 @@ ssize_t IOCoordinator::read(const char *_filename, uint8_t *data, off_t offset, vector relevants = meta.metadataRead(offset, length); map journalFDs, objectFDs; map keyToJournalName, keyToObjectName; - vector 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 newJournalEntries; + vector > 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 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)); - newJournalEntries.push_back(newObj.key); + size_t tmp = bf::file_size(newJournalFile); + cache->newJournalEntry(tmp); + newJournalEntries.push_back(pair(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 &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 &objData, size } count += err; } + if (lengthOfRead < offlen[1]) + ::lseek(journalFD, offlen[1] - lengthOfRead, SEEK_CUR); } return 0; } diff --git a/src/LocalStorage.cpp b/src/LocalStorage.cpp index db0b92cb0..3bab4f988 100644 --- a/src/LocalStorage.cpp +++ b/src/LocalStorage.cpp @@ -5,6 +5,7 @@ #include #include #include +#include #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 *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 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; } diff --git a/src/LocalStorage.h b/src/LocalStorage.h index f72e5284a..ec571a639 100644 --- a/src/LocalStorage.h +++ b/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(); }; } diff --git a/src/MetadataFile.cpp b/src/MetadataFile.cpp index 705a3abd0..502a94e0e 100755 --- a/src/MetadataFile.cpp +++ b/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 &ret) { diff --git a/src/MetadataFile.h b/src/MetadataFile.h index 9c69d3626..75dd7b41d 100755 --- a/src/MetadataFile.h +++ b/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); diff --git a/src/SMLogging.cpp b/src/SMLogging.cpp index c3cd0be69..5d2922ac3 100644 --- a/src/SMLogging.cpp +++ b/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); diff --git a/src/Synchronizer.cpp b/src/Synchronizer.cpp index fc7270e78..fbe686cf9 100755 --- a/src/Synchronizer.cpp +++ b/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(new PendingOps(JOURNAL)); } -void Synchronizer::newJournalEntry(const string &key) +void Synchronizer::newJournalEntry(const string &key, size_t size) { boost::unique_lock s(mutex); - _newJournalEntry(key); + _newJournalEntry(key, size); + if (uncommittedJournalSize > journalSizeThreshold) + { + uncommittedJournalSize = 0; + s.unlock(); + forceFlush(); + } } -void Synchronizer::newJournalEntries(const vector &keys) +void Synchronizer::newJournalEntries(const vector > &keys) { boost::unique_lock s(mutex); - for (const string &key : keys) - _newJournalEntry(key); + for (auto &keysize : keys) + _newJournalEntry(keysize.first, keysize.second); + if (uncommittedJournalSize > journalSizeThreshold) + { + uncommittedJournalSize = 0; + s.unlock(); + forceFlush(); + } } void Synchronizer::newObjects(const vector &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 lock(mutex); syncThread.interrupt(); - lock.unlock(); } void Synchronizer::makeJob(const string &key) @@ -359,12 +374,13 @@ void Synchronizer::synchronize(const string &sourceFile, list::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::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 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 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()); diff --git a/src/Synchronizer.h b/src/Synchronizer.h index d4b19ab57..dc72ce343 100755 --- a/src/Synchronizer.h +++ b/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 newJournalEntries(const std::vector &keys); + void newJournalEntry(const std::string &key, size_t len); + void newJournalEntries(const std::vector > &keys); void newObjects(const std::vector &keys); void deletedObjects(const std::vector &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::iterator key); void synchronize(const std::string &sourceFile, std::list::iterator &it); void synchronizeDelete(const std::string &sourceFile, std::list::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; diff --git a/src/Utilities.cpp b/src/Utilities.cpp index 304cb433c..88af9dcd9 100644 --- a/src/Utilities.cpp +++ b/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() { +ScopedCloser::~ScopedCloser() { + if (fd < 0) + return; int s_errno = errno; ::close(fd); errno = s_errno; diff --git a/src/Utilities.h b/src/Utilities.h index f71fc925b..1df8766a4 100644 --- a/src/Utilities.h +++ b/src/Utilities.h @@ -37,6 +37,7 @@ struct ScopedWriteLock struct ScopedCloser { + ScopedCloser(); ScopedCloser(int f); ~ScopedCloser(); int fd; diff --git a/src/main.cpp b/src/main.cpp index f7d974bd1..a02639478 100755 --- a/src/main.cpp +++ b/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; } diff --git a/src/unit_tests.cpp b/src/unit_tests.cpp index 3ceac0d40..b342dbb49 100755 --- a/src/unit_tests.cpp +++ b/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); diff --git a/storagemanager.cnf b/storagemanager.cnf index 6195f992f..26c77ae58 100644 --- a/storagemanager.cnf +++ b/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