diff --git a/CMakeLists.txt b/CMakeLists.txt index 4995c644c..cbf87eeaf 100755 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -27,6 +27,7 @@ set(storagemanager_SRCS src/Cache.cpp src/SMLogging.cpp src/Downloader.cpp + src/Synchronizer.cpp ) option(TRACE "Enable some tracing output" OFF) diff --git a/src/Cache.cpp b/src/Cache.cpp index 81b8cdcd5..401c0cd4e 100644 --- a/src/Cache.cpp +++ b/src/Cache.cpp @@ -6,6 +6,9 @@ #include #include #include +#include +#include +#include using namespace std; using namespace boost::filesystem; @@ -13,33 +16,52 @@ using namespace boost::filesystem; namespace storagemanager { -Cache::Cache() +Cache::Cache() : currentCacheSize(0) { Config *conf = Config::get(); - - string ssize = conf->getValue("Cache", "cache_size"); - if (ssize.empty()) + logger = SMLogging::get(); + sync = Synchronizer::get(); + + string stmp = conf->getValue("Cache", "cache_size"); + if (stmp.empty()) { - syslog(LOG_CRIT, "Cache/cache_size is not set"); + logger->log(LOG_CRIT, "Cache/cache_size is not set"); throw runtime_error("Please set Cache/cache_size in the storagemanager.cnf file"); } try { - maxCacheSize = stol(ssize); + maxCacheSize = stoul(stmp); } catch (invalid_argument &) { - syslog(LOG_CRIT, "Cache/cache_size is not a number"); + logger->log(LOG_CRIT, "Cache/cache_size is not a number"); throw runtime_error("Please set Cache/cache_size to a number"); } //cout << "Cache got cache size " << maxCacheSize << endl; + stmp = conf->getValue("ObjectStorage", "object_size"); + if (stmp.empty()) + { + logger->log(LOG_CRIT, "ObjectStorage/object_size is not set"); + throw runtime_error("Please set ObjectStorage/object_size in the storagemanager.cnf file"); + } + try + { + objectSize = stoul(stmp); + } + catch (invalid_argument &) + { + logger->log(LOG_CRIT, "ObjectStorage/object_size is not a number"); + throw runtime_error("Please set ObjectStorage/object_size to a number"); + } + prefix = conf->getValue("Cache", "path"); if (prefix.empty()) { - syslog(LOG_CRIT, "Cache/path is not set"); + logger->log(LOG_CRIT, "Cache/path is not set"); throw runtime_error("Please set Cache/path in the storagemanager.cnf file"); } + try { boost::filesystem::create_directories(prefix); @@ -82,6 +104,11 @@ void Cache::read(const vector &keys) // not in the cache, put it in the list to download keysToFetch.push_back(&key); } + + // TODO: get the sizes of the objects to download and make space + // For now using an estimate + makeSpace(keys.size() * objectSize); + s.unlock(); // start downloading the keys to fetch @@ -154,6 +181,10 @@ const boost::filesystem::path & Cache::getCachePath() void Cache::exists(const vector &keys, vector *out) { + out->resize(keys.size()); + boost::unique_lock s(lru_mutex); + for (int i = 0; i < keys.size(); i++) + (*out)[i] = (m_lru.find(keys[i]) == m_lru.end()); } void Cache::newObject(const string &key, size_t size) @@ -164,12 +195,51 @@ void Cache::deletedObject(const string &key, size_t size) { } -void Cache::setCacheSize(size_t size) +void Cache::setMaxCacheSize(size_t size) { } +// call this holding lru_mutex void Cache::makeSpace(size_t size) { + ssize_t thisMuch = currentCacheSize + size - maxCacheSize; + if (thisMuch <= 0) + return; + + struct stat statbuf; + LRU_t::iterator it = lru.begin(); + while (it != lru.end() && thisMuch > 0) + { + if (doNotEvict.find(it) != doNotEvict.end()) + { + ++it; + continue; // it's in the do-not-evict list + } + + boost::filesystem::path cachedFile = prefix / *it; + int err = stat(cachedFile.string().c_str(), &statbuf); + if (err) + { + logger->log(LOG_WARNING, "Downloader: There seems to be a cached file that couldn't be stat'ed: %s", cachedFile.string().c_str()); + ++it; + continue; + } + + /* + TODO: tell Synchronizer that this key will be evicted + delete the file + remove it from our structs + update current size + */ + assert(currentCacheSize >= statbuf.st_size); + currentCacheSize -= statbuf.st_size; + thisMuch -= statbuf.st_size; + sync->flushObject(*it); + boost::filesystem::remove(cachedFile); + LRU_t::iterator toRemove = it++; + lru.erase(toRemove); + m_lru.erase(*toRemove); + } } diff --git a/src/Cache.h b/src/Cache.h index bc608c730..555fa0aa5 100644 --- a/src/Cache.h +++ b/src/Cache.h @@ -3,6 +3,8 @@ #define CACHE_H_ #include "Downloader.h" +#include "SMLogging.h" +#include "Synchronizer.h" #include #include @@ -25,14 +27,21 @@ class Cache : public boost::noncopyable void exists(const std::vector &keys, std::vector *out); void newObject(const std::string &key, size_t size); void deletedObject(const std::string &key, size_t size); - void setCacheSize(size_t size); - void makeSpace(size_t size); - + void setMaxCacheSize(size_t size); + size_t getCurrentCacheSize(); + // test helpers const boost::filesystem::path &getCachePath(); private: boost::filesystem::path prefix; size_t maxCacheSize; + size_t objectSize; + size_t currentCacheSize; + Downloader downloader; + Synchronizer *sync; + SMLogging *logger; + + void makeSpace(size_t size); /* The main cache structures */ // lru owns the string memory for the filenames it manages. m_lru and DNE point to those strings. @@ -83,9 +92,6 @@ class Cache : public boost::noncopyable void addToDNE(const LRU_t::iterator &key); void removeFromDNE(const LRU_t::iterator &key); boost::mutex lru_mutex; // protects the main cache structures & the do-not-evict set - - Downloader downloader; - }; diff --git a/src/CloudStorage.cpp b/src/CloudStorage.cpp index 7ad580e01..17e9acb93 100644 --- a/src/CloudStorage.cpp +++ b/src/CloudStorage.cpp @@ -13,7 +13,7 @@ using namespace std; namespace { boost::mutex m; - storagemanager::CloudStorage *inst; + storagemanager::CloudStorage *inst = NULL; string tolower(const string &s) { @@ -29,10 +29,10 @@ namespace storagemanager { CloudStorage * CloudStorage::get() { - SMLogging* logger = SMLogging::get(); if (inst) return inst; - + + SMLogging* logger = SMLogging::get(); Config *conf = Config::get(); string type = tolower(conf->getValue("ObjectStorage", "service")); boost::mutex::scoped_lock s(m); diff --git a/src/Config.h b/src/Config.h index 8f103c53b..deaae973e 100644 --- a/src/Config.h +++ b/src/Config.h @@ -3,12 +3,12 @@ #define CONFIG_H_ #include -//#include #include #include #include +/* TODO. Need a config change listener impl. */ namespace storagemanager { diff --git a/src/Downloader.cpp b/src/Downloader.cpp index b935aa641..a248aefa6 100644 --- a/src/Downloader.cpp +++ b/src/Downloader.cpp @@ -1,5 +1,6 @@ #include "Downloader.h" #include "Config.h" +#include "SMLogging.h" #include #include #include @@ -18,11 +19,12 @@ Downloader::Downloader() : maxDownloads(0) } catch(invalid_argument) { - // log something + logger->log(LOG_WARNING, "Downloader: Invalid arg for ObjectStorage/max_concurrent_downloads, using default of 20"); } if (maxDownloads == 0) maxDownloads = 20; workers.reset(new ThreadPool(maxDownloads)); + logger = SMLogging::get(); } Downloader::~Downloader() @@ -92,7 +94,11 @@ int Downloader::download(const vector &keys, vector *errnos auto &dl = dls[i]; (*errnos)[i] = dl->dl_errno; if (dl->dl_errno != 0) + { + char buf[80]; + logger->log(LOG_ERR, "Downloader: failed to download %s, got %s", keys[i]->c_str(), strerror_r(dl->dl_errno, buf, 80)); ret = -1; + } } } diff --git a/src/Downloader.h b/src/Downloader.h index 8ddb0f004..bd300d13f 100644 --- a/src/Downloader.h +++ b/src/Downloader.h @@ -3,6 +3,7 @@ #include "ThreadPool.h" #include "CloudStorage.h" +#include "SMLogging.h" #include #include #include @@ -68,6 +69,7 @@ class Downloader boost::mutex &getDownloadMutex(); boost::scoped_ptr workers; CloudStorage *storage; + SMLogging *logger; }; } diff --git a/src/Synchronizer.cpp b/src/Synchronizer.cpp new file mode 100644 index 000000000..aa850cbd6 --- /dev/null +++ b/src/Synchronizer.cpp @@ -0,0 +1,40 @@ + +#include "Synchronizer.h" +#include + +using namespace std; + +namespace +{ + storagemanager::Synchronizer *instance = NULL; + boost::mutex inst_mutex; +} + + +namespace storagemanager +{ + +Synchronizer * Synchronizer::get() +{ + if (instance) + return instance; + boost::unique_lock lock(inst_mutex); + if (instance) + return instance; + instance = new Synchronizer(); + return instance; +} + +Synchronizer::Synchronizer() +{ +} + +Synchronizer::~Synchronizer() +{ +} + +void Synchronizer::flushObject(const string &key) +{ +} + +} diff --git a/src/Synchronizer.h b/src/Synchronizer.h new file mode 100644 index 000000000..bd354da49 --- /dev/null +++ b/src/Synchronizer.h @@ -0,0 +1,24 @@ + +#ifndef SYNCHRONIZER_H_ +#define SYNCHRONIZER_H_ + +#include +#include + +namespace storagemanager +{ + +class Synchronizer : public boost::noncopyable +{ + public: + static Synchronizer *get(); + virtual ~Synchronizer(); + + void flushObject(const std::string &key); + + private: + Synchronizer(); +}; + +} +#endif