From 1c7e6e074155499155adbbb651db1c66f1762ba2 Mon Sep 17 00:00:00 2001 From: dec05eba Date: Thu, 1 Feb 2018 21:15:13 +0100 Subject: Sync time with ntp server, starting with basic operations --- src/Database.cpp | 113 +++++++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 106 insertions(+), 7 deletions(-) (limited to 'src/Database.cpp') diff --git a/src/Database.cpp b/src/Database.cpp index 4fa8405..33aef54 100644 --- a/src/Database.cpp +++ b/src/Database.cpp @@ -1,9 +1,22 @@ #include "../include/Database.hpp" +#include "../include/Group.hpp" +#include "../include/User.hpp" #include #include +#include +#include +#include +#include using namespace dht; using namespace std; +using namespace chrono_literals; + +static int databaseCount = 0; +// TODO: Verify time_t is always signed +static time_t timeOffset = 0; // Updated by comparing local time with ntp server +static thread *ntpThread = nullptr; +static bool timestampSynced = false; namespace odhtdb { @@ -13,36 +26,122 @@ namespace odhtdb fmt::MemoryWriter portStr; portStr << port; node.bootstrap(bootstrapNodeAddr, portStr.c_str()); + + // TODO: Make this work for multiple threads initializing database at same time + ++databaseCount; + if(databaseCount == 1) + { + if(ntpThread) + delete ntpThread; + + ntpThread = new thread([]() + { + ntp::NtpClient ntpClient("pool.ntp.org"); + while(databaseCount > 0) + { + ntp::NtpTimestamp ntpTimestamp = ntpClient.getTimestamp(); + timeOffset = time(nullptr) - ntpTimestamp.seconds; + timestampSynced = true; + // TODO: Also use timestamp fraction (milliseconds) + this_thread::sleep_for(60s); + } + timestampSynced = false; + }); + + // TODO: Catch std::system_error instead of this if-statement + if(ntpThread->joinable()) + ntpThread->detach(); + } + + while(!timestampSynced) + { + this_thread::sleep_for(10ms); + } } Database::~Database() { + // TODO: Make this work for multiple threads removing database object at same time + --databaseCount; node.join(); } + 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 }); + } + void Database::add(const Key &key, DataView data) { - stagedObjects.emplace_back(StagedObject{ key, data }); + // TODO: Append fractions to get real microseconds time + u64 timeMicroseconds = ((u64)getSyncedTimestampUtc().seconds) * 1000000ull; + stagedAddObjects.emplace_back(StagedAddObject{ key, data, timeMicroseconds }); } void Database::commit() { - // TODO: Combine staged objects into one object - for(StagedObject stagedObject : stagedObjects) + // TODO: Combine staged objects into one object for efficiency. + // TODO: Add rollback + + for(StagedCreateObject &stagedObject : stagedCreateObjects) { - commitStagedObject(stagedObject); + commitStagedCreateObject(stagedObject); } - stagedObjects.clear(); + stagedCreateObjects.clear(); + + for(StagedAddObject &stagedObject : stagedAddObjects) + { + commitStagedAddObject(stagedObject); + } + stagedAddObjects.clear(); + } + + void Database::commitStagedCreateObject(const StagedCreateObject &stagedObject) + { + // 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; + serializer.add(stagedObject.key.hashedKey.data(), stagedObject.key.hashedKey.size()); + 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()); + for(User *user : stagedObject.primaryAdminGroup->getUsers()) + { + serializer.add((u8)user->getName().size()); + serializer.add((u8*)user->getName().data(), user->getName().size()); + } + + // TODO: Verify if serializer buffer needs to survive longer than this method + Value value(serializer.getBuffer().data(), serializer.getBuffer().size()); + node.put(stagedObject.key.hashedKey, move(value), [](bool ok) + { + // TODO: Handle failure to put data + if(!ok) + fprintf(stderr, "Failed to put: %s, what to do?\n", "commitStagedCreateObject"); + }, time_point(), false); } - void Database::commitStagedObject(const StagedObject &stagedObject) + void Database::commitStagedAddObject(const StagedAddObject &stagedObject) { + // 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) { // TODO: Handle failure to put data if(!ok) - fprintf(stderr, "Failed to put: %s, what to do?\n", "commitStagedObject"); + fprintf(stderr, "Failed to put: %s, what to do?\n", "commitStagedAddObject"); }, time_point(), false); } + + ntp::NtpTimestamp Database::getSyncedTimestampUtc() const + { + assert(timestampSynced); + ntp::NtpTimestamp timestamp; + timestamp.seconds = time(nullptr) - timeOffset; + timestamp.fractions = 0; // TODO: Set this + return timestamp; + } } \ No newline at end of file -- cgit v1.2.3