diff options
author | Aleksi Lindeman <aleksi_888@hotmail.com> | 2018-02-01 21:15:13 +0100 |
---|---|---|
committer | Aleksi Lindeman <aleksi_888@hotmail.com> | 2018-02-01 21:15:19 +0100 |
commit | 6f1089db78f14b52b869f5aaa979e52ff5e4c2d7 (patch) | |
tree | d79f7abfb0fb9f3d58f9716741512b63fb4193ec /src/Database.cpp | |
parent | 80a9c135b8bdca64246f147f22d98485e0f05ee5 (diff) |
Sync time with ntp server, starting with basic operations
Diffstat (limited to 'src/Database.cpp')
-rw-r--r-- | src/Database.cpp | 113 |
1 files changed, 106 insertions, 7 deletions
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 <opendht.h> #include <fmt/format.h> +#include <sodium/crypto_box_curve25519xchacha20poly1305.h> +#include <thread> +#include <chrono> +#include <sibs/SafeSerializer.hpp> 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 |