Feature/unsafe wipe all

CMakeLists.txt

@@ -6,10 +6,15 @@ project( fdb5 LANGUAGES C CXX )
 
 set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
-# set(CMAKE_COMPILE_WARNING_AS_ERROR ON)
+
 # add_compile_options(-fsanitize=address)
 # add_compile_options(-fsanitize=address,undefined -fno-omit-frame-pointer)
 # add_link_options(-fsanitize=address)
+
+# add_compile_options(-fsanitize=thread)
+# add_compile_options(-fsanitize=thread,undefined -fno-omit-frame-pointer)
+# add_link_options(-fsanitize=thread)
+
 # set(CMAKE_CXX_FLAGS "-Wall -Wextra -Wno-unused-parameter -Wno-unused-variable -Wno-sign-compare")
 # set(CMAKE_CXX_FLAGS "-Wno-unused-parameter -Wno-unused-variable -Wno-reorder -Wno-sign-compare -Wvla-cxx-extension")
 

src/fdb5/api/FDBFactory.cc

@@ -28,22 +28,8 @@ namespace fdb5 {
 //----------------------------------------------------------------------------------------------------------------------
 
 
-FDBBase::FDBBase(const Config& config, const std::string& name) : name_(name), config_(config) {
-
-    bool writable = config.getBool("writable", true);
-    bool visitable = config.getBool("visitable", true);
-    if (!config.getBool("list", visitable)) {
-        controlIdentifiers_ |= ControlIdentifier::List;
-    }
-    if (!config.getBool("retrieve", visitable)) {
-        controlIdentifiers_ |= ControlIdentifier::Retrieve;
-    }
-    if (!config.getBool("archive", writable)) {
-        controlIdentifiers_ |= ControlIdentifier::Archive;
-    }
-    if (!config.getBool("wipe", writable)) {
-        controlIdentifiers_ |= ControlIdentifier::Wipe;
-    }
+FDBBase::FDBBase(const Config& config, const std::string& name) :
+    name_(name), config_(config), controlIdentifiers_(ControlIdentifiers::parse(config)) {
 
     LOG_DEBUG_LIB(LibFdb5) << "FDBBase: " << config << std::endl;
 }

src/fdb5/api/RemoteFDB.cc

@@ -1,5 +1,6 @@
 #include <cstdlib>
 #include <ctime>
+#include <random>
 
 #include "eckit/config/Resource.h"
 #include "eckit/io/Buffer.h"
@@ -139,7 +140,9 @@ const net::Endpoint& RemoteFDB::storeEndpoint() const {
     if (storesLocalFields_.empty()) {
         throw SeriousBug("Unable to find a store to serve local data");
     }
-    return storesLocalFields_.at(std::rand() % storesLocalFields_.size());
+    static std::mt19937 rd;
+    static std::uniform_int_distribution<size_t> dist(0, storesLocalFields_.size() - 1);
+    return storesLocalFields_.at(dist(rd));
 }
 const net::Endpoint& RemoteFDB::storeEndpoint(const net::Endpoint& fieldLocationEndpoint) const {
     // looking for an alias for the given endpoint
@@ -158,7 +161,7 @@ const net::Endpoint& RemoteFDB::storeEndpoint(const net::Endpoint& fieldLocation
 
 RemoteFDB::RemoteFDB(const Configuration& config, const std::string& name) : LocalFDB(config, name), Client(config) {
 
-    Buffer buf = controlWriteReadResponse(remote::Message::Stores, generateRequestID());
+    Buffer buf = controlWriteReadResponse(Message::Stores, generateRequestID());
     MemoryStream s(buf);
     size_t numStores;
     s >> numStores;
@@ -209,21 +212,18 @@ RemoteFDB::RemoteFDB(const Configuration& config, const std::string& name) : Loc
         fieldLocationEndpoints.push_back("");
     }
 
-    Buffer buf2 = controlWriteReadResponse(remote::Message::Schema, generateRequestID());
+    Buffer buf2 = controlWriteReadResponse(Message::Schema, generateRequestID());
     MemoryStream s2(buf2);
 
     Schema* schema = Reanimator<Schema>::reanimate(s2);
 
     config_.set("stores", stores);
     config_.set("fieldLocationEndpoints", fieldLocationEndpoints);
     config_.overrideSchema(static_cast<std::string>(controlEndpoint()) + "/schema", schema);
+}
 
-    /// @note: We must instantiate the ReadLimiter before any RemoteStores due to their static initialisation.
-    /// @todo: this may change in future.
-    static size_t memoryLimit =
-        Resource<size_t>("$FDB_READ_LIMIT;fdbReadLimit",
-                         config_.userConfig().getUnsigned("limits.read", size_t(1) * 1024 * 1024 * 1024));  // 1GiB
-    ReadLimiter::init(memoryLimit);
+RemoteFDB::~RemoteFDB() {
+    deregister();
 }
 
 // -----------------------------------------------------------------------------------------------------
@@ -248,14 +248,16 @@ auto RemoteFDB::forwardApiCall(const HelperClass& helper, const FDBToolRequest&
 
     // Ensure we have an entry in the message queue before we trigger anything that
     // will result in return messages
-
     uint32_t id = generateRequestID();
-    auto entry = messageQueues_.emplace(id, std::make_shared<MessageQueue>(HelperClass::queueSize()));
-    ASSERT(entry.second);
-    std::shared_ptr<MessageQueue> messageQueue(entry.first->second);
+    std::shared_ptr<MessageQueue> messageQueue;
+    {
+        std::lock_guard<std::mutex> lock(messageMutex_);
+        auto entry = messageQueues_.emplace(id, std::make_shared<MessageQueue>(HelperClass::queueSize()));
+        ASSERT(entry.second);
+        messageQueue = entry.first->second;
+    }
 
     // Encode the request and send it to the server
-
     Buffer encodeBuffer(HelperClass::bufferSize());
     MemoryStream s(encodeBuffer);
     s << request;
@@ -313,64 +315,69 @@ const Configuration& RemoteFDB::clientConfig() const {
     return config();
 }
 
-bool RemoteFDB::handle(remote::Message message, uint32_t requestID) {
+bool RemoteFDB::handle(Message message, uint32_t requestID) {
 
     switch (message) {
         case Message::Complete: {
-
+            std::lock_guard<std::mutex> lock(messageMutex_);
             auto it = messageQueues_.find(requestID);
             if (it == messageQueues_.end()) {
                 return false;
             }
 
             it->second->close();
-            // Remove entry (shared_ptr --> message queue will be destroyed when it
-            // goes out of scope in the worker thread).
+            // Remove entry (shared_ptr --> message queue will be destroyed when it goes out of scope in the worker
+            // thread).
             messageQueues_.erase(it);
             return true;
         }
         case Message::Error: {
-
-            std::ostringstream ss;
-            ss << "RemoteFDB - client id: " << clientId()
-               << " - received an error without error description for requestID " << requestID << std::endl;
-            throw RemoteFDBException(ss.str(), controlEndpoint());
-
-            return false;
+            std::lock_guard<std::mutex> lock(messageMutex_);
+            // Received Error message without error description. Remove the corresponding entry from the message queue
+            // and let the caller know & complain
+            auto it = messageQueues_.find(requestID);
+            if (it != messageQueues_.end()) {
+                it->second->interrupt(
+                    std::make_exception_ptr(RemoteFDBException("no error description provided", controlEndpoint())));
+                // Remove entry (shared_ptr --> message queue will be destroyed when it goes out of scope in the worker
+                // thread).
+                messageQueues_.erase(it);
+            }
+            return true;
         }
         default:
+            Log::error() << *this << " - Received unexpected [message=" << message << ",requestID=" << requestID << "]"
+                         << std::endl;
             return false;
     }
 }
-bool RemoteFDB::handle(remote::Message message, uint32_t requestID, Buffer&& payload) {
+bool RemoteFDB::handle(Message message, uint32_t requestID, Buffer&& payload) {
 
     switch (message) {
         case Message::Blob: {
+            std::lock_guard<std::mutex> lock(messageMutex_);
             auto it = messageQueues_.find(requestID);
             if (it == messageQueues_.end()) {
                 return false;
             }
-
             it->second->emplace(std::move(payload));
             return true;
         }
-
         case Message::Error: {
-
+            std::lock_guard<std::mutex> lock(messageMutex_);
             auto it = messageQueues_.find(requestID);
-            if (it == messageQueues_.end()) {
-                return false;
+            if (it != messageQueues_.end()) {
+                std::string errmsg{static_cast<const char*>(payload.data()), payload.size()};
+                it->second->interrupt(std::make_exception_ptr(RemoteFDBException(errmsg, controlEndpoint())));
+                // Remove entry (shared_ptr --> message queue will be destroyed when it goes out of scope in the worker
+                // thread).
+                messageQueues_.erase(it);
             }
-            std::string msg;
-            msg.resize(payload.size(), ' ');
-            payload.copy(&msg[0], payload.size());
-            it->second->interrupt(std::make_exception_ptr(RemoteFDBException(msg, controlEndpoint())));
-            // Remove entry (shared_ptr --> message queue will be destroyed when it
-            // goes out of scope in the worker thread).
-            messageQueues_.erase(it);
             return true;
         }
         default:
+            Log::warning() << *this << " - Received unexpected [message=" << message << ",requestID=" << requestID
+                           << ",payloadSize=" << payload.size() << "]" << std::endl;
             return false;
     }
 }

src/fdb5/api/RemoteFDB.h

@@ -41,7 +41,7 @@ class RemoteFDB : public LocalFDB, public Client {
 public:  // method
 
     RemoteFDB(const eckit::Configuration& config, const std::string& name);
-    ~RemoteFDB() override {}
+    ~RemoteFDB() override;
 
     ListIterator inspect(const metkit::mars::MarsRequest& request) override;
 
@@ -94,6 +94,7 @@ class RemoteFDB : public LocalFDB, public Client {
     // The shared_ptr allows this removal to be asynchronous with the actual task
     // cleaning up and returning to the client.
     std::unordered_map<uint32_t, std::shared_ptr<MessageQueue>> messageQueues_;
+    std::mutex messageMutex_;
 };
 
 //----------------------------------------------------------------------------------------------------------------------

src/fdb5/api/helpers/ControlIterator.cc

@@ -10,6 +10,8 @@
 
 #include "fdb5/api/helpers/ControlIterator.h"
 
+#include <optional>
+
 #include "eckit/serialisation/Stream.h"
 
 #include "fdb5/database/Catalogue.h"
@@ -32,6 +34,36 @@ eckit::Stream& operator>>(eckit::Stream& s, ControlAction& a) {
 
 //----------------------------------------------------------------------------------------------------------------------
 
+std::ostream& operator<<(std::ostream& s, const ControlIdentifier& i) {
+    switch (i) {
+        case ControlIdentifier::None:
+            s << "None";
+            break;
+        case ControlIdentifier::List:
+            s << "List";
+            break;
+        case ControlIdentifier::Retrieve:
+            s << "Retrieve";
+            break;
+        case ControlIdentifier::Archive:
+            s << "Archive";
+            break;
+        case ControlIdentifier::Wipe:
+            s << "Wipe";
+            break;
+        case ControlIdentifier::UniqueRoot:
+            s << "UniqueRoot";
+            break;
+        case ControlIdentifier::UnsafeWipeAll:
+            s << "UnsafeWipeAll";
+            break;
+    }
+    s << "(" << static_cast<typename std::underlying_type<ControlIdentifier>::type>(i) << ")";
+    return s;
+}
+
+//----------------------------------------------------------------------------------------------------------------------
+
 ControlIdentifierIterator::ControlIdentifierIterator(const ControlIdentifiers& identifiers) :
     value_(0), remaining_(identifiers.value_) {
 
@@ -86,6 +118,61 @@ ControlIdentifiers::ControlIdentifiers(eckit::Stream& s) {
     s >> value_;
 }
 
+ControlIdentifiers ControlIdentifiers::parse(const eckit::LocalConfiguration& config, bool unsafeWipeAllDefault) {
+    ControlIdentifiers identifiers;
+
+    bool writable = config.getBool("writable", true);
+    bool visitable = config.getBool("visitable", true);
+    if (!config.getBool("list", visitable)) {
+        identifiers.value_ |= static_cast<value_type>(ControlIdentifier::List);
+    }
+    if (!config.getBool("retrieve", visitable)) {
+        identifiers.value_ |= static_cast<value_type>(ControlIdentifier::Retrieve);
+    }
+    if (!config.getBool("archive", writable)) {
+        identifiers.value_ |= static_cast<value_type>(ControlIdentifier::Archive);
+    }
+    if (!config.getBool("wipe", writable)) {
+        identifiers.value_ |= static_cast<value_type>(ControlIdentifier::Wipe);
+    }
+    // Unsafe Wipe all is disabled by default, unless explicitly enabled in the configuration file
+    if (!config.getBool("unsafeWipeAll", identifiers.enabled(ControlIdentifier::Wipe) && unsafeWipeAllDefault)) {
+        identifiers.value_ |= static_cast<value_type>(ControlIdentifier::UnsafeWipeAll);
+    }
+    return identifiers;
+}
+
+ControlIdentifiers ControlIdentifiers::parse(const eckit::LocalConfiguration& config, ControlIdentifiers defaultValue) {
+    ControlIdentifiers identifiers = defaultValue;
+
+    std::optional<bool> writable;
+    if (config.has("writable")) {
+        writable = config.getBool("writable");
+    }
+    std::optional<bool> visitable;
+    if (config.has("visitable")) {
+        visitable = config.getBool("visitable");
+    }
+    if (!config.getBool("list", visitable ? *visitable : defaultValue.enabled(ControlIdentifier::List))) {
+        identifiers.value_ |= static_cast<value_type>(ControlIdentifier::List);
+    }
+    if (!config.getBool("retrieve", visitable ? *visitable : defaultValue.enabled(ControlIdentifier::Retrieve))) {
+        identifiers.value_ |= static_cast<value_type>(ControlIdentifier::Retrieve);
+    }
+    if (!config.getBool("archive", writable ? *writable : defaultValue.enabled(ControlIdentifier::Archive))) {
+        identifiers.value_ |= static_cast<value_type>(ControlIdentifier::Archive);
+    }
+    if (!config.getBool("wipe", writable ? *writable : defaultValue.enabled(ControlIdentifier::Wipe))) {
+        identifiers.value_ |= static_cast<value_type>(ControlIdentifier::Wipe);
+    }
+    // Unsafe Wipe all is disabled by default, unless explicitly enabled in the configuration file
+    if (!config.getBool("unsafeWipeAll", identifiers.enabled(ControlIdentifier::Wipe) &&
+                                             defaultValue.enabled(ControlIdentifier::UnsafeWipeAll))) {
+        identifiers.value_ |= static_cast<value_type>(ControlIdentifier::UnsafeWipeAll);
+    }
+    return identifiers;
+}
+
 ControlIdentifiers& ControlIdentifiers::operator|=(const ControlIdentifier& val) {
     value_ |= static_cast<value_type>(val);
     return *this;

src/fdb5/api/helpers/ControlIterator.h

@@ -15,6 +15,7 @@
 
 #include <cstdint>
 
+#include "eckit/config/LocalConfiguration.h"
 #include "eckit/filesystem/URI.h"
 
 #include "fdb5/api/helpers/APIIterator.h"
@@ -49,12 +50,15 @@ enum class ControlIdentifier : uint16_t {
     Retrieve = 1 << 1,
     Archive = 1 << 2,
     Wipe = 1 << 3,
-    UniqueRoot = 1 << 4
+    UniqueRoot = 1 << 4,
+    UnsafeWipeAll = 1 << 5
 };
 
+std::ostream& operator<<(std::ostream& s, const ControlIdentifier& m);
+
 static const std::initializer_list<ControlIdentifier> ControlIdentifierList{
-    ControlIdentifier::List, ControlIdentifier::Retrieve, ControlIdentifier::Archive, ControlIdentifier::Wipe,
-    ControlIdentifier::UniqueRoot};
+    ControlIdentifier::List, ControlIdentifier::Retrieve,   ControlIdentifier::Archive,
+    ControlIdentifier::Wipe, ControlIdentifier::UniqueRoot, ControlIdentifier::UnsafeWipeAll};
 //----------------------------------------------------------------------------------------------------------------------
 
 // An iterator to facilitate working with the ControlIdentifiers structure
@@ -96,6 +100,9 @@ class ControlIdentifiers {
     ControlIdentifiers(const ControlIdentifier& val);
     ControlIdentifiers(eckit::Stream& s);
 
+    static ControlIdentifiers parse(const eckit::LocalConfiguration& config, bool unsafeWipeAllDefault = true);
+    static ControlIdentifiers parse(const eckit::LocalConfiguration& config, ControlIdentifiers defaultValue);
+
     ControlIdentifiers& operator|=(const ControlIdentifier& val);
     ControlIdentifiers operator|(const ControlIdentifier& val);