Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 38 additions & 0 deletions src/atomdb/AtomDBFactory.cc
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
#include "AtomDBFactory.h"

#include "InMemoryDB.h"
#include "MorkDB.h"
#include "RedisMongoDB.h"
#include "Utils.h"

using namespace atomdb;
using namespace commons;

// --------------------------------------------------------------------------------
// Public methods

shared_ptr<AtomDB> AtomDBFactory::create(const JsonConfig& config, const string& context) {
return wrap_if_protected(create_backend(config, context));
}

shared_ptr<AtomDB> AtomDBFactory::create_backend(const JsonConfig& config, const string& context) {
auto atomdb_type = config.at_path("type").get_or<string>("");

if (atomdb_type == "redismongodb") {
return shared_ptr<AtomDB>(new RedisMongoDB(context, false, config));
Comment thread
andre-senna marked this conversation as resolved.
Outdated
}
if (atomdb_type == "morkdb") {
return shared_ptr<AtomDB>(new MorkDB(context, config));
}
if (atomdb_type == "inmemorydb") {
return make_shared<InMemoryDB>(context.empty() ? "inmemorydb_" : context);
}

RAISE_ERROR("AtomDBFactory: unsupported AtomDB type: " + atomdb_type);
return shared_ptr<AtomDB>{};
}

shared_ptr<AtomDB> AtomDBFactory::wrap_if_protected(shared_ptr<AtomDB> backend) {
// AtomDBFactory::wrap_if_protected() is not implemented yet.
return backend;
}
39 changes: 39 additions & 0 deletions src/atomdb/AtomDBFactory.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
#pragma once

#include <memory>
#include <string>

#include "AtomDB.h"
#include "JsonConfig.h"

using namespace std;
using namespace commons;

namespace atomdb {

/**
* @brief Single entry point to construct concrete AtomDB backends.
*
* Use this instead of calling RedisMongoDB/MorkDB/InMemoryDB constructors directly.
Comment thread
marcocapozzoli marked this conversation as resolved.
Outdated
*/
class AtomDBFactory {
public:
/**
* @brief Creates a backend and wraps it with ProtectedAtomDB when is_protected().
*/
static shared_ptr<AtomDB> create(const JsonConfig& config, const string& context = "");

/**
* @brief Creates a concrete backend without authorization wrapping.
*
* Supported types: redismongodb, morkdb, inmemorydb.
*/
static shared_ptr<AtomDB> create_backend(const JsonConfig& config, const string& context = "");

/**
* @brief Wraps backend with ProtectedAtomDB when protected and not already wrapped.
*/
static shared_ptr<AtomDB> wrap_if_protected(shared_ptr<AtomDB> backend);
};

} // namespace atomdb
36 changes: 18 additions & 18 deletions src/atomdb/AtomDBSingleton.cc
Original file line number Diff line number Diff line change
@@ -1,8 +1,7 @@
#include "AtomDBSingleton.h"

#include "AdapterDB.h"
#include "MorkDB.h"
#include "RedisMongoDB.h"
#include "AtomDBFactory.h"
#include "RemoteAtomDB.h"
#include "Utils.h"

Expand All @@ -19,24 +18,25 @@ void AtomDBSingleton::init(const JsonConfig& atomdb_config) {
if (AtomDBSingleton::initialized) {
RAISE_ERROR(
"AtomDBSingleton already initialized. AtomDBSingleton::init() should be called only once.");
}

shared_ptr<AtomDB> atomdb;
auto atomdb_type = atomdb_config.at_path("type").get_or<string>("");

if (atomdb_type == "remotedb") {
auto remote_peers_config =
atomdb_config.at_path("remote_peers").get_or<JsonConfig>(JsonConfig());
atomdb = shared_ptr<AtomDB>(new RemoteAtomDB(remote_peers_config));
atomdb = AtomDBFactory::wrap_if_protected(atomdb);
} else if (atomdb_type == "adapterdb") {
atomdb = shared_ptr<AtomDB>(new AdapterDB(atomdb_config));
atomdb = AtomDBFactory::wrap_if_protected(atomdb);
} else {
auto atomdb_type = atomdb_config.at_path("type").get_or<string>("");
if (atomdb_type == "morkdb") {
AtomDBSingleton::atom_db = shared_ptr<AtomDB>(new MorkDB("", atomdb_config));
} else if (atomdb_type == "redismongodb") {
AtomDBSingleton::atom_db = shared_ptr<AtomDB>(new RedisMongoDB("", false, atomdb_config));
} else if (atomdb_type == "remotedb") {
auto remote_peers_config =
atomdb_config.at_path("remote_peers").get_or<JsonConfig>(JsonConfig());
AtomDBSingleton::atom_db = shared_ptr<AtomDB>(new RemoteAtomDB(remote_peers_config));
} else if (atomdb_type == "adapterdb") {
AtomDBSingleton::atom_db = shared_ptr<AtomDB>(new AdapterDB(atomdb_config));
} else {
RAISE_ERROR("Invalid AtomDB type: " + atomdb_type);
}

AtomDBSingleton::initialized = true;
atomdb = AtomDBFactory::create(atomdb_config);
}

AtomDBSingleton::atom_db = atomdb;
AtomDBSingleton::initialized = true;
}

