From 5c1a20c4dacfe03db90b70c2665e66a76574196c Mon Sep 17 00:00:00 2001 From: dec05eba Date: Sat, 3 Feb 2018 19:33:05 +0100 Subject: Add seed function, not yet finished --- include/Database.hpp | 20 +++------- include/Key.hpp | 1 + include/StagedObject.hpp | 38 +++++++++++++++++++ src/Database.cpp | 97 +++++++++++++++++++++++++++++++++++++++++++++--- tests/main.cpp | 2 +- 5 files changed, 137 insertions(+), 21 deletions(-) create mode 100644 include/StagedObject.hpp diff --git a/include/Database.hpp b/include/Database.hpp index 20d7513..68fff62 100644 --- a/include/Database.hpp +++ b/include/Database.hpp @@ -2,6 +2,7 @@ #include "types.hpp" #include "Key.hpp" +#include "StagedObject.hpp" #include "DataView.hpp" #include #include @@ -11,26 +12,13 @@ namespace odhtdb { class Group; - struct StagedCreateObject - { - Key key; - Group *primaryAdminGroup; - u64 timestamp; // In microseconds - }; - - struct StagedAddObject - { - Key key; - DataView data; - u64 timestamp; // In microseconds - }; - class Database { public: Database(const char *bootstrapNodeAddr, u16 port); ~Database(); + void seed(); void create(const Key &key, Group *primaryAdminGroup); void add(const Key &key, DataView data); void commit(); @@ -38,6 +26,10 @@ namespace odhtdb void commitStagedCreateObject(const StagedCreateObject &stagedObject); void commitStagedAddObject(const StagedAddObject &stagedObject); ntp::NtpTimestamp getSyncedTimestampUtc() const; + StagedCreateObject deserializeCreateRequest(const std::shared_ptr &value); + StagedAddObject deserializeAddRequest(const std::shared_ptr &value); + bool listenCreateData(const std::vector> &values); + bool listenAddData(const std::vector> &values); private: dht::DhtRunner node; std::vector stagedCreateObjects; diff --git a/include/Key.hpp b/include/Key.hpp index e9f0591..505050d 100644 --- a/include/Key.hpp +++ b/include/Key.hpp @@ -7,6 +7,7 @@ namespace odhtdb class Key { public: + Key() {} Key(const char *key) : hashedKey(dht::InfoHash::get(key)) {} dht::InfoHash hashedKey; diff --git a/include/StagedObject.hpp b/include/StagedObject.hpp new file mode 100644 index 0000000..61e1073 --- /dev/null +++ b/include/StagedObject.hpp @@ -0,0 +1,38 @@ +#pragma once + +#include "Key.hpp" +#include "types.hpp" +#include "DataView.hpp" + +namespace odhtdb +{ + class Group; + + struct StagedCreateObject + { + Key key; + Group *primaryAdminGroup; + u64 timestamp; // In microseconds + + StagedCreateObject() : key(), primaryAdminGroup(nullptr), timestamp(0) {} + StagedCreateObject(const Key &_key, Group *_primaryAdminGroup, u64 _timestamp) : + key(_key), primaryAdminGroup(_primaryAdminGroup), timestamp(_timestamp) + { + + } + }; + + struct StagedAddObject + { + Key key; + DataView data; + u64 timestamp; // In microseconds + + StagedAddObject() : key(), data(), timestamp(0) {} + StagedAddObject(const Key &_key, const DataView &_data, u64 _timestamp) : + key(_key), data(_data), timestamp(_timestamp) + { + + } + }; +} \ No newline at end of file diff --git a/src/Database.cpp b/src/Database.cpp index 33aef54..c2d08cb 100644 --- a/src/Database.cpp +++ b/src/Database.cpp @@ -7,6 +7,8 @@ #include #include #include +#include +#include using namespace dht; using namespace std; @@ -17,6 +19,10 @@ static int databaseCount = 0; static time_t timeOffset = 0; // Updated by comparing local time with ntp server static thread *ntpThread = nullptr; static bool timestampSynced = false; +static InfoHash CREATE_DATA_HASH = InfoHash::get("__odhtdb__.create_data"); +static InfoHash ADD_DATA_HASH = InfoHash::get("__odhtdb__.add_data"); + +#define OPENDHT_INFOHASH_LEN 20 namespace odhtdb { @@ -66,18 +72,30 @@ namespace odhtdb node.join(); } + void Database::seed() + { + // TODO: Use cached files and seed those. If none exists, request new files to seed. + // If nobody requests my cached files in a long time, request new files to seed and remove cached files + // (only if there are plenty of other seeders for the cached files. This could also cause race issue + // where all nodes with a cached file delete it at same time) + + using std::placeholders::_1; + node.listen(CREATE_DATA_HASH, bind(&Database::listenCreateData, this, _1)); + node.listen(ADD_DATA_HASH, bind(&Database::listenAddData, this, _1)); + } + void Database::create(const Key &key, Group *primaryAdminGroup) { // TODO: Append fractions to get real microseconds time u64 timeMicroseconds = ((u64)getSyncedTimestampUtc().seconds) * 1000000ull; - stagedCreateObjects.emplace_back(StagedCreateObject{ key, primaryAdminGroup, timeMicroseconds }); + stagedCreateObjects.emplace_back(StagedCreateObject(key, primaryAdminGroup, timeMicroseconds)); } void Database::add(const Key &key, DataView data) { // TODO: Append fractions to get real microseconds time u64 timeMicroseconds = ((u64)getSyncedTimestampUtc().seconds) * 1000000ull; - stagedAddObjects.emplace_back(StagedAddObject{ key, data, timeMicroseconds }); + stagedAddObjects.emplace_back(StagedAddObject(key, data, timeMicroseconds)); } void Database::commit() @@ -103,7 +121,9 @@ namespace odhtdb // TODO: Use (ed25519 or poly1305) and curve25519 // TODO: Implement gas and price (refill when serving content (seeding) or by waiting. This is done to prevent spamming and bandwidth leeching) sibs::SafeSerializer serializer; + assert(stagedObject.key.hashedKey.size() == OPENDHT_INFOHASH_LEN); serializer.add(stagedObject.key.hashedKey.data(), stagedObject.key.hashedKey.size()); + serializer.add(stagedObject.timestamp); serializer.add((u8)stagedObject.primaryAdminGroup->getName().size()); serializer.add((u8*)stagedObject.primaryAdminGroup->getName().data(), stagedObject.primaryAdminGroup->getName().size()); serializer.add((u32)stagedObject.primaryAdminGroup->getUsers().size()); @@ -113,9 +133,9 @@ namespace odhtdb serializer.add((u8*)user->getName().data(), user->getName().size()); } - // TODO: Verify if serializer buffer needs to survive longer than this method + // TODO: Verify if serializer buffer needs to survive longer than this scope Value value(serializer.getBuffer().data(), serializer.getBuffer().size()); - node.put(stagedObject.key.hashedKey, move(value), [](bool ok) + node.put(CREATE_DATA_HASH, move(value), [](bool ok) { // TODO: Handle failure to put data if(!ok) @@ -127,8 +147,15 @@ namespace odhtdb { // TODO: Use (ed25519 or poly1305) and curve25519 // TODO: Implement gas and price (refill when serving content (seeding) or by waiting. This is done to prevent spamming and bandwidth leeching) - Value value((const u8*)stagedObject.data.data, stagedObject.data.size); - node.put(stagedObject.key.hashedKey, move(value), [](bool ok) + sibs::SafeSerializer serializer; + assert(stagedObject.key.hashedKey.size() == OPENDHT_INFOHASH_LEN); + serializer.add(stagedObject.key.hashedKey.data(), stagedObject.key.hashedKey.size()); + serializer.add(stagedObject.timestamp); + serializer.add((u8*)stagedObject.data.data, stagedObject.data.size); + + // TODO: Verify if serializer buffer needs to survive longer than this scope + Value value(serializer.getBuffer().data(), serializer.getBuffer().size()); + node.put(ADD_DATA_HASH, move(value), [](bool ok) { // TODO: Handle failure to put data if(!ok) @@ -144,4 +171,62 @@ namespace odhtdb timestamp.fractions = 0; // TODO: Set this return timestamp; } + + StagedCreateObject Database::deserializeCreateRequest(const std::shared_ptr &value) + { + StagedCreateObject result; + + sibs::SafeDeserializer deserializer(value->data.data(), value->data.size()); + u8 entryKeyRaw[OPENDHT_INFOHASH_LEN]; + deserializer.extract(entryKeyRaw, OPENDHT_INFOHASH_LEN); + result.key.hashedKey = InfoHash(entryKeyRaw, OPENDHT_INFOHASH_LEN); + result.timestamp = deserializer.extract(); + + return result; + } + + StagedAddObject Database::deserializeAddRequest(const std::shared_ptr &value) + { + StagedAddObject result; + + sibs::SafeDeserializer deserializer(value->data.data(), value->data.size()); + u8 entryKeyRaw[OPENDHT_INFOHASH_LEN]; + deserializer.extract(entryKeyRaw, OPENDHT_INFOHASH_LEN); + result.key.hashedKey = InfoHash(entryKeyRaw, OPENDHT_INFOHASH_LEN); + result.timestamp = deserializer.extract(); + + return result; + } + + bool Database::listenCreateData(const std::vector> &values) + { + for(const shared_ptr &value : values) + { + try + { + StagedCreateObject createObject = deserializeCreateRequest(value); + } + catch (sibs::DeserializeException &e) + { + fprintf(stderr, "Warning: Failed to deserialize 'create' request: %s\n", e.what()); + } + } + return true; + } + + bool Database::listenAddData(const std::vector> &values) + { + for(const shared_ptr &value : values) + { + try + { + StagedAddObject addObject = deserializeAddRequest(value); + } + catch (sibs::DeserializeException &e) + { + fprintf(stderr, "Warning: Failed to deserialize 'add' request: %s\n", e.what()); + } + } + return true; + } } \ No newline at end of file diff --git a/tests/main.cpp b/tests/main.cpp index 2618d18..5ddcbe9 100644 --- a/tests/main.cpp +++ b/tests/main.cpp @@ -10,9 +10,9 @@ int main() // TODO: For tests, dont run against bootstrap.ring.cx. // Run against a bootstrap node made only for testing which doesn't persist added data. Database database("bootstrap.ring.cx", 4222); + database.seed(); LocalUser *localUser = LocalUser::create("dec05eba"); - Group group("admin"); group.addUser(localUser); database.create("galax.channel.latenight.chat", &group); -- cgit v1.2.3