mirror/ternfs-XTXMarkets
1
0
Fork 0
mirror of https://github.com/XTXMarkets/ternfs.git synced 2026-01-06 11:00:10 -06:00
Code Issues Packages Projects Releases Wiki Activity

Move things around a bit in preparation for cpp CDC

Browse Source 
Also, make log preparation in shard read-only, in preparation for
distributed consensus.
This commit is contained in:
Francesco Mazzoli
2022-12-11 19:32:54 +00:00
parent f94c911f14
commit ba0f8a7a91
10 changed files with 688 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
cpp/AssertiveLock.hpp Normal file
View File

@@ -0,0 +1,39 @@
#pragma once
#include <atomic>
#include "Common.hpp"
#include "Exception.hpp"
struct AssertiveLocked {
private:
std::atomic<bool>& _held;
public:
AssertiveLocked(std::atomic<bool>& held): _held(held) {
bool expected = false;
if (!_held.compare_exchange_strong(expected, true)) {
throw EGGS_EXCEPTION("could not aquire lock, are you using this function concurrently?");
}
}
AssertiveLocked(const AssertiveLocked&) = delete;
AssertiveLocked& operator=(AssertiveLocked&) = delete;
~AssertiveLocked() {
_held.store(false);
}
};
struct AssertiveLock {
private:
std::atomic<bool> _held;
public:
AssertiveLock(): _held(false) {}
AssertiveLock(const AssertiveLock&) = delete;
AssertiveLock& operator=(AssertiveLock&) = delete;
AssertiveLocked lock() {
return AssertiveLocked(_held);
}
};

307
cpp/MsgsGen.cpp
View File