shared_ptr<AtomDB> AtomDBSingleton::get_instance() {
Expand Down
18 changes: 16 additions & 2 deletions src/atomdb/BUILD
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ cc_library(
deps = [
":atomdb",
":atomdb_api_types",
":atomdb_factory",
":atomdb_singleton",
":atomdbutils",
"//atomdb/adapterdb:adapterdb_lib",
Expand All @@ -18,6 +19,20 @@ cc_library(
],
)

cc_library(
name = "atomdb_factory",
srcs = ["AtomDBFactory.cc"],
hdrs = ["AtomDBFactory.h"],
includes = ["."],
deps = [
":atomdb",
"//atomdb/inmemorydb",
"//atomdb/morkdb",
"//atomdb/redis_mongodb",
"//commons:commons_lib",
],
)

cc_library(
name = "atomdb",
hdrs = ["AtomDB.h"],
Expand Down Expand Up @@ -56,10 +71,9 @@ cc_library(
hdrs = ["AtomDBSingleton.h"],
includes = ["."],
deps = [
":atomdb_factory",
"//atomdb:atomdb_api_types",
"//atomdb/adapterdb",
"//atomdb/morkdb",
"//atomdb/redis_mongodb",
"//atomdb/remotedb:remotedb_lib",
"//commons:commons_lib",
],
Expand Down
14 changes: 7 additions & 7 deletions src/atomdb/adapterdb/AdapterDB.cc
Original file line number Diff line number Diff line change
Expand Up @@ -3,17 +3,16 @@
#include <chrono>
#include <thread>

#include "AtomDBFactory.h"
#include "AtomPersister.h"
#include "BoundedSharedQueue.h"
#include "DatabaseOrchestrator.h"
#include "DedicatedThread.h"
#include "MongoInitializer.h"
#include "MorkDB.h"
#include "MorkMappingStrategy.h"
#include "PostgresMappingStrategy.h"
#include "PostgresWrapper.h"
#include "Processor.h"
#include "RedisMongoDB.h"
#include "RemoteAtomDB.h"
#include "Utils.h"
#include "expression_hasher.h"
Expand Down Expand Up @@ -311,15 +310,16 @@ void AdapterDB::persistence_setup() {
}

void AdapterDB::atomdb_backend_setup() {
// TODO: create_backend(): raw AtomDB on purpose. Wrapping nested RemoteAtomDB/AdapterDB stores with
// ProtectedAtomDB is still under discussion.
auto atomdb_backend_config =
this->config.at_path("adapterdb.atomdb_backend").get_or<JsonConfig>(JsonConfig());
string atomdb_backend_type = atomdb_backend_config.at_path("type").get_or<string>("");
if (atomdb_backend_type == "morkdb") {
this->atomdb_backend = shared_ptr<AtomDB>(new MorkDB("", atomdb_backend_config));
} else if (atomdb_backend_type == "redismongodb") {
this->atomdb_backend = shared_ptr<AtomDB>(new RedisMongoDB("", false, atomdb_backend_config));
} else if (atomdb_backend_type == "remotedb") {
if (atomdb_backend_type == "remotedb") {
this->atomdb_backend = shared_ptr<AtomDB>(new RemoteAtomDB(atomdb_backend_config));
} else if (atomdb_backend_type == "morkdb" || atomdb_backend_type == "redismongodb" ||
atomdb_backend_type == "inmemorydb") {
this->atomdb_backend = AtomDBFactory::create_backend(atomdb_backend_config);
} else {
RAISE_ERROR("Invalid AtomDB type: " + atomdb_backend_type);
}
Expand Down
3 changes: 1 addition & 2 deletions src/atomdb/adapterdb/BUILD
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,7 @@ cc_library(
includes = ["."],
deps = [
"//atomdb",
"//atomdb/morkdb",
"//atomdb/redis_mongodb",
"//atomdb:atomdb_factory",
"//atomdb/remotedb:remotedb_lib",
"//commons:commons_lib",
"//commons/atoms:atoms_lib",
Expand Down
6 changes: 5 additions & 1 deletion src/atomdb/redis_mongodb/RedisMongoDB.h
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@ enum MONGODB_FIELD { ID = 0, NAME, TARGETS, NAMED_TYPE, size };

class RedisMongoDB : public AtomDB {
public:
RedisMongoDB(const string& context, bool skip_redis, const JsonConfig& config);
~RedisMongoDB();

bool allow_nested_indexing() override;
Expand Down Expand Up @@ -146,6 +145,11 @@ class RedisMongoDB : public AtomDB {
map<string, vector<string>>& composite_type_entries_map);

private:
friend class AtomDBFactory;
friend class MorkDB;
Comment thread
marcocapozzoli marked this conversation as resolved.
Outdated

RedisMongoDB(const string& context, bool skip_redis, const JsonConfig& config);

string context;
bool skip_redis_;
bool composite_type_enabled_;
Expand Down
3 changes: 1 addition & 2 deletions src/atomdb/remotedb/BUILD
Original file line number Diff line number Diff line change
Expand Up @@ -24,10 +24,9 @@ cc_library(
deps = [
"//atomdb",
"//atomdb:atomdb_api_types",
"//atomdb:atomdb_factory",
"//atomdb/inmemorydb",
"//atomdb/inmemorydb:inmemorydb_api_types",
"//atomdb/morkdb:morkdb_lib",
"//atomdb/redis_mongodb:redis_mongodb_lib",
"//commons:commons_lib",
"//commons/atoms:atoms_lib",
"//commons/processor:processor_lib",
Expand Down
58 changes: 25 additions & 33 deletions src/atomdb/remotedb/RemoteAtomDB.cc
Original file line number Diff line number Diff line change
Expand Up @@ -7,11 +7,9 @@
#include <sstream>
#include <utility>

#include "InMemoryDB.h"
#include "AtomDBFactory.h"
#include "InMemoryDBAPITypes.h"
#include "Logger.h"
#include "MorkDB.h"
#include "RedisMongoDB.h"
#include "Utils.h"

using namespace atomdb;
Expand All @@ -20,48 +18,33 @@ using namespace commons;

using json = nlohmann::json;

namespace {

shared_ptr<AtomDB> create_atomdb_from_config(const JsonConfig& config) {
string uid = config.at_path("uid").get_or<string>("");
string type = config.at_path("type").get_or<string>("");
string context = config.at_path("context").get_or<string>("");

if (type == "inmemorydb") {
return make_shared<InMemoryDB>(context.empty() ? "remotedb_" : context);
}

if (type == "redismongodb") {
RedisMongoDB::initialize_statics(context);
auto atomdb = make_shared<RedisMongoDB>(context, false, config);
return atomdb;
}

if (type == "morkdb") {
auto atomdb = make_shared<MorkDB>(context, config);
return atomdb;
}

RAISE_ERROR("Unknown AtomDB type for peer " + uid + ": " + type);
return nullptr;
}

} // namespace

RemoteAtomDB::RemoteAtomDB(const JsonConfig& peers_config) {
for (auto& entry : peers_config) {
auto peer_config = JsonConfig(entry);
string uid = peer_config.at_path("uid").get_or<string>("");
if (uid.empty()) continue;

string context = peer_config.at_path("context").get_or<string>("");
if (context.empty()) {
context = "remotedb_" + uid;
}

shared_ptr<AtomDB> local_persistence = nullptr;
auto local_persistence_config =
peer_config.at_path("local_persistence").get_or<JsonConfig>(JsonConfig());
if (!local_persistence_config.empty()) {
local_persistence = create_atomdb_from_config(local_persistence_config);
string local_context = local_persistence_config.at_path("context").get_or<string>(context);
if (local_context.empty()) {
local_context = context;
}
// TODO: create_backend(): raw AtomDB on purpose. Wrapping nested RemoteAtomDB/AdapterDB
// stores with ProtectedAtomDB is still under discussion.
local_persistence = AtomDBFactory::create_backend(local_persistence_config, local_context);
}
// TODO: create_backend(): raw AtomDB on purpose. Wrapping nested RemoteAtomDB/AdapterDB stores
// with ProtectedAtomDB is still under discussion.
remote_db_[uid] = make_shared<RemoteAtomDBPeer>(
create_atomdb_from_config(peer_config), local_persistence, uid);
AtomDBFactory::create_backend(peer_config, context), local_persistence, uid);
}

LOG_INFO("RemoteAtomDB initialized with " << remote_db_.size() << " remote peers");
Expand All @@ -87,6 +70,15 @@ bool RemoteAtomDB::composite_type_enabled() const {
return false;
}

bool RemoteAtomDB::is_protected() const {
for (auto& [uid, peer] : remote_db_) {
if (peer->is_protected()) {
return true;
}
}
return false;
}

void RemoteAtomDB::derive_nested_indexing() {
// Derive the aggregated nested-indexing capability from the peers. A single global boolean
// cannot describe a heterogeneous result set, so mixed configurations are normalized to the
Expand Down
2 changes: 1 addition & 1 deletion src/main/BUILD
Original file line number Diff line number Diff line change
Expand Up @@ -58,9 +58,9 @@ cc_library(
srcs = ["db_loader.cc"],
deps = [
"//atomdb:atomdb_api_types",
"//atomdb:atomdb_factory",
"//atomdb:atomdb_singleton",
"//atomdb/adapterdb:adapterdb_lib",
"//atomdb/morkdb:morkdb_lib",
"//atomdb/redis_mongodb:redis_mongodb_lib",
"//atomdb/remotedb:remotedb_lib",
Comment thread
marcocapozzoli marked this conversation as resolved.
Outdated
"//commons:commons_lib",
Expand Down
14 changes: 5 additions & 9 deletions src/main/db_loader.cc
Original file line number Diff line number Diff line change
Expand Up @@ -8,12 +8,12 @@
#include <vector>

#include "AdapterDB.h"
#include "AtomDBFactory.h"
#include "AtomDBSingleton.h"
#include "JsonConfig.h"
#include "JsonConfigParser.h"
#include "MettaParser.h"
#include "MettaParserActions.h"
#include "MorkDB.h"
#include "RedisMongoDB.h"
#include "RemoteAtomDB.h"
#include "Utils.h"
Expand Down Expand Up @@ -73,18 +73,14 @@ int main(int argc, char* argv[]) {
auto atomdb_config = json_config.at_path("atomdb").get_or<JsonConfig>(JsonConfig());

auto atomdb_type = atomdb_config.at_path("type").get_or<string>("");
if (atomdb_type == "redismongodb") {
AtomDBSingleton::provide(make_shared<RedisMongoDB>(context, false, atomdb_config));
} else if (atomdb_type == "morkdb") {
AtomDBSingleton::provide(make_shared<MorkDB>(context, atomdb_config));
} else if (atomdb_type == "remotedb") {
if (atomdb_type == "remotedb") {
auto remote_peers_config =
atomdb_config.at_path("remote_peers").get_or<JsonConfig>(JsonConfig());
AtomDBSingleton::provide(make_shared<RemoteAtomDB>(remote_peers_config));
AtomDBSingleton::provide(shared_ptr<AtomDB>(new RemoteAtomDB(remote_peers_config)));
} else if (atomdb_type == "adapterdb") {
AtomDBSingleton::provide(make_shared<AdapterDB>(atomdb_config));
AtomDBSingleton::provide(shared_ptr<AtomDB>(new AdapterDB(atomdb_config)));
} else {
RAISE_ERROR("Invalid AtomDB type: " + atomdb_type);
AtomDBSingleton::provide(AtomDBFactory::create_backend(atomdb_config, context));
Comment thread
marcocapozzoli marked this conversation as resolved.
Outdated
}

signal(SIGINT, &ctrl_c_handler);
Expand Down
Loading
Loading