@@ -3554,6 +3554,307 @@ std::ostream& operator<<(std::ostream& out, const ShardRespContainer& x) {
return out;
}
std::ostream& operator<<(std::ostream& out, CDCMessageKind kind) {
switch (kind) {
case CDCMessageKind::MAKE_DIRECTORY:
out << "MAKE_DIRECTORY";
break;
case CDCMessageKind::RENAME_FILE:
out << "RENAME_FILE";
break;
case CDCMessageKind::SOFT_UNLINK_DIRECTORY:
out << "SOFT_UNLINK_DIRECTORY";
break;
case CDCMessageKind::RENAME_DIRECTORY:
out << "RENAME_DIRECTORY";
break;
case CDCMessageKind::HARD_UNLINK_DIRECTORY:
out << "HARD_UNLINK_DIRECTORY";
break;
case CDCMessageKind::HARD_UNLINK_FILE:
out << "HARD_UNLINK_FILE";
break;
default:
out << "CDCMessageKind(" << ((int)kind) << ")";
break;
}
return out;
}
const MakeDirectoryReq& CDCReqContainer::getMakeDirectory() const {
ALWAYS_ASSERT(_kind == CDCMessageKind::MAKE_DIRECTORY, "%s != %s", _kind, CDCMessageKind::MAKE_DIRECTORY);
return std::get<0>(_data);
}
MakeDirectoryReq& CDCReqContainer::setMakeDirectory() {
_kind = CDCMessageKind::MAKE_DIRECTORY;
auto& x = std::get<0>(_data);
x.clear();
return x;
}
const RenameFileReq& CDCReqContainer::getRenameFile() const {
ALWAYS_ASSERT(_kind == CDCMessageKind::RENAME_FILE, "%s != %s", _kind, CDCMessageKind::RENAME_FILE);
return std::get<1>(_data);
}
RenameFileReq& CDCReqContainer::setRenameFile() {
_kind = CDCMessageKind::RENAME_FILE;
auto& x = std::get<1>(_data);
x.clear();
return x;
}
const SoftUnlinkDirectoryReq& CDCReqContainer::getSoftUnlinkDirectory() const {
ALWAYS_ASSERT(_kind == CDCMessageKind::SOFT_UNLINK_DIRECTORY, "%s != %s", _kind, CDCMessageKind::SOFT_UNLINK_DIRECTORY);
return std::get<2>(_data);
}
SoftUnlinkDirectoryReq& CDCReqContainer::setSoftUnlinkDirectory() {
_kind = CDCMessageKind::SOFT_UNLINK_DIRECTORY;
auto& x = std::get<2>(_data);
x.clear();
return x;
}
const RenameDirectoryReq& CDCReqContainer::getRenameDirectory() const {
ALWAYS_ASSERT(_kind == CDCMessageKind::RENAME_DIRECTORY, "%s != %s", _kind, CDCMessageKind::RENAME_DIRECTORY);
return std::get<3>(_data);
}
RenameDirectoryReq& CDCReqContainer::setRenameDirectory() {
_kind = CDCMessageKind::RENAME_DIRECTORY;
auto& x = std::get<3>(_data);
x.clear();
return x;
}
const HardUnlinkDirectoryReq& CDCReqContainer::getHardUnlinkDirectory() const {
ALWAYS_ASSERT(_kind == CDCMessageKind::HARD_UNLINK_DIRECTORY, "%s != %s", _kind, CDCMessageKind::HARD_UNLINK_DIRECTORY);
return std::get<4>(_data);
}
HardUnlinkDirectoryReq& CDCReqContainer::setHardUnlinkDirectory() {
_kind = CDCMessageKind::HARD_UNLINK_DIRECTORY;
auto& x = std::get<4>(_data);
x.clear();
return x;
}
const HardUnlinkFileReq& CDCReqContainer::getHardUnlinkFile() const {
ALWAYS_ASSERT(_kind == CDCMessageKind::HARD_UNLINK_FILE, "%s != %s", _kind, CDCMessageKind::HARD_UNLINK_FILE);
return std::get<5>(_data);
}
HardUnlinkFileReq& CDCReqContainer::setHardUnlinkFile() {
_kind = CDCMessageKind::HARD_UNLINK_FILE;
auto& x = std::get<5>(_data);
x.clear();
return x;
}
void CDCReqContainer::pack(BincodeBuf& buf) const {
switch (_kind) {
case CDCMessageKind::MAKE_DIRECTORY:
std::get<0>(_data).pack(buf);
break;
case CDCMessageKind::RENAME_FILE:
std::get<1>(_data).pack(buf);
break;
case CDCMessageKind::SOFT_UNLINK_DIRECTORY:
std::get<2>(_data).pack(buf);
break;
case CDCMessageKind::RENAME_DIRECTORY:
std::get<3>(_data).pack(buf);
break;
case CDCMessageKind::HARD_UNLINK_DIRECTORY:
std::get<4>(_data).pack(buf);
break;
case CDCMessageKind::HARD_UNLINK_FILE:
std::get<5>(_data).pack(buf);
break;
default:
throw EGGS_EXCEPTION("bad CDCMessageKind kind %s", _kind);
}
}
void CDCReqContainer::unpack(BincodeBuf& buf, CDCMessageKind kind) {
_kind = kind;
switch (kind) {
case CDCMessageKind::MAKE_DIRECTORY:
std::get<0>(_data).unpack(buf);
break;
case CDCMessageKind::RENAME_FILE:
std::get<1>(_data).unpack(buf);
break;
case CDCMessageKind::SOFT_UNLINK_DIRECTORY:
std::get<2>(_data).unpack(buf);
break;
case CDCMessageKind::RENAME_DIRECTORY:
std::get<3>(_data).unpack(buf);
break;
case CDCMessageKind::HARD_UNLINK_DIRECTORY:
std::get<4>(_data).unpack(buf);
break;
case CDCMessageKind::HARD_UNLINK_FILE:
std::get<5>(_data).unpack(buf);
break;
default:
throw BINCODE_EXCEPTION("bad CDCMessageKind kind %s", kind);
}
}
std::ostream& operator<<(std::ostream& out, const CDCReqContainer& x) {
switch (x.kind()) {
case CDCMessageKind::MAKE_DIRECTORY:
out << x.getMakeDirectory();
break;
case CDCMessageKind::RENAME_FILE:
out << x.getRenameFile();
break;
case CDCMessageKind::SOFT_UNLINK_DIRECTORY:
out << x.getSoftUnlinkDirectory();
break;
case CDCMessageKind::RENAME_DIRECTORY:
out << x.getRenameDirectory();
break;
case CDCMessageKind::HARD_UNLINK_DIRECTORY:
out << x.getHardUnlinkDirectory();
break;
case CDCMessageKind::HARD_UNLINK_FILE:
out << x.getHardUnlinkFile();
break;
default:
throw EGGS_EXCEPTION("bad CDCMessageKind kind %s", x.kind());
}
return out;
}
const MakeDirectoryResp& CDCRespContainer::getMakeDirectory() const {
ALWAYS_ASSERT(_kind == CDCMessageKind::MAKE_DIRECTORY, "%s != %s", _kind, CDCMessageKind::MAKE_DIRECTORY);
return std::get<0>(_data);
}
MakeDirectoryResp& CDCRespContainer::setMakeDirectory() {
_kind = CDCMessageKind::MAKE_DIRECTORY;
auto& x = std::get<0>(_data);
x.clear();
return x;
}
const RenameFileResp& CDCRespContainer::getRenameFile() const {
ALWAYS_ASSERT(_kind == CDCMessageKind::RENAME_FILE, "%s != %s", _kind, CDCMessageKind::RENAME_FILE);
return std::get<1>(_data);
}
RenameFileResp& CDCRespContainer::setRenameFile() {
_kind = CDCMessageKind::RENAME_FILE;
auto& x = std::get<1>(_data);
x.clear();
return x;
}
const SoftUnlinkDirectoryResp& CDCRespContainer::getSoftUnlinkDirectory() const {
ALWAYS_ASSERT(_kind == CDCMessageKind::SOFT_UNLINK_DIRECTORY, "%s != %s", _kind, CDCMessageKind::SOFT_UNLINK_DIRECTORY);
return std::get<2>(_data);
}
SoftUnlinkDirectoryResp& CDCRespContainer::setSoftUnlinkDirectory() {
_kind = CDCMessageKind::SOFT_UNLINK_DIRECTORY;
auto& x = std::get<2>(_data);
x.clear();
return x;
}
const RenameDirectoryResp& CDCRespContainer::getRenameDirectory() const {
ALWAYS_ASSERT(_kind == CDCMessageKind::RENAME_DIRECTORY, "%s != %s", _kind, CDCMessageKind::RENAME_DIRECTORY);
return std::get<3>(_data);
}
RenameDirectoryResp& CDCRespContainer::setRenameDirectory() {
_kind = CDCMessageKind::RENAME_DIRECTORY;
auto& x = std::get<3>(_data);
x.clear();
return x;
}
const HardUnlinkDirectoryResp& CDCRespContainer::getHardUnlinkDirectory() const {
ALWAYS_ASSERT(_kind == CDCMessageKind::HARD_UNLINK_DIRECTORY, "%s != %s", _kind, CDCMessageKind::HARD_UNLINK_DIRECTORY);
return std::get<4>(_data);
}
HardUnlinkDirectoryResp& CDCRespContainer::setHardUnlinkDirectory() {
_kind = CDCMessageKind::HARD_UNLINK_DIRECTORY;
auto& x = std::get<4>(_data);
x.clear();
return x;
}
const HardUnlinkFileResp& CDCRespContainer::getHardUnlinkFile() const {
ALWAYS_ASSERT(_kind == CDCMessageKind::HARD_UNLINK_FILE, "%s != %s", _kind, CDCMessageKind::HARD_UNLINK_FILE);
return std::get<5>(_data);
}
HardUnlinkFileResp& CDCRespContainer::setHardUnlinkFile() {
_kind = CDCMessageKind::HARD_UNLINK_FILE;
auto& x = std::get<5>(_data);
x.clear();
return x;
}
void CDCRespContainer::pack(BincodeBuf& buf) const {
switch (_kind) {
case CDCMessageKind::MAKE_DIRECTORY:
std::get<0>(_data).pack(buf);
break;
case CDCMessageKind::RENAME_FILE:
std::get<1>(_data).pack(buf);
break;
case CDCMessageKind::SOFT_UNLINK_DIRECTORY:
std::get<2>(_data).pack(buf);
break;
case CDCMessageKind::RENAME_DIRECTORY:
std::get<3>(_data).pack(buf);
break;
case CDCMessageKind::HARD_UNLINK_DIRECTORY:
std::get<4>(_data).pack(buf);
break;
case CDCMessageKind::HARD_UNLINK_FILE:
std::get<5>(_data).pack(buf);
break;
default:
throw EGGS_EXCEPTION("bad CDCMessageKind kind %s", _kind);
}
}
void CDCRespContainer::unpack(BincodeBuf& buf, CDCMessageKind kind) {
_kind = kind;
switch (kind) {
case CDCMessageKind::MAKE_DIRECTORY:
std::get<0>(_data).unpack(buf);
break;
case CDCMessageKind::RENAME_FILE:
std::get<1>(_data).unpack(buf);
break;
case CDCMessageKind::SOFT_UNLINK_DIRECTORY:
std::get<2>(_data).unpack(buf);
break;
case CDCMessageKind::RENAME_DIRECTORY:
std::get<3>(_data).unpack(buf);
break;
case CDCMessageKind::HARD_UNLINK_DIRECTORY:
std::get<4>(_data).unpack(buf);
break;
case CDCMessageKind::HARD_UNLINK_FILE:
std::get<5>(_data).unpack(buf);
break;
default:
throw BINCODE_EXCEPTION("bad CDCMessageKind kind %s", kind);
}
}
std::ostream& operator<<(std::ostream& out, const CDCRespContainer& x) {
switch (x.kind()) {
case CDCMessageKind::MAKE_DIRECTORY:
out << x.getMakeDirectory();
break;
case CDCMessageKind::RENAME_FILE:
out << x.getRenameFile();
break;
case CDCMessageKind::SOFT_UNLINK_DIRECTORY:
out << x.getSoftUnlinkDirectory();
break;
case CDCMessageKind::RENAME_DIRECTORY:
out << x.getRenameDirectory();
break;
case CDCMessageKind::HARD_UNLINK_DIRECTORY:
out << x.getHardUnlinkDirectory();
break;
case CDCMessageKind::HARD_UNLINK_FILE:
out << x.getHardUnlinkFile();
break;
default:
throw EGGS_EXCEPTION("bad CDCMessageKind kind %s", x.kind());
}
return out;
}
std::ostream& operator<<(std::ostream& out, ShardLogEntryKind err) {
switch (err) {
case ShardLogEntryKind::CONSTRUCT_FILE:
@@ -3609,32 +3910,28 @@ std::ostream& operator<<(std::ostream& out, ShardLogEntryKind err) {
}
void ConstructFileEntry::pack(BincodeBuf& buf) const {
id.pack(buf);
buf.packScalar<uint8_t>(type);
deadlineTime.pack(buf);
buf.packBytes(note);
}
void ConstructFileEntry::unpack(BincodeBuf& buf) {
id.unpack(buf);
type = buf.unpackScalar<uint8_t>();
deadlineTime.unpack(buf);
buf.unpackBytes(note);
}
void ConstructFileEntry::clear() {
id = InodeId();
type = uint8_t(0);
deadlineTime = EggsTime();
note.clear();
}
bool ConstructFileEntry::operator==(const ConstructFileEntry& rhs) const {
if ((InodeId)this->id != (InodeId)rhs.id) { return false; };
if ((uint8_t)this->type != (uint8_t)rhs.type) { return false; };
if ((EggsTime)this->deadlineTime != (EggsTime)rhs.deadlineTime) { return false; };
if (note != rhs.note) { return false; };
return true;
}
std::ostream& operator<<(std::ostream& out, const ConstructFileEntry& x) {
out << "ConstructFileEntry(" << "Id=" << x.id << ", " << "Type=" << (int)x.type << ", " << "DeadlineTime=" << x.deadlineTime << ", " << "Note=" << x.note << ")";
out << "ConstructFileEntry(" << "Type=" << (int)x.type << ", " << "DeadlineTime=" << x.deadlineTime << ", " << "Note=" << x.note << ")";
return out;
}

70
cpp/MsgsGen.hpp
View File

@@ -2258,6 +2258,72 @@ public:
std::ostream& operator<<(std::ostream& out, const ShardRespContainer& x);
enum class CDCMessageKind : uint8_t {
ERROR = 0,
MAKE_DIRECTORY = 1,
RENAME_FILE = 2,
SOFT_UNLINK_DIRECTORY = 3,
RENAME_DIRECTORY = 4,
HARD_UNLINK_DIRECTORY = 5,
HARD_UNLINK_FILE = 6,
};
std::ostream& operator<<(std::ostream& out, CDCMessageKind kind);
struct CDCReqContainer {
private:
CDCMessageKind _kind = (CDCMessageKind)0;
std::tuple<MakeDirectoryReq, RenameFileReq, SoftUnlinkDirectoryReq, RenameDirectoryReq, HardUnlinkDirectoryReq, HardUnlinkFileReq> _data;
public:
CDCMessageKind kind() const { return _kind; }
const MakeDirectoryReq& getMakeDirectory() const;
MakeDirectoryReq& setMakeDirectory();
const RenameFileReq& getRenameFile() const;
RenameFileReq& setRenameFile();
const SoftUnlinkDirectoryReq& getSoftUnlinkDirectory() const;
SoftUnlinkDirectoryReq& setSoftUnlinkDirectory();
const RenameDirectoryReq& getRenameDirectory() const;
RenameDirectoryReq& setRenameDirectory();
const HardUnlinkDirectoryReq& getHardUnlinkDirectory() const;
HardUnlinkDirectoryReq& setHardUnlinkDirectory();
const HardUnlinkFileReq& getHardUnlinkFile() const;
HardUnlinkFileReq& setHardUnlinkFile();
void clear() { _kind = (CDCMessageKind)0; };
void pack(BincodeBuf& buf) const;
void unpack(BincodeBuf& buf, CDCMessageKind kind);
};
std::ostream& operator<<(std::ostream& out, const CDCReqContainer& x);
struct CDCRespContainer {
private:
CDCMessageKind _kind = (CDCMessageKind)0;
std::tuple<MakeDirectoryResp, RenameFileResp, SoftUnlinkDirectoryResp, RenameDirectoryResp, HardUnlinkDirectoryResp, HardUnlinkFileResp> _data;
public:
CDCMessageKind kind() const { return _kind; }
const MakeDirectoryResp& getMakeDirectory() const;
MakeDirectoryResp& setMakeDirectory();
const RenameFileResp& getRenameFile() const;
RenameFileResp& setRenameFile();
const SoftUnlinkDirectoryResp& getSoftUnlinkDirectory() const;
SoftUnlinkDirectoryResp& setSoftUnlinkDirectory();
const RenameDirectoryResp& getRenameDirectory() const;
RenameDirectoryResp& setRenameDirectory();
const HardUnlinkDirectoryResp& getHardUnlinkDirectory() const;
HardUnlinkDirectoryResp& setHardUnlinkDirectory();
const HardUnlinkFileResp& getHardUnlinkFile() const;
HardUnlinkFileResp& setHardUnlinkFile();
void clear() { _kind = (CDCMessageKind)0; };
void pack(BincodeBuf& buf) const;
void unpack(BincodeBuf& buf, CDCMessageKind kind);
};
std::ostream& operator<<(std::ostream& out, const CDCRespContainer& x);
enum class ShardLogEntryKind : uint16_t {
CONSTRUCT_FILE = 1,
LINK_FILE = 2,
@@ -2279,18 +2345,16 @@ enum class ShardLogEntryKind : uint16_t {
std::ostream& operator<<(std::ostream& out, ShardLogEntryKind err);
struct ConstructFileEntry {
InodeId id;
uint8_t type;
EggsTime deadlineTime;
BincodeBytes note;
static constexpr uint16_t STATIC_SIZE = 8 + 1 + 8 + BincodeBytes::STATIC_SIZE; // id + type + deadlineTime + note
static constexpr uint16_t STATIC_SIZE = 1 + 8 + BincodeBytes::STATIC_SIZE; // type + deadlineTime + note
ConstructFileEntry() { clear(); }
uint16_t packedSize() const {
uint16_t _size = 0;
_size += 8; // id
_size += 1; // type
_size += 8; // deadlineTime
_size += note.packedSize(); // note

139
cpp/RocksDBUtils.hpp
View File

@@ -79,3 +79,142 @@ public:
return snapshot;
}
};
// We use byteswap to go from LE to BE, and vice-versa.
static_assert(std::endian::native == std::endian::little);
inline uint64_t byteswapU64(uint64_t x) {
return __builtin_bswap64(x);
}
template<typename T>
struct StaticValue {
private:
std::array<char, T::MAX_SIZE> _data;
T _val;
public:
StaticValue() {
_val._data = &_data[0];
}
rocksdb::Slice toSlice() const {
return rocksdb::Slice(&_data[0], _val.size());
}
T* operator->() {
return &_val;
}
};
template<typename T>
struct ExternalValue {
private:
T _val;
public:
ExternalValue() {
_val._data = nullptr;
}
ExternalValue(char* data, size_t size) {
_val._data = data;
_val.checkSize(size);
}
ExternalValue(std::string& s): ExternalValue(s.data(), s.size()) {}
static const ExternalValue<T> FromSlice(const rocksdb::Slice& slice) {
return ExternalValue((char*)slice.data(), slice.size());
}
T* operator->() {
return &_val;
}
rocksdb::Slice toSlice() {
return rocksdb::Slice(_val._data, _val.size());
}
};
template<typename T>
struct OwnedValue {
T _val;
public:
OwnedValue() = delete;
template<typename ...Args>
OwnedValue(Args&&... args) {
size_t sz = T::calcSize(std::forward<Args>(args)...);
_val._data = (char*)malloc(sz);
ALWAYS_ASSERT(_val._data);
}
~OwnedValue() {
free(_val._data);
}
rocksdb::Slice toSlice() const {
return rocksdb::Slice(_val._data, _val.size());
}
T* operator->() {
return &_val;
}
};
#define LE_VAL(type, name, setName, offset) \
static_assert(sizeof(type) > 1); \
type name() const { \
type x; \
memcpy(&x, _data+offset, sizeof(x)); \
return x; \
} \
void setName(type x) { \
memcpy(_data+offset, &x, sizeof(x)); \
}
#define U8_VAL(type, name, setName, offset) \
static_assert(sizeof(type) == sizeof(uint8_t)); \
type name() const { \
type x; \
memcpy(&x, _data+offset, sizeof(x)); \
return x; \
} \
void setName(type x) { \
memcpy(_data+offset, &x, sizeof(x)); \
}
#define BYTES_VAL(name, setName, offset) \
const BincodeBytes name() const { \
BincodeBytes bs; \
bs.length = (uint8_t)(int)*(_data+offset); \
bs.data = (const uint8_t*)(_data+offset+1); \
return bs; \
} \
void setName(const BincodeBytes& bs) { \
*(_data+offset) = (char)(int)bs.length; \
memcpy(_data+offset+1, bs.data, bs.length); \
}
#define FBYTES_VAL(sz, getName, setName, offset) \
void getName(std::array<uint8_t, sz>& bs) const { \
memcpy(&bs[0], _data+offset, sz); \
} \
void setName(const std::array<uint8_t, sz>& bs) { \
memcpy(_data+offset, &bs[0], sz); \
}
#define BE64_VAL(type, name, setName, offset) \
static_assert(sizeof(type) == sizeof(uint64_t)); \
type name() const { \
uint64_t x; \
memcpy(&x, _data+offset, sizeof(x)); \
x = byteswapU64(x); /* BE -> LE */ \
type v; \
memcpy(&v, &x, sizeof(uint64_t)); \
return v; \
} \
void setName(type v) { \
uint64_t x; \
memcpy(&x, &v, sizeof(uint64_t)); \
x = byteswapU64(x); /* LE -> BE */ \
memcpy(_data+offset, &x, sizeof(x)); \
}

161
cpp/ShardDB.cpp
View File

@@ -20,6 +20,7 @@
#include "Time.hpp"
#include "RocksDBUtils.hpp"
#include "ShardDBData.hpp"
#include "AssertiveLock.hpp"
// TODO maybe revisit all those ALWAYS_ASSERTs
// TODO can we pre-allocate the value strings rather than allocating each stupid time?
@@ -29,11 +30,13 @@
// Right now, we're not implementing distributed consensus, but we want to prepare
// for this as best we can, since we need it for HA.
//
// We use two RocksDB database: one for the log entries, one for the state. When
// a request comes in, we first generate and persist a log entry for it. The log
// entry can then be applied to the state when consensus has been reached on it.
// After the log entry has been applied to the state, a response is sent to the
// client.
// This class implements the state storage, assuming that log entries will be persisted
// separatedly (if at all). The state storage is also used to prepare the log entries --
// e.g. there's a tiny bit of state that we read/write to without regards for the
// distributed consensus.
//
// Specifically, we allocate fresh ids, and we read info about block services, when
// preparing log entries.
//
// One question is how we write transactions concurrently, given that we rely on
// applying first one log entry and then the next in that order. We could be smart
@@ -41,28 +44,7 @@
// that it's probably not worth doing this, at least since the beginning, and that
// it's better to just serialize the transactions manually with a mutex here in C++.
//
// ## RocksDB schema (log)
//
// Two column families: the default one storing metadata (e.g. the Raft state,
// at some point), and the "log" column family storing the log entries. Separate
// column families since we'll keep deleting large ranges from the beginning of
// the log CF, and therefore seeking at the beginning of that won't be nice, so
// we don't want to put stuff at the beginning, but we also want to easily know
// when we're done, so we don't want to put stuff at the end either.
//
// The log CF is just uint64_t log index to serialized log entry.
//
// For now, in a pre-raft word, the default CF needs:
//
// * LogMetadataKey::NEXT_FILE_ID: the next available file id
// * LogMetadataKey::NEXT_SYMLINK_ID: the next available symlink id
// * LogMetadataKey::NEXT_BLOCK_ID: the next available block id
//
// When preparing log entries we also allocate blocks, and the blocks are stored
// in the state db, but we never modify the block state when adding a log entry,
// so we can safely read them.
//
// ## RocksDB schema (state)
// ## RocksDB schema
//
// Overview of the column families we use. Assumptions from the document:
//
@@ -156,39 +138,6 @@
//
// TODO fill in results
struct AssertiveLocked {
private:
std::atomic<bool>& _held;
public:
AssertiveLocked(std::atomic<bool>& held): _held(held) {
bool expected = false;
if (!_held.compare_exchange_strong(expected, true)) {
throw EGGS_EXCEPTION("could not aquire lock for applyLogEntry, are you using this function concurrently?");
}
}
AssertiveLocked(const AssertiveLocked&) = delete;
AssertiveLocked& operator=(AssertiveLocked&) = delete;
~AssertiveLocked() {
_held.store(false);
}
};
struct AssertiveLock {
private:
std::atomic<bool> _held;
public:
AssertiveLock(): _held(false) {}
AssertiveLock(const AssertiveLock&) = delete;
AssertiveLock& operator=(AssertiveLock&) = delete;
AssertiveLocked lock() {
return AssertiveLocked(_held);
}
};
static uint64_t computeHash(HashMode mode, const BincodeBytes& bytes) {
switch (mode) {
case HashMode::XXH3_63:
@@ -221,7 +170,7 @@ void ShardLogEntry::pack(BincodeBuf& buf) const {
}
void ShardLogEntry::unpack(BincodeBuf& buf) {
time = buf.unpackScalar<uint64_t>();
time.unpack(buf);
ShardLogEntryKind kind = (ShardLogEntryKind)buf.unpackScalar<uint16_t>();
body.unpack(buf, kind);
}
@@ -240,7 +189,6 @@ struct ShardDBImpl {
rocksdb::ColumnFamilyHandle* _directoriesCf;
rocksdb::ColumnFamilyHandle* _edgesCf;
AssertiveLock _prepareLogEntryLock;
AssertiveLock _applyLogEntryLock;
// ----------------------------------------------------------------
@@ -264,8 +212,8 @@ struct ShardDBImpl {
{"edges", {}},
};
std::vector<rocksdb::ColumnFamilyHandle*> familiesHandles;
auto dbPath = path + "/db-state";
LOG_INFO(_env, "initializing state RocksDB in %s", dbPath);
auto dbPath = path + "/db";
LOG_INFO(_env, "initializing shard %s RocksDB in %s", _shid, dbPath);
ROCKS_DB_CHECKED_MSG(
rocksdb::DB::Open(options, dbPath, familiesDescriptors, &familiesHandles, &_db),
"could not open RocksDB %s", dbPath
@@ -281,6 +229,22 @@ struct ShardDBImpl {
_initDb();
}
void close() {
LOG_INFO(_env, "destroying column families and closing database");
ROCKS_DB_CHECKED(_db->DestroyColumnFamilyHandle(_defaultCf));
ROCKS_DB_CHECKED(_db->DestroyColumnFamilyHandle(_filesCf));
ROCKS_DB_CHECKED(_db->DestroyColumnFamilyHandle(_spansCf));
ROCKS_DB_CHECKED(_db->DestroyColumnFamilyHandle(_transientCf));
ROCKS_DB_CHECKED(_db->DestroyColumnFamilyHandle(_directoriesCf));
ROCKS_DB_CHECKED(_db->DestroyColumnFamilyHandle(_edgesCf));
ROCKS_DB_CHECKED(_db->Close());
}
~ShardDBImpl() {
delete _db;
}
void _initDb() {
bool shardInfoExists;
{
@@ -355,22 +319,6 @@ struct ShardDBImpl {
}
}
void close() {
LOG_INFO(_env, "destroying column families and closing database");
ROCKS_DB_CHECKED(_db->DestroyColumnFamilyHandle(_defaultCf));
ROCKS_DB_CHECKED(_db->DestroyColumnFamilyHandle(_filesCf));
ROCKS_DB_CHECKED(_db->DestroyColumnFamilyHandle(_spansCf));
ROCKS_DB_CHECKED(_db->DestroyColumnFamilyHandle(_transientCf));
ROCKS_DB_CHECKED(_db->DestroyColumnFamilyHandle(_directoriesCf));
ROCKS_DB_CHECKED(_db->DestroyColumnFamilyHandle(_edgesCf));
ROCKS_DB_CHECKED(_db->Close());
}
~ShardDBImpl() {
delete _db;
}
// ----------------------------------------------------------------
// read-only path
@@ -684,21 +632,8 @@ struct ShardDBImpl {
// ----------------------------------------------------------------
// log preparation
EggsError _prepareConstructFile(rocksdb::WriteBatch& batch, EggsTime time, const ConstructFileReq& req, BincodeBytesScratchpad& scratch, ConstructFileEntry& entry) {
const auto nextFileId = [this, &batch](const ShardMetadataKey* key) -> InodeId {
std::string value;
ROCKS_DB_CHECKED(_db->Get({}, shardMetadataKey(key), &value));
ExternalValue<InodeIdValue> inodeId(value);
inodeId->setId(InodeId::FromU64(inodeId->id().u64 + 0x100));
ROCKS_DB_CHECKED(batch.Put(shardMetadataKey(key), inodeId.toSlice()));
return inodeId->id();
};
if (req.type == (uint8_t)InodeType::FILE) {
entry.id = nextFileId(&NEXT_FILE_ID_KEY);
} else if (req.type == (uint8_t)InodeType::SYMLINK) {
entry.id = nextFileId(&NEXT_SYMLINK_ID_KEY);
} else {
EggsError _prepareConstructFile(EggsTime time, const ConstructFileReq& req, BincodeBytesScratchpad& scratch, ConstructFileEntry& entry) {
if (req.type != (uint8_t)InodeType::FILE && req.type != (uint8_t)InodeType::SYMLINK) {
return EggsError::TYPE_IS_DIRECTORY;
}
entry.type = req.type;
@@ -947,20 +882,17 @@ struct ShardDBImpl {
EggsError prepareLogEntry(const ShardReqContainer& req, BincodeBytesScratchpad& scratch, ShardLogEntry& logEntry) {
LOG_DEBUG(_env, "processing write request of kind %s", req.kind());
auto locked = _prepareLogEntryLock.lock();
scratch.reset();
logEntry.clear();
EggsError err = NO_ERROR;
rocksdb::WriteBatch batch;
EggsTime time = eggsNow().ns;
EggsTime time = eggsNow();
logEntry.time = time;
auto& logEntryBody = logEntry.body;
switch (req.kind()) {
case ShardMessageKind::CONSTRUCT_FILE:
err = _prepareConstructFile(batch, time, req.getConstructFile(), scratch, logEntryBody.setConstructFile());
err = _prepareConstructFile(time, req.getConstructFile(), scratch, logEntryBody.setConstructFile());
break;
case ShardMessageKind::LINK_FILE:
err = _prepareLinkFile(time, req.getLinkFile(), scratch, logEntryBody.setLinkFile());
@@ -1017,11 +949,6 @@ struct ShardDBImpl {
if (err == NO_ERROR) {
LOG_DEBUG(_env, "prepared log entry of kind %s, for request of kind %s", logEntryBody.kind(), req.kind());
// Here we must always sync, we can't lose updates to the next file id (or similar)
// which then might be persisted in the log but lost here.
rocksdb::WriteOptions options;
options.sync = true;
ROCKS_DB_CHECKED(_db->Write(options, &batch));
} else {
LOG_INFO(_env, "could not prepare log entry for request of kind %s: %s", req.kind(), err);
}
@@ -1042,6 +969,23 @@ struct ShardDBImpl {
}
EggsError _applyConstructFile(rocksdb::WriteBatch& batch, EggsTime time, const ConstructFileEntry& entry, BincodeBytesScratchpad& scratch, ConstructFileResp& resp) {
const auto nextFileId = [this, &batch](const ShardMetadataKey* key) -> InodeId {
std::string value;
ROCKS_DB_CHECKED(_db->Get({}, shardMetadataKey(key), &value));
ExternalValue<InodeIdValue> inodeId(value);
inodeId->setId(InodeId::FromU64(inodeId->id().u64 + 0x100));
ROCKS_DB_CHECKED(batch.Put(shardMetadataKey(key), inodeId.toSlice()));
return inodeId->id();
};
InodeId id;
if (entry.type == (uint8_t)InodeType::FILE) {
id = nextFileId(&NEXT_FILE_ID_KEY);
} else if (entry.type == (uint8_t)InodeType::SYMLINK) {
id = nextFileId(&NEXT_SYMLINK_ID_KEY);
} else {
ALWAYS_ASSERT(false, "Bad type %s", (int)entry.type);
}
// write to rocks
StaticValue<TransientFileBody> transientFile;
transientFile->setFileSize(0);
@@ -1049,11 +993,11 @@ struct ShardDBImpl {
transientFile->setDeadline(entry.deadlineTime);
transientFile->setLastSpanState(SpanState::CLEAN);
transientFile->setNote(entry.note);
auto k = InodeIdKey::Static(entry.id);
auto k = InodeIdKey::Static(id);
ROCKS_DB_CHECKED(batch.Put(_transientCf, k.toSlice(), transientFile.toSlice()));
// prepare response
resp.id = entry.id;
resp.id = id;
_calcCookie(resp.id, resp.cookie.data);
return NO_ERROR;
@@ -1587,6 +1531,7 @@ struct ShardDBImpl {
// set the owner, and restore the directory entry if needed.
StaticValue<DirectoryBody> newDir;
newDir->setOwnerId(entry.ownerId);
newDir->setMtime(dir->mtime());
newDir->setHashMode(dir->hashMode());
newDir->setInfoInherited(dir->infoInherited());
// remove directory info if it was inherited, now that we have an
@@ -1953,7 +1898,7 @@ void ShardDB::close() {
}
ShardDB::~ShardDB() {
delete ((ShardDBImpl*)_impl);
delete (ShardDBImpl*)_impl;
_impl = nullptr;
}

23
cpp/ShardDB.hpp
View File

@@ -2,7 +2,6 @@
#include "Common.hpp"
#include "Msgs.hpp"
#include "MsgsGen.hpp"
#include "Env.hpp"
struct ShardLogEntry {
@@ -45,24 +44,22 @@ public:
// Prepares and persists a log entry to be applied.
//
// This function will fail if called concurrently. We do write to the database
// to prepare the log entry, to generate fresh file/block ids.
// This function can be called concurrently. We only read from the database to
// do this. Note that the reading is limited to data which is OK to be a bit
// stale, such as block service information, and which is not used in a way
// which relies on the log entries being successfully applied.
//
// The log entry is not persisted in any way -- the idea is that a consensus module
// can be built on top of this to first persist the log entry across a consensus
// of machines, and then the log entries can be applied.
//
// Note that while we write to the database to produce log entries, we do so in a
// way that allows log entries to be dropped without damaging the consistency of
// the system (again we just need to do fresh generation of)
//
// Morally this function always succeeds, the "real" error checking is all done at
// log application time. The reasoning here is that
// log preparation is not reading from the latest state anyway, since there might
// be many log entries in flight which we have prepared but not applied, and therefore
// this function does not have the latest view of the state. We are still
// allowed to read from the state at log preparation time (e.g. to gather needed info
// about block services), but knowing that we might lag a bit behind.
// log application time. The reasoning here is that log preparation is not reading
// from the latest state anyway, since there might be many log entries in flight which
// we have prepared but not applied, and therefore this function does not have the latest
// view of the state. We are still allowed to read from the state at log preparation time
// (e.g. to gather needed info about block services), but knowing that we might lag a bit
// behind.
//
// However we still do some "type checking" here (e.g. we have ids in the right
// shard and of the right type, good transient file cookies), and we might still

176
cpp/ShardDBData.hpp
View File

@@ -10,146 +10,10 @@
#include "Msgs.hpp"
#include "MsgsGen.hpp"
#include "Time.hpp"
// We use byteswap to go from LE to BE.
static_assert(std::endian::native == std::endian::little);
inline uint64_t byteswapU64(uint64_t x) {
return __builtin_bswap64(x);
}
template<typename T>
struct StaticValue {
private:
std::array<char, T::MAX_SIZE> _data;
T _val;
public:
StaticValue() {
_val.data = &_data[0];
}
rocksdb::Slice toSlice() const {
return rocksdb::Slice(&_data[0], _val.size());
}
T* operator->() {
return &_val;
}
};
template<typename T>
struct ExternalValue {
private:
T _val;
public:
ExternalValue() {
_val.data = nullptr;
}
ExternalValue(char* data, size_t size) {
_val.data = data;
_val.checkSize(size);
}
ExternalValue(std::string& s): ExternalValue(s.data(), s.size()) {}
static const ExternalValue<T> FromSlice(const rocksdb::Slice& slice) {
return ExternalValue((char*)slice.data(), slice.size());
}
T* operator->() {
return &_val;
}
rocksdb::Slice toSlice() {
return rocksdb::Slice(_val.data, _val.size());
}
};
template<typename T>
struct OwnedValue {
T _val;
public:
OwnedValue() = delete;
template<typename ...Args>
OwnedValue(Args&&... args) {
size_t sz = T::calcSize(std::forward<Args>(args)...);
_val.data = (char*)malloc(sz);
ALWAYS_ASSERT(_val.data);
}
~OwnedValue() {
free(_val.data);
}
rocksdb::Slice toSlice() const {
return rocksdb::Slice(_val.data, _val.size());
}
T* operator->() {
return &_val;
}
};
#define LE_VAL(type, name, setName, offset) \
static_assert(sizeof(type) > 1); \
type name() const { \
type x; \
memcpy(&x, data+offset, sizeof(x)); \
return x; \
} \
void setName(type x) { \
memcpy(data+offset, &x, sizeof(x)); \
}
#define U8_VAL(type, name, setName, offset) \
static_assert(sizeof(type) == sizeof(uint8_t)); \
type name() const { \
type x; \
memcpy(&x, data+offset, sizeof(x)); \
return x; \
} \
void setName(type x) { \
memcpy(data+offset, &x, sizeof(x)); \
}
#define BYTES_VAL(name, setName, offset) \
const BincodeBytes name() const { \
BincodeBytes bs; \
bs.length = (uint8_t)(int)*(data+offset); \
bs.data = (const uint8_t*)(data+offset+1); \
return bs; \
} \
void setName(const BincodeBytes& bs) { \
*(data+offset) = (char)(int)bs.length; \
memcpy(data+offset+1, bs.data, bs.length); \
}
#define FBYTES_VAL(sz, getName, setName, offset) \
void getName(std::array<uint8_t, sz>& bs) const { \
memcpy(&bs[0], data+offset, sz); \
} \
void setName(const std::array<uint8_t, sz>& bs) { \
memcpy(data+offset, &bs[0], sz); \
}
#define BE64_VAL(type, name, setName, offset) \
static_assert(sizeof(type) == sizeof(uint64_t)); \
type name() const { \
uint64_t x; \
memcpy(&x, data+offset, sizeof(x)); \
x = byteswapU64(x); /* BE -> LE */ \
type v; \
memcpy(&v, &x, sizeof(uint64_t)); \
return v; \
} \
void setName(type v) { \
uint64_t x; \
memcpy(&x, &v, sizeof(uint64_t)); \
x = byteswapU64(x); /* LE -> BE */ \
memcpy(data+offset, &x, sizeof(x)); \
}
#include "RocksDBUtils.hpp"
struct InodeIdValue {
char* data;
char* _data;
static constexpr size_t MAX_SIZE = sizeof(InodeId);
size_t size() const { return MAX_SIZE; }
@@ -166,7 +30,7 @@ struct InodeIdValue {
// When we need a simple u64 value (e.g. log index)
struct U64Value {
char* data;
char* _data;
static constexpr size_t MAX_SIZE = sizeof(uint64_t);
size_t size() const { return MAX_SIZE; }
@@ -200,7 +64,7 @@ inline rocksdb::Slice shardMetadataKey(const ShardMetadataKey* k) {
}
struct ShardInfoBody {
char* data;
char* _data;
static constexpr size_t MAX_SIZE =
sizeof(ShardId) + // shardId
@@ -217,7 +81,7 @@ struct ShardInfoBody {
// (and therefore BE), but it does make it a bit nicer to be able to traverse
// them like that.
struct InodeIdKey {
char* data;
char* _data;
static constexpr size_t MAX_SIZE = sizeof(InodeId);
size_t size() const { return MAX_SIZE; }
@@ -241,7 +105,7 @@ enum class SpanState : uint8_t {
std::ostream& operator<<(std::ostream& out, SpanState state);
struct TransientFileBody {
char* data;
char* _data;
static constexpr size_t MIN_SIZE =
sizeof(uint64_t) + // size
@@ -268,7 +132,7 @@ struct TransientFileBody {
};
struct FileBody {
char* data;
char* _data;
static constexpr size_t MAX_SIZE =
sizeof(uint64_t) + // size
@@ -281,7 +145,7 @@ struct FileBody {
};
struct SpanKey {
char* data;
char* _data;
static constexpr size_t MAX_SIZE =
sizeof(InodeId) + // id
@@ -299,7 +163,7 @@ struct PACKED BlockBody {
};
struct SpanBody {
char* data;
char* _data;
static constexpr size_t MIN_SIZE =
sizeof(uint64_t) + // size
@@ -340,7 +204,7 @@ struct SpanBody {
if (storageClass() == ZERO_FILL_STORAGE) {
// no-op
} else if (storageClass() == INLINE_STORAGE) {
sz += 1 + (uint8_t)(int)*(data + MIN_SIZE);
sz += 1 + (uint8_t)(int)*(_data + MIN_SIZE);
} else {
sz += BLOCK_BODY_SIZE * parity().blocks();
}
@@ -356,26 +220,26 @@ struct SpanBody {
ALWAYS_ASSERT(storageClass() == INLINE_STORAGE);
size_t offset = MIN_SIZE;
BincodeBytes bs;
bs.length = (uint8_t)(int)*(data+offset);
bs.data = (const uint8_t*)(data+offset+1);
bs.length = (uint8_t)(int)*(_data+offset);
bs.data = (const uint8_t*)(_data+offset+1);
return bs;
}
void setInlineBody(const BincodeBytes& bs) {
ALWAYS_ASSERT(storageClass() == INLINE_STORAGE);
size_t offset = MIN_SIZE;
*(data+offset) = (char)(int)bs.length;
memcpy(data+offset+1, bs.data, bs.length);
*(_data+offset) = (char)(int)bs.length;
memcpy(_data+offset+1, bs.data, bs.length);
}
void block(uint64_t ix, BlockBody& b) const {
ALWAYS_ASSERT(storageClass() != ZERO_FILL_STORAGE && storageClass() != INLINE_STORAGE);
ALWAYS_ASSERT(ix < parity().blocks());
memcpy(&b, data + MIN_SIZE + ix * sizeof(BlockBody), sizeof(BlockBody));
memcpy(&b, _data + MIN_SIZE + ix * sizeof(BlockBody), sizeof(BlockBody));
}
void setBlock(uint64_t ix, const BlockBody& b) {
ALWAYS_ASSERT(storageClass() != ZERO_FILL_STORAGE && storageClass() != INLINE_STORAGE);
ALWAYS_ASSERT(ix < parity().blocks());
memcpy(data + MIN_SIZE + ix * sizeof(BlockBody), &b, sizeof(BlockBody));
memcpy(_data + MIN_SIZE + ix * sizeof(BlockBody), &b, sizeof(BlockBody));
}
};
@@ -385,7 +249,7 @@ enum class HashMode : uint8_t {
};
struct DirectoryBody {
char* data;
char* _data;
// `infoInherited` tells us whether we should inherit the info from the
// parent directory. Moreover, we might have a directory with no owner
@@ -436,7 +300,7 @@ struct DirectoryBody {
};
struct EdgeKey {
char* data;
char* _data;
static constexpr size_t MIN_SIZE =
sizeof(uint64_t) + // first 63 bits: dirId, then whether the edge is current
@@ -493,7 +357,7 @@ struct EdgeKey {
std::ostream& operator<<(std::ostream& out, const EdgeKey& edgeKey);
struct SnapshotEdgeBody {
char* data;
char* _data;
static constexpr size_t MAX_SIZE =
sizeof(InodeIdExtra); // target, and if owned
@@ -504,7 +368,7 @@ struct SnapshotEdgeBody {
};
struct CurrentEdgeBody {
char* data;
char* _data;
static constexpr size_t MAX_SIZE =
sizeof(InodeIdExtra) + // target, and if locked

67
cpp/tests.app.cpp
View File

@@ -426,7 +426,8 @@ TEST_CASE("touch file") {
auto reqContainer = std::make_unique<ShardReqContainer>();
auto respContainer = std::make_unique<ShardRespContainer>();
auto logEntry = std::make_unique<ShardLogEntryWithIndex>();
auto logEntry = std::make_unique<ShardLogEntry>();
uint64_t logEntryIndex = 0;
BincodeBytesScratchpad dbScratch;
BincodeBytesScratchpad ourScratch;
@@ -440,8 +441,8 @@ TEST_CASE("touch file") {
req.type = (uint8_t)InodeType::FILE;
ourScratch.copyTo("test note", req.note);
NO_EGGS_ERROR(db->prepareLogEntry(*reqContainer, dbScratch, *logEntry));
constructTime = logEntry->entry.time;
NO_EGGS_ERROR(db->applyLogEntry(logEntry->index, dbScratch, *respContainer));
constructTime = logEntry->time;
NO_EGGS_ERROR(db->applyLogEntry(true, ++logEntryIndex, *logEntry, dbScratch, *respContainer));
auto& resp = respContainer->getConstructFile();
id = resp.id;
cookie = resp.cookie;
@@ -463,8 +464,8 @@ TEST_CASE("touch file") {
req.ownerId = ROOT_DIR_INODE_ID;
ourScratch.copyTo(name, req.name);
NO_EGGS_ERROR(db->prepareLogEntry(*reqContainer, dbScratch, *logEntry));
linkTime = logEntry->entry.time;
NO_EGGS_ERROR(db->applyLogEntry(logEntry->index, dbScratch, *respContainer));
linkTime = logEntry->time;
NO_EGGS_ERROR(db->applyLogEntry(true, ++logEntryIndex, *logEntry, dbScratch, *respContainer));
}
{
ourScratch.reset();
@@ -499,50 +500,13 @@ TEST_CASE("touch file") {
}
}
TEST_CASE("log replay") {
TempShardDB db(LogLevel::LOG_ERROR, ShardId(0));
auto reqContainer = std::make_unique<ShardReqContainer>();
auto respContainer = std::make_unique<ShardRespContainer>();
auto logEntry = std::make_unique<ShardLogEntryWithIndex>();
BincodeBytesScratchpad dbScratch;
BincodeBytesScratchpad ourScratch;
// first just try a simple restart
std::array<uint8_t, 16> secretKeyBefore = db.db->secretKey();
db.restart();
REQUIRE(secretKeyBefore == db.db->secretKey());
for (int i = 0; i < 2; i++) {
// create log entry, but don't apply
{
ourScratch.reset();
auto& req = reqContainer->setConstructFile();
req.type = (uint8_t)InodeType::FILE;
ourScratch.copyTo("test note", req.note);
NO_EGGS_ERROR(db->prepareLogEntry(*reqContainer, dbScratch, *logEntry));
}
// now restart, the log above will be automatically replayed
db.restart();
// verify that it is
{
auto& req = reqContainer->setVisitTransientFiles();
NO_EGGS_ERROR(db->read(*reqContainer, dbScratch, *respContainer));
auto& resp = respContainer->getVisitTransientFiles();
REQUIRE(resp.nextId == NULL_INODE_ID);
REQUIRE(resp.files.els.size() == i+1);
}
}
}
TEST_CASE("override") {
TempShardDB db(LogLevel::LOG_ERROR, ShardId(0));
auto reqContainer = std::make_unique<ShardReqContainer>();
auto respContainer = std::make_unique<ShardRespContainer>();
auto logEntry = std::make_unique<ShardLogEntryWithIndex>();
auto logEntry = std::make_unique<ShardLogEntry>();
uint64_t logEntryIndex = 0;
BincodeBytesScratchpad dbScratch;
BincodeBytesScratchpad ourScratch;
@@ -555,7 +519,7 @@ TEST_CASE("override") {
req.type = (uint8_t)InodeType::FILE;
ourScratch.copyTo("test note", req.note);
NO_EGGS_ERROR(db->prepareLogEntry(*reqContainer, dbScratch, *logEntry));
NO_EGGS_ERROR(db->applyLogEntry(logEntry->index, dbScratch, *respContainer));
NO_EGGS_ERROR(db->applyLogEntry(true, ++logEntryIndex, *logEntry, dbScratch, *respContainer));
auto& resp = respContainer->getConstructFile();
id = resp.id;
cookie = resp.cookie;
@@ -572,7 +536,7 @@ TEST_CASE("override") {
req.ownerId = ROOT_DIR_INODE_ID;
ourScratch.copyTo(name, req.name);
NO_EGGS_ERROR(db->prepareLogEntry(*reqContainer, dbScratch, *logEntry));
NO_EGGS_ERROR(db->applyLogEntry(logEntry->index, dbScratch, *respContainer));
NO_EGGS_ERROR(db->applyLogEntry(true, ++logEntryIndex, *logEntry, dbScratch, *respContainer));
}
return id;
};
@@ -588,7 +552,7 @@ TEST_CASE("override") {
ourScratch.copyTo("foo", req.oldName);
ourScratch.copyTo("bar", req.newName);
NO_EGGS_ERROR(db->prepareLogEntry(*reqContainer, dbScratch, *logEntry));
NO_EGGS_ERROR(db->applyLogEntry(logEntry->index, dbScratch, *respContainer));
NO_EGGS_ERROR(db->applyLogEntry(true, ++logEntryIndex, *logEntry, dbScratch, *respContainer));
}
{
auto& req = reqContainer->setFullReadDir();
@@ -617,7 +581,8 @@ TEST_CASE("make/rm directory") {
auto reqContainer = std::make_unique<ShardReqContainer>();
auto respContainer = std::make_unique<ShardRespContainer>();
auto logEntry = std::make_unique<ShardLogEntryWithIndex>();
auto logEntry = std::make_unique<ShardLogEntry>();
uint64_t logEntryIndex = 0;
BincodeBytesScratchpad dbScratch;
BincodeBytesScratchpad ourScratch;
@@ -631,7 +596,7 @@ TEST_CASE("make/rm directory") {
req.info.inherited = true;
req.ownerId = ROOT_DIR_INODE_ID;
NO_EGGS_ERROR(db->prepareLogEntry(*reqContainer, dbScratch, *logEntry));
NO_EGGS_ERROR(db->applyLogEntry(logEntry->index, dbScratch, *respContainer));
NO_EGGS_ERROR(db->applyLogEntry(true, ++logEntryIndex, *logEntry, dbScratch, *respContainer));
respContainer->getCreateDirectoryInode();
}
{
@@ -639,7 +604,7 @@ TEST_CASE("make/rm directory") {
req.dirId = id;
req.info = defaultInfo;
NO_EGGS_ERROR(db->prepareLogEntry(*reqContainer, dbScratch, *logEntry));
NO_EGGS_ERROR(db->applyLogEntry(logEntry->index, dbScratch, *respContainer));
NO_EGGS_ERROR(db->applyLogEntry(true, ++logEntryIndex, *logEntry, dbScratch, *respContainer));
}
{
auto& req = reqContainer->setStatDirectory();
@@ -653,6 +618,6 @@ TEST_CASE("make/rm directory") {
req.dirId = id;
req.ownerId = ROOT_DIR_INODE_ID;
NO_EGGS_ERROR(db->prepareLogEntry(*reqContainer, dbScratch, *logEntry));
NO_EGGS_ERROR(db->applyLogEntry(logEntry->index, dbScratch, *respContainer));
NO_EGGS_ERROR(db->applyLogEntry(true, ++logEntryIndex, *logEntry, dbScratch, *respContainer));
}
}

67
go/bincodegen/bincodegen.go
View File

@@ -954,9 +954,9 @@ func generateCppContainer(hpp io.Writer, cpp io.Writer, name string, kindTypeNam
fmt.Fprintf(cpp, "}\n\n")
}
func generateCppLogEntries(hpp io.Writer, cpp io.Writer, types []reflect.Type) {
func generateCppLogEntries(hpp io.Writer, cpp io.Writer, what string, types []reflect.Type) {
containerTypes := make([]containerType, len(types))
fmt.Fprintf(hpp, "enum class ShardLogEntryKind : uint16_t {\n")
fmt.Fprintf(hpp, "enum class %sLogEntryKind : uint16_t {\n", what)
for i, typ := range types {
fmt.Fprintf(hpp, " %s = %d,\n", enumName(typ), i+1) // skip 0 since we use it as an empty value
containerTypes[i].typ = typ
@@ -967,16 +967,16 @@ func generateCppLogEntries(hpp io.Writer, cpp io.Writer, types []reflect.Type) {
containerTypes[i].name = string([]byte(types[i].Name())[:len(types[i].Name())-len("Entry")])
}
fmt.Fprintf(hpp, "};\n\n")
fmt.Fprintf(hpp, "std::ostream& operator<<(std::ostream& out, ShardLogEntryKind err);\n\n")
fmt.Fprintf(cpp, "std::ostream& operator<<(std::ostream& out, ShardLogEntryKind err) {\n")
fmt.Fprintf(hpp, "std::ostream& operator<<(std::ostream& out, %sLogEntryKind err);\n\n", what)
fmt.Fprintf(cpp, "std::ostream& operator<<(std::ostream& out, %sLogEntryKind err) {\n", what)
fmt.Fprintf(cpp, " switch (err) {\n")
for _, typ := range containerTypes {
fmt.Fprintf(cpp, " case ShardLogEntryKind::%s:\n", typ.enum)
fmt.Fprintf(cpp, " case %sLogEntryKind::%s:\n", what, typ.enum)
fmt.Fprintf(cpp, " out << \"%s\";\n", typ.enum)
fmt.Fprintf(cpp, " break;\n")
}
fmt.Fprintf(cpp, " default:\n")
fmt.Fprintf(cpp, " out << \"ShardLogEntryKind(\" << ((int)err) << \")\";\n")
fmt.Fprintf(cpp, " out << \"%sLogEntryKind(\" << ((int)err) << \")\";\n", what)
fmt.Fprintf(cpp, " break;\n")
fmt.Fprintf(cpp, " }\n")
fmt.Fprintf(cpp, " return out;\n")
@@ -985,7 +985,29 @@ func generateCppLogEntries(hpp io.Writer, cpp io.Writer, types []reflect.Type) {
for _, typ := range types {
generateCppSingle(hpp, cpp, typ)
}
generateCppContainer(hpp, cpp, "ShardLogEntryContainer", "ShardLogEntryKind", containerTypes)
generateCppContainer(hpp, cpp, what+"LogEntryContainer", what+"LogEntryKind", containerTypes)
}
func generateCppReqResp(hpp io.Writer, cpp io.Writer, what string, reqResps []reqRespType) {
generateCppKind(hpp, cpp, what, reqResps)
reqContainerTypes := make([]containerType, len(reqResps))
for i, reqResp := range reqResps {
reqContainerTypes[i] = containerType{
name: string([]byte(reqResp.req.Name())[:len(reqResp.req.Name())-len("Req")]),
enum: reqRespEnum(reqResp),
typ: reqResp.req,
}
}
generateCppContainer(hpp, cpp, what+"ReqContainer", what+"MessageKind", reqContainerTypes)
respContainerTypes := make([]containerType, len(reqResps))
for i, reqResp := range reqResps {
respContainerTypes[i] = containerType{
name: string([]byte(reqResp.resp.Name())[:len(reqResp.resp.Name())-len("Resp")]),
enum: reqRespEnum(reqResp),
typ: reqResp.resp,
}
}
generateCppContainer(hpp, cpp, what+"RespContainer", what+"MessageKind", respContainerTypes)
}
func generateCpp(errors []string, shardReqResps []reqRespType, cdcReqResps []reqRespType, extras []reflect.Type) ([]byte, []byte) {
@@ -1035,29 +1057,13 @@ func generateCpp(errors []string, shardReqResps []reqRespType, cdcReqResps []req
generateCppSingle(hppOut, cppOut, reqResp.resp)
}
generateCppKind(hppOut, cppOut, "Shard", shardReqResps)
reqContainerTypes := make([]containerType, len(shardReqResps))
for i, reqResp := range shardReqResps {
reqContainerTypes[i] = containerType{
name: string([]byte(reqResp.req.Name())[:len(reqResp.req.Name())-len("Req")]),
enum: reqRespEnum(reqResp),
typ: reqResp.req,
}
}
generateCppContainer(hppOut, cppOut, "ShardReqContainer", "ShardMessageKind", reqContainerTypes)
respContainerTypes := make([]containerType, len(shardReqResps))
for i, reqResp := range shardReqResps {
respContainerTypes[i] = containerType{
name: string([]byte(reqResp.resp.Name())[:len(reqResp.resp.Name())-len("Resp")]),
enum: reqRespEnum(reqResp),
typ: reqResp.resp,
}
}
generateCppContainer(hppOut, cppOut, "ShardRespContainer", "ShardMessageKind", respContainerTypes)
generateCppReqResp(hppOut, cppOut, "Shard", shardReqResps)
generateCppReqResp(hppOut, cppOut, "CDC", cdcReqResps)
generateCppLogEntries(
hppOut,
cppOut,
"Shard",
[]reflect.Type{
reflect.TypeOf(msgs.ConstructFileEntry{}),
reflect.TypeOf(msgs.LinkFileEntry{}),
@@ -1077,6 +1083,15 @@ func generateCpp(errors []string, shardReqResps []reqRespType, cdcReqResps []req
},
)
/*
generateCppLogEntries(
hppOut,
cppOut,
"CDC",
[]reflect.Type{},
)
*/
return hppOut.Bytes(), cppOut.Bytes()
}

3
go/msgs/msgs.go
View File

@@ -802,7 +802,7 @@ type DirectoryInfoBody struct {
}
// --------------------------------------------------------------------
// log entries
// shard log entries
//
// these are only used internally, but we define them here for codegen
// simplicity.
@@ -812,7 +812,6 @@ type DirectoryInfoBody struct {
// and the index, so we don't also store it here.
type ConstructFileEntry struct {
Id InodeId
Type InodeType
DeadlineTime EggsTime
Note string