merge with edges, from and to are moved to the EdgeRecord, they are not copied along with properties inside the MVCC, TODO: graph interface, Accessor and Stores

This commit is contained in:
Kruno Tomola Fabro
2016-08-11 15:13:19 +01:00
280 changed files with 1457 additions and 967 deletions

View File

@@ -0,0 +1,26 @@
#pragma once
#include "communication/bolt/v1/states.hpp"
#include "io/network/socket.hpp"
#include "dbms/dbms.hpp"
namespace bolt
{
class Session;
class Bolt
{
friend class Session;
public:
Bolt();
Session* create_session(io::Socket&& socket);
void close(Session* session);
States states;
Dbms dbms;
};
}

View File

@@ -0,0 +1,14 @@
#pragma once
#include <cstddef>
namespace bolt
{
namespace config
{
static constexpr size_t N = 65535; /* chunk size */
static constexpr size_t C = N + 2; /* end mark */
}
}

View File

@@ -0,0 +1,46 @@
#pragma once
#include "utils/types/byte.hpp"
#include "utils/underlying_cast.hpp"
namespace bolt
{
enum class MessageCode : byte
{
Init = 0x01,
AckFailure = 0x0E,
Reset = 0x0F,
Run = 0x10,
DiscardAll = 0x2F,
PullAll = 0x3F,
Record = 0x71,
Success = 0x70,
Ignored = 0x7E,
Failure = 0x7F
};
inline bool operator==(byte value, MessageCode code)
{
return value == underlying_cast(code);
}
inline bool operator==(MessageCode code, byte value)
{
return operator==(value, code);
}
inline bool operator!=(byte value, MessageCode code)
{
return !operator==(value, code);
}
inline bool operator!=(MessageCode code, byte value)
{
return operator!=(value, code);
}
}

View File

@@ -0,0 +1,58 @@
#pragma once
#include <cstdint>
namespace bolt
{
namespace pack
{
enum Code : uint8_t
{
TinyString = 0x80,
TinyList = 0x90,
TinyMap = 0xA0,
TinyStruct = 0xB0,
Null = 0xC0,
Float64 = 0xC1,
False = 0xC2,
True = 0xC3,
Int8 = 0xC8,
Int16 = 0xC9,
Int32 = 0xCA,
Int64 = 0xCB,
Bytes8 = 0xCC,
Bytes16 = 0xCD,
Bytes32 = 0xCE,
String8 = 0xD0,
String16 = 0xD1,
String32 = 0xD2,
List8 = 0xD4,
List16 = 0xD5,
List32 = 0xD6,
Map8 = 0xD8,
Map16 = 0xD9,
Map32 = 0xDA,
MapStream = 0xDB,
Node = 0x4E,
Relationship = 0x52,
Path = 0x50,
Struct8 = 0xDC,
Struct16 = 0xDD,
EndOfStream = 0xDF,
};
}
}

View File

@@ -0,0 +1,21 @@
#pragma once
namespace bolt
{
enum class PackType
{
Null, // denotes absence of a value
Boolean, // denotes a type with two possible values (t/f)
Integer, // 64-bit signed integral number
Float, // 64-bit floating point number
Bytes, // binary data
String, // unicode string
List, // collection of values
Map, // collection of zero or more key/value pairs
Struct, // zero or more packstream values
EndOfStream, // denotes stream value end
Reserved // reserved for future use
};
}

View File

@@ -0,0 +1,157 @@
#pragma once
#include "communication/bolt/v1/transport/bolt_encoder.hpp"
#include "communication/bolt/v1/packing/codes.hpp"
#include "storage/vertex_accessor.hpp"
#include "storage/edge_accessor.hpp"
#include "storage/model/properties/properties.hpp"
#include "storage/model/properties/all.hpp"
namespace bolt
{
template <class Stream>
class BoltSerializer
{
friend class Property;
// TODO: here shoud be friend but it doesn't work
// template <class Handler>
// friend void accept(const Property &property, Handler &h);
public:
BoltSerializer(Stream& stream) : encoder(stream) {}
/* Serializes the vertex accessor into the packstream format
*
* struct[size = 3] Vertex [signature = 0x4E] {
* Integer node_id;
* List<String> labels;
* Map<String, Value> properties;
* }
*
*/
void write(const Vertex::Accessor& vertex)
{
// write signatures for the node struct and node data type
encoder.write_struct_header(3);
encoder.write(underlying_cast(pack::Node));
// write the identifier for the node
encoder.write_integer(vertex.id());
// write the list of labels
auto labels = vertex.labels();
encoder.write_list_header(labels.size());
for(auto& label : labels)
encoder.write_string(label.get());
// write the property map
auto props = vertex.properties();
encoder.write_map_header(props.size());
for(auto& prop : props) {
write(prop.first);
write(*prop.second);
}
}
/* Serializes the vertex accessor into the packstream format
*
* struct[size = 5] Edge [signature = 0x52] {
* Integer edge_id;
* Integer start_node_id;
* Integer end_node_id;
* String type;
* Map<String, Value> properties;
* }
*
*/
void write(const Edge::Accessor& edge)
{
// write signatures for the edge struct and edge data type
encoder.write_struct_header(5);
encoder.write(underlying_cast(pack::Relationship));
// write the identifier for the node
encoder.write_integer(edge.id());
// TODO refactor when from() and to() start returning Accessors
encoder.write_integer(edge.from()->id);
encoder.write_integer(edge.to()->id);
// write the type of the edge
encoder.write_string(edge.edge_type());
// write the property map
auto props = edge.properties();
encoder.write_map_header(props.size());
for(auto& prop : props) {
write(prop.first);
write(*prop.second);
}
}
void write(const Property& prop)
{
accept(prop, *this);
}
void write_null()
{
encoder.write_null();
}
void write(const Bool& prop)
{
encoder.write_bool(prop.value());
}
void write(const Float& prop)
{
encoder.write_double(prop.value);
}
void write(const Double& prop)
{
encoder.write_double(prop.value);
}
void write(const Int32& prop)
{
encoder.write_integer(prop.value);
}
void write(const Int64& prop)
{
encoder.write_integer(prop.value);
}
void write(const std::string& value)
{
encoder.write_string(value);
}
void write(const String& prop)
{
encoder.write_string(prop.value);
}
template <class T>
void handle(const T& prop)
{
write(prop);
}
protected:
Stream& encoder;
};
}

View File

@@ -0,0 +1,130 @@
#pragma once
#include "communication/bolt/v1/serialization/bolt_serializer.hpp"
#include "communication/bolt/v1/transport/chunked_buffer.hpp"
#include "communication/bolt/v1/transport/chunked_encoder.hpp"
#include "communication/bolt/v1/transport/socket_stream.hpp"
#include "logging/default.hpp"
namespace bolt
{
// compiled queries have to use this class in order to return results
// query code should not know about bolt protocol
template <class Socket>
class RecordStream
{
public:
RecordStream(Socket &socket) : socket(socket)
{
logger = logging::log->logger("Record Stream");
}
// TODO: create apstract methods that are not bolt specific ---------------
void write_success()
{
logger.trace("write_success");
bolt_encoder.message_success();
}
void write_success_empty()
{
logger.trace("write_success_empty");
bolt_encoder.message_success_empty();
}
void write_ignored()
{
logger.trace("write_ignored");
bolt_encoder.message_ignored();
}
void write_fields(const std::vector<std::string> &fields)
{
// TODO: that should be one level below?
bolt_encoder.message_success();
bolt_encoder.write_map_header(1);
bolt_encoder.write_string("fields");
write_list_header(fields.size());
for (auto &name : fields) {
bolt_encoder.write_string(name);
}
flush();
}
void write_field(const std::string& field)
{
bolt_encoder.message_success();
bolt_encoder.write_map_header(1);
bolt_encoder.write_string("fields");
write_list_header(1);
bolt_encoder.write_string(field);
flush();
}
void write_list_header(size_t size)
{
bolt_encoder.write_list_header(size);
}
void write_record()
{
bolt_encoder.message_record();
}
// -- BOLT SPECIFIC METHODS -----------------------------------------------
void write(const Vertex::Accessor &vertex) { serializer.write(vertex); }
void write(const Edge::Accessor &edge) { serializer.write(edge); }
void write(const Property &prop) { serializer.write(prop); }
void write(const Bool& prop) { serializer.write(prop); }
void write(const Float& prop) { serializer.write(prop); }
void write(const Int32& prop) { serializer.write(prop); }
void write(const Int64& prop) { serializer.write(prop); }
void write(const Double& prop) { serializer.write(prop); }
void write(const String& prop) { serializer.write(prop); }
void flush()
{
chunked_encoder.flush();
chunked_buffer.flush();
}
void _write_test()
{
logger.trace("write_test");
write_fields({{"name"}});
write_record();
write_list_header(1);
write(String("max"));
write_record();
write_list_header(1);
write(String("paul"));
write_success_empty();
}
protected:
Logger logger;
private:
using buffer_t = ChunkedBuffer<SocketStream>;
using chunked_encoder_t = ChunkedEncoder<buffer_t>;
using bolt_encoder_t = BoltEncoder<chunked_encoder_t>;
using bolt_serializer_t = BoltSerializer<bolt_encoder_t>;
SocketStream socket;
buffer_t chunked_buffer{socket};
chunked_encoder_t chunked_encoder{chunked_buffer};
bolt_encoder_t bolt_encoder{chunked_encoder};
bolt_serializer_t serializer{bolt_encoder};
};
}

View File

@@ -0,0 +1,67 @@
#pragma once
#include <vector>
#include <memory>
#include <thread>
#include <atomic>
#include <cassert>
#include "io/network/server.hpp"
#include "communication/bolt/v1/bolt.hpp"
namespace bolt
{
template <class Worker>
class Server : public io::Server<Server<Worker>>
{
public:
Server(io::Socket&& socket)
: io::Server<Server<Worker>>(std::forward<io::Socket>(socket)) {}
void start(size_t n)
{
workers.reserve(n);
for(size_t i = 0; i < n; ++i)
{
workers.push_back(std::make_shared<Worker>(bolt));
workers.back()->start(alive);
}
while(alive)
{
this->wait_and_process_events();
}
}
void shutdown()
{
alive.store(false);
for(auto& worker : workers)
worker->thread.join();
}
void on_connect()
{
assert(idx < workers.size());
if(UNLIKELY(!workers[idx]->accept(this->socket)))
return;
idx = idx == workers.size() - 1 ? 0 : idx + 1;
}
void on_wait_timeout() {}
private:
Bolt bolt;
std::vector<typename Worker::sptr> workers;
std::atomic<bool> alive {true};
int idx {0};
};
}

View File

@@ -0,0 +1,104 @@
#pragma once
#include <atomic>
#include <cstdio>
#include <iomanip>
#include <memory>
#include <sstream>
#include <thread>
#include "communication/bolt/v1/bolt.hpp"
#include "communication/bolt/v1/session.hpp"
#include "logging/default.hpp"
#include "io/network/stream_reader.hpp"
namespace bolt
{
template <class Worker>
class Server;
class Worker : public io::StreamReader<Worker, Session>
{
friend class bolt::Server<Worker>;
public:
using sptr = std::shared_ptr<Worker>;
Worker(Bolt &bolt) : bolt(bolt)
{
logger = logging::log->logger("Network");
}
Session &on_connect(io::Socket &&socket)
{
logger.trace("Accepting connection on socket {}", socket.id());
return *bolt.get().create_session(std::forward<io::Socket>(socket));
}
void on_error(Session &)
{
logger.trace("[on_error] errno = {}", errno);
#ifndef NDEBUG
auto err = io::NetworkError("");
logger.debug("{}", err.what());
#endif
logger.error("Error occured in this session");
}
void on_wait_timeout() {}
Buffer on_alloc(Session &)
{
/* logger.trace("[on_alloc] Allocating {}B", sizeof buf); */
return Buffer{buf, sizeof buf};
}
void on_read(Session &session, Buffer &buf)
{
logger.trace("[on_read] Received {}B", buf.len);
#ifndef NDEBUG
std::stringstream stream;
for (size_t i = 0; i < buf.len; ++i)
stream << fmt::format("{:02X} ", static_cast<byte>(buf.ptr[i]));
logger.trace("[on_read] {}", stream.str());
#endif
try {
session.execute(reinterpret_cast<const byte *>(buf.ptr), buf.len);
} catch (const std::exception &e) {
logger.error("Error occured while executing statement.");
logger.error("{}", e.what());
}
}
void on_close(Session &session)
{
logger.trace("[on_close] Client closed the connection");
session.close();
}
char buf[65536];
protected:
std::reference_wrapper<Bolt> bolt;
Logger logger;
std::thread thread;
void start(std::atomic<bool> &alive)
{
thread = std::thread([&, this]() {
while (alive)
wait_and_process_events();
});
}
};
}

View File

@@ -0,0 +1,43 @@
#pragma once
#include "io/network/socket.hpp"
#include "io/network/tcp/stream.hpp"
#include "communication/bolt/v1/bolt.hpp"
#include "communication/bolt/v1/serialization/record_stream.hpp"
#include "communication/bolt/v1/states/state.hpp"
#include "communication/bolt/v1/transport/bolt_decoder.hpp"
#include "communication/bolt/v1/transport/bolt_encoder.hpp"
#include "communication/communication.hpp"
#include "logging/default.hpp"
namespace bolt
{
class Session : public io::tcp::Stream<io::Socket>
{
public:
using Decoder = BoltDecoder;
using OutputStream = communication::OutputStream;
Session(io::Socket &&socket, Bolt &bolt);
bool alive() const;
void execute(const byte *data, size_t len);
void close();
Bolt &bolt;
Db &active_db();
Decoder decoder;
OutputStream output_stream{socket};
bool connected{false};
State *state;
protected:
Logger logger;
};
}

View File

@@ -0,0 +1,20 @@
#pragma once
#include "communication/bolt/v1/states/state.hpp"
#include "logging/log.hpp"
namespace bolt
{
class States
{
public:
States();
State::uptr handshake;
State::uptr init;
State::uptr executor;
State::uptr error;
};
}

View File

@@ -0,0 +1,15 @@
#pragma once
#include "communication/bolt/v1/session.hpp"
#include "communication/bolt/v1/states/state.hpp"
namespace bolt
{
class Error : public State
{
public:
State *run(Session &session) override;
};
}

View File

@@ -0,0 +1,46 @@
#pragma once
#include "communication/bolt/v1/states/state.hpp"
#include "communication/bolt/v1/session.hpp"
#include "query_engine/query_engine.hpp"
namespace bolt
{
class Executor : public State
{
struct Query
{
std::string statement;
};
public:
Executor();
State* run(Session& session) override final;
protected:
Logger logger;
/* Execute an incoming query
*
*/
void run(Session& session, Query& query);
/* Send all remaining results to the client
*
*/
void pull_all(Session& session);
/* Discard all remaining results
*
*/
void discard_all(Session& session);
private:
QueryEngine query_engine;
};
}

View File

@@ -0,0 +1,14 @@
#pragma once
#include "communication/bolt/v1/states/state.hpp"
namespace bolt
{
class Handshake : public State
{
public:
State* run(Session& session) override;
};
}

View File

@@ -0,0 +1,22 @@
#pragma once
#include "communication/bolt/v1/states/message_parser.hpp"
namespace bolt
{
class Init : public MessageParser<Init>
{
public:
struct Message
{
std::string client_name;
};
Init();
State* parse(Session& session, Message& message);
State* execute(Session& session, Message& message);
};
}

View File

@@ -0,0 +1,33 @@
#pragma once
#include "communication/bolt/v1/session.hpp"
#include "communication/bolt/v1/states/state.hpp"
#include "utils/crtp.hpp"
namespace bolt
{
template <class Derived>
class MessageParser : public State, public Crtp<Derived>
{
public:
MessageParser(Logger &&logger) : logger(std::forward<Logger>(logger)) {}
State *run(Session &session) override final
{
typename Derived::Message message;
logger.debug("Parsing message");
auto next = this->derived().parse(session, message);
// return next state if parsing was unsuccessful (i.e. error state)
if (next != &this->derived()) return next;
logger.debug("Executing state");
return this->derived().execute(session, message);
}
protected:
Logger logger;
};
}

View File

@@ -0,0 +1,23 @@
#pragma once
#include <cstdlib>
#include <cstdint>
#include <memory>
namespace bolt
{
class Session;
class State
{
public:
using uptr = std::unique_ptr<State>;
State() = default;
virtual ~State() = default;
virtual State* run(Session& session) = 0;
};
}

View File

@@ -0,0 +1,43 @@
#pragma once
#include "communication/bolt/v1/transport/buffer.hpp"
#include "communication/bolt/v1/transport/chunked_decoder.hpp"
#include "utils/types/byte.hpp"
namespace bolt
{
class BoltDecoder
{
public:
void handshake(const byte *&data, size_t len);
bool decode(const byte *&data, size_t len);
bool empty() const;
void reset();
byte peek() const;
byte read_byte();
void read_bytes(void *dest, size_t n);
int16_t read_int16();
uint16_t read_uint16();
int32_t read_int32();
uint32_t read_uint32();
int64_t read_int64();
uint64_t read_uint64();
double read_float64();
std::string read_string();
private:
Buffer buffer;
ChunkedDecoder<Buffer> decoder{buffer};
size_t pos{0};
const byte *raw() const;
};
}

View File

@@ -0,0 +1,274 @@
#pragma once
#include <string>
#include "communication/bolt/v1/packing/codes.hpp"
#include "communication/bolt/v1/messaging/codes.hpp"
#include "utils/types/byte.hpp"
#include "utils/bswap.hpp"
#include "logging/default.hpp"
namespace bolt
{
template <class Stream>
class BoltEncoder
{
static constexpr int64_t plus_2_to_the_31 = 2147483648L;
static constexpr int64_t plus_2_to_the_15 = 32768L;
static constexpr int64_t plus_2_to_the_7 = 128L;
static constexpr int64_t minus_2_to_the_4 = -16L;
static constexpr int64_t minus_2_to_the_7 = -128L;
static constexpr int64_t minus_2_to_the_15 = -32768L;
static constexpr int64_t minus_2_to_the_31 = -2147483648L;
public:
BoltEncoder(Stream& stream) : stream(stream)
{
logger = logging::log->logger("Bolt Encoder");
}
void flush()
{
stream.flush();
}
void write(byte value)
{
write_byte(value);
}
void write_byte(byte value)
{
logger.trace("write byte: {}", value);
stream.write(value);
}
void write(const byte* values, size_t n)
{
stream.write(values, n);
}
void write_null()
{
stream.write(pack::Null);
}
void write(bool value)
{
write_bool(value);
}
void write_bool(bool value)
{
if(value) write_true(); else write_false();
}
void write_true()
{
stream.write(pack::True);
}
void write_false()
{
stream.write(pack::False);
}
template <class T>
void write_value(T value)
{
value = bswap(value);
stream.write(reinterpret_cast<const byte*>(&value), sizeof(value));
}
void write_integer(int64_t value)
{
if(value >= minus_2_to_the_4 && value < plus_2_to_the_7)
{
write(static_cast<byte>(value));
}
else if(value >= minus_2_to_the_7 && value < minus_2_to_the_4)
{
write(pack::Int8);
write(static_cast<byte>(value));
}
else if(value >= minus_2_to_the_15 && value < plus_2_to_the_15)
{
write(pack::Int16);
write_value(static_cast<int16_t>(value));
}
else if(value >= minus_2_to_the_31 && value < plus_2_to_the_31)
{
write(pack::Int32);
write_value(static_cast<int32_t>(value));
}
else
{
write(pack::Int64);
write_value(value);
}
}
void write(double value)
{
write_double(value);
}
void write_double(double value)
{
write(pack::Float64);
write_value(*reinterpret_cast<const int64_t*>(&value));
}
void write_map_header(size_t size)
{
if(size < 0x10)
{
write(static_cast<byte>(pack::TinyMap | size));
}
else if(size <= 0xFF)
{
write(pack::Map8);
write(static_cast<byte>(size));
}
else if(size <= 0xFFFF)
{
write(pack::Map16);
write_value<uint16_t>(size);
}
else
{
write(pack::Map32);
write_value<uint32_t>(size);
}
}
void write_empty_map()
{
write(pack::TinyMap);
}
void write_list_header(size_t size)
{
if(size < 0x10)
{
write(static_cast<byte>(pack::TinyList | size));
}
else if(size <= 0xFF)
{
write(pack::List8);
write(static_cast<byte>(size));
}
else if(size <= 0xFFFF)
{
write(pack::List16);
write_value<uint16_t>(size);
}
else
{
write(pack::List32);
write_value<uint32_t>(size);
}
}
void write_empty_list()
{
write(pack::TinyList);
}
void write_string_header(size_t size)
{
if(size < 0x10)
{
write(static_cast<byte>(pack::TinyString | size));
}
else if(size <= 0xFF)
{
write(pack::String8);
write(static_cast<byte>(size));
}
else if(size <= 0xFFFF)
{
write(pack::String16);
write_value<uint16_t>(size);
}
else
{
write(pack::String32);
write_value<uint32_t>(size);
}
}
void write_string(const std::string& str)
{
write_string(str.c_str(), str.size());
}
void write_string(const char* str, size_t len)
{
write_string_header(len);
write(reinterpret_cast<const byte*>(str), len);
}
void write_struct_header(size_t size)
{
if(size < 0x10)
{
write(static_cast<byte>(pack::TinyStruct | size));
}
else if(size <= 0xFF)
{
write(pack::Struct8);
write(static_cast<byte>(size));
}
else
{
write(pack::Struct16);
write_value<uint16_t>(size);
}
}
void message_success()
{
write_struct_header(1);
write(underlying_cast(MessageCode::Success));
}
void message_success_empty()
{
message_success();
write_empty_map();
}
void message_record()
{
write_struct_header(1);
write(underlying_cast(MessageCode::Record));
}
void message_record_empty()
{
message_record();
write_empty_list();
}
void message_ignored()
{
write_struct_header(1);
write(underlying_cast(MessageCode::Ignored));
}
void message_ignored_empty()
{
message_ignored();
write_empty_map();
}
protected:
Logger logger;
private:
Stream& stream;
};
}

View File

@@ -0,0 +1,39 @@
#pragma once
#include <cstdint>
#include <cstdlib>
#include <vector>
#include "utils/types/byte.hpp"
namespace bolt
{
class Buffer
{
public:
void write(const byte* data, size_t len);
void clear();
size_t size() const
{
return buffer.size();
}
byte operator[](size_t idx) const
{
return buffer[idx];
}
const byte* data() const
{
return buffer.data();
}
private:
std::vector<byte> buffer;
};
}

View File

@@ -0,0 +1,74 @@
#pragma once
#include <memory>
#include <vector>
#include <cstring>
#include "communication/bolt/v1/config.hpp"
#include "utils/types/byte.hpp"
#include "logging/default.hpp"
namespace bolt
{
template <class Stream>
class ChunkedBuffer
{
static constexpr size_t C = bolt::config::C; /* chunk size */
public:
ChunkedBuffer(Stream &stream) : stream(stream)
{
logger = logging::log->logger("Chunked Buffer");
}
void write(const byte *values, size_t n)
{
// TODO: think about shared pointer
// TODO: this is naive implementation, it can be implemented much better
logger.trace("write {} bytes", n);
byte *chunk = chunk = (byte *)std::malloc(n * sizeof(byte));
last_size = n;
std::memcpy(chunk, values, n);
buffer.push_back(chunk);
}
void flush()
{
logger.trace("Flush");
for (size_t i = 0; i < buffer.size(); ++i) {
if (i == buffer.size() - 1)
stream.get().write(buffer[i], last_size);
else
stream.get().write(buffer[i], C);
}
destroy();
}
~ChunkedBuffer()
{
destroy();
}
private:
Logger logger;
std::reference_wrapper<Stream> stream;
std::vector<byte *> buffer;
size_t last_size {0}; // last chunk size (it is going to be less than C)
void destroy()
{
for (size_t i = 0; i < buffer.size(); ++i) {
std::free(buffer[i]);
}
buffer.clear();
}
};
}

View File

@@ -0,0 +1,70 @@
#pragma once
#include <cassert>
#include <cstring>
#include <functional>
#include "logging/default.hpp"
#include "utils/exceptions/basic_exception.hpp"
#include "utils/likely.hpp"
#include "utils/types/byte.hpp"
namespace bolt
{
template <class Stream>
class ChunkedDecoder
{
public:
class DecoderError : public BasicException
{
public:
using BasicException::BasicException;
};
ChunkedDecoder(Stream& stream) : stream(stream) {}
/* Decode chunked data
*
* Chunk format looks like:
*
* |Header| Data ||Header| Data || ... || End |
* | 2B | size bytes || 2B | size bytes || ... ||00 00|
*/
bool decode(const byte *&chunk, size_t n)
{
while (n > 0)
{
// get size from first two bytes in the chunk
auto size = get_size(chunk);
if (UNLIKELY(size + 2 > n))
throw DecoderError("Chunk size larger than available data.");
// advance chunk to pass those two bytes
chunk += 2;
n -= 2;
// if chunk size is 0, we're done!
if (size == 0) return true;
stream.get().write(chunk, size);
chunk += size;
n -= size;
}
return false;
}
bool operator()(const byte *&chunk, size_t n) { return decode(chunk, n); }
private:
std::reference_wrapper<Stream> stream;
size_t get_size(const byte *chunk)
{
return size_t(chunk[0]) << 8 | chunk[1];
}
};
}

View File

@@ -0,0 +1,102 @@
#pragma once
#include <array>
#include <cstring>
#include <functional>
#include "utils/likely.hpp"
#include "communication/bolt/v1/config.hpp"
#include "logging/default.hpp"
namespace bolt
{
template <class Stream>
class ChunkedEncoder
{
static constexpr size_t N = bolt::config::N;
static constexpr size_t C = bolt::config::C;
public:
using byte = unsigned char;
ChunkedEncoder(Stream& stream) : stream(stream)
{
logger = logging::log->logger("Chunked Encoder");
}
static constexpr size_t chunk_size = N - 2;
void write(byte value)
{
if(UNLIKELY(pos == N))
end_chunk();
chunk[pos++] = value;
}
void write(const byte* values, size_t n)
{
logger.trace("write {} bytes", n);
while(n > 0)
{
auto size = n < N - pos ? n : N - pos;
std::memcpy(chunk.data() + pos, values, size);
pos += size;
n -= size;
if(pos == N)
end_chunk();
}
}
void flush()
{
write_chunk_header();
// write two zeros to signal message end
chunk[pos++] = 0x00;
chunk[pos++] = 0x00;
flush_stream();
}
private:
Logger logger;
std::reference_wrapper<Stream> stream;
std::array<byte, C> chunk;
size_t pos {2};
void end_chunk()
{
// TODO: this call is unnecessary bacause the same method is called
// inside the flush method
// write_chunk_header();
flush();
}
void write_chunk_header()
{
// write the size of the chunk
uint16_t size = pos - 2;
// write the higher byte
chunk[0] = size >> 8;
// write the lower byte
chunk[1] = size & 0xFF;
}
void flush_stream()
{
// write chunk to the stream
stream.get().write(chunk.data(), pos);
pos = 2;
}
};
}

View File

@@ -0,0 +1,38 @@
#pragma once
#include <cstdint>
#include <vector>
#include <cstdio>
#include "io/network/socket.hpp"
#include "communication/bolt/v1/transport/stream_error.hpp"
namespace bolt
{
class SocketStream
{
public:
using byte = uint8_t;
SocketStream(io::Socket& socket) : socket(socket) {}
void write(const byte* data, size_t n)
{
while(n > 0)
{
auto written = socket.get().write(data, n);
if(UNLIKELY(written == -1))
throw StreamError("Can't write to stream");
n -= written;
data += written;
}
}
private:
std::reference_wrapper<io::Socket> socket;
};
}

View File

@@ -0,0 +1,14 @@
#pragma once
#include "utils/exceptions/basic_exception.hpp"
namespace bolt
{
class StreamError : BasicException
{
public:
using BasicException::BasicException;
};
}

View File

@@ -0,0 +1,9 @@
#pragma once
#include "io/network/socket.hpp"
#include "communication/bolt/v1/serialization/record_stream.hpp"
namespace communication
{
using OutputStream = bolt::RecordStream<io::Socket>;
}

View File

@@ -0,0 +1,4 @@
#pragma once
/* Memgraph Communication protocol
* gate is the first name proposal for the protocol */

View File

@@ -0,0 +1,3 @@
#pragma once
/* HTTP & HTTPS implementation */

View File

@@ -0,0 +1,153 @@
#pragma once
#include <cassert>
#include <atomic>
#include "threading/sync/lockable.hpp"
#include "threading/sync/spinlock.hpp"
template <class block_t = uint8_t, size_t chunk_size = 32768>
class DynamicBitset : Lockable<SpinLock>
{
struct Block
{
Block() = default;
Block(Block&) = delete;
Block(Block&&) = delete;
static constexpr size_t size = sizeof(block_t) * 8;
constexpr block_t bitmask(size_t group_size) const
{
return (block_t)(-1) >> (size - group_size);
}
block_t at(size_t k, size_t n, std::memory_order order)
{
assert(k + n - 1 < size);
return (block.load(order) >> k) & bitmask(n);
}
void set(size_t k, size_t n, std::memory_order order)
{
assert(k + n - 1 < size);
block.fetch_or(bitmask(n) << k, order);
}
void clear(size_t k, size_t n, std::memory_order order)
{
assert(k + n - 1 < size);
block.fetch_and(~(bitmask(n) << k), order);
}
std::atomic<block_t> block {0};
};
struct Chunk
{
Chunk() : next(nullptr)
{
static_assert(chunk_size % sizeof(block_t) == 0,
"chunk size not divisible by block size");
}
Chunk(Chunk&) = delete;
Chunk(Chunk&&) = delete;
~Chunk()
{
delete next;
}
static constexpr size_t size = chunk_size * Block::size;
static constexpr size_t n_blocks = chunk_size / sizeof(block_t);
block_t at(size_t k, size_t n, std::memory_order order)
{
return blocks[k / Block::size].at(k % Block::size, n, order);
}
void set(size_t k, size_t n, std::memory_order order)
{
blocks[k / Block::size].set(k % Block::size, n, order);
}
void clear(size_t k, size_t n, std::memory_order order)
{
blocks[k / Block::size].clear(k % Block::size, n, order);
}
Block blocks[n_blocks];
std::atomic<Chunk*> next;
};
public:
DynamicBitset() : head(new Chunk()) {}
DynamicBitset(DynamicBitset&) = delete;
DynamicBitset(DynamicBitset&&) = delete;
block_t at(size_t k, size_t n)
{
auto& chunk = find_chunk(k);
return chunk.at(k, n, std::memory_order_seq_cst);
}
bool at(size_t k)
{
auto& chunk = find_chunk(k);
return chunk.at(k, 1, std::memory_order_seq_cst);
}
void set(size_t k, size_t n = 1)
{
auto& chunk = find_chunk(k);
return chunk.set(k, n, std::memory_order_seq_cst);
}
void clear(size_t k, size_t n = 1)
{
auto& chunk = find_chunk(k);
return chunk.clear(k, n, std::memory_order_seq_cst);
}
private:
Chunk& find_chunk(size_t& k)
{
Chunk* chunk = head.load(), *next = nullptr;
// while i'm not in the right chunk
// (my index is bigger than the size of this chunk)
while(k >= Chunk::size)
{
next = chunk->next.load();
// if a next chunk exists, switch to it and decrement my
// pointer by the size of the current chunk
if(next != nullptr)
{
chunk = next;
k -= Chunk::size;
continue;
}
// the next chunk does not exist and we need it. take an exclusive
// lock to prevent others that also want to create a new chunk
// from creating it
auto guard = acquire_unique();
// double-check locking. if the chunk exists now, some other thread
// has just created it, continue searching for my chunk
if(chunk->next.load() != nullptr)
continue;
chunk->next.store(new Chunk());
}
assert(chunk != nullptr);
return *chunk;
}
std::atomic<Chunk*> head;
};

View File

@@ -0,0 +1,93 @@
#pragma once
#include "data_structures/concurrent/skiplist.hpp"
#include "utils/total_ordering.hpp"
using std::pair;
template <typename K, typename T>
class Item : public TotalOrdering<Item<K, T>>,
public TotalOrdering<K, Item<K, T>>,
public TotalOrdering<Item<K, T>, K>,
public pair<const K, T>
{
public:
using pair<const K, T>::pair;
friend constexpr bool operator<(const Item &lhs, const Item &rhs)
{
return lhs.first < rhs.first;
}
friend constexpr bool operator==(const Item &lhs, const Item &rhs)
{
return lhs.first == rhs.first;
}
friend constexpr bool operator<(const K &lhs, const Item &rhs)
{
return lhs < rhs.first;
}
friend constexpr bool operator==(const K &lhs, const Item &rhs)
{
return lhs == rhs.first;
}
friend constexpr bool operator<(const Item &lhs, const K &rhs)
{
return lhs.first < rhs;
}
friend constexpr bool operator==(const Item &lhs, const K &rhs)
{
return lhs.first == rhs;
}
};
template <typename T>
class AccessorBase
{
typedef SkipList<T> list;
typedef typename SkipList<T>::Iterator list_it;
typedef typename SkipList<T>::ConstIterator list_it_con;
protected:
AccessorBase(list *skiplist) : accessor(skiplist->access()) {}
public:
AccessorBase(const AccessorBase &) = delete;
AccessorBase(AccessorBase &&other) : accessor(std::move(other.accessor)) {}
~AccessorBase() {}
list_it begin() { return accessor.begin(); }
list_it_con begin() const { return accessor.cbegin(); }
list_it_con cbegin() const { return accessor.cbegin(); }
list_it end() { return accessor.end(); }
list_it_con end() const { return accessor.cend(); }
list_it_con cend() const { return accessor.cend(); }
template <class K>
typename SkipList<T>::template MultiIterator<K> end(const K &data)
{
return accessor.template mend<K>(data);
}
template <class K>
typename SkipList<T>::template MultiIterator<K> mend(const K &data)
{
return accessor.template mend<K>(data);
}
size_t size() const { return accessor.size(); }
protected:
typename list::Accessor accessor;
};

View File

@@ -0,0 +1,60 @@
#pragma once
#include "data_structures/concurrent/common.hpp"
#include "data_structures/concurrent/skiplist.hpp"
using std::pair;
template <typename K, typename T>
class ConcurrentMap
{
typedef Item<K, T> item_t;
typedef SkipList<item_t> list;
typedef typename SkipList<item_t>::Iterator list_it;
typedef typename SkipList<item_t>::ConstIterator list_it_con;
public:
ConcurrentMap() {}
class Accessor : public AccessorBase<item_t>
{
friend class ConcurrentMap;
using AccessorBase<item_t>::AccessorBase;
private:
using AccessorBase<item_t>::accessor;
public:
std::pair<list_it, bool> insert(const K &key, const T &data)
{
return accessor.insert(item_t(key, data));
}
std::pair<list_it, bool> insert(const K &key, T &&data)
{
return accessor.insert(item_t(key, std::forward<T>(data)));
}
std::pair<list_it, bool> insert(K &&key, T &&data)
{
return accessor.insert(
item_t(std::forward<K>(key), std::forward<T>(data)));
}
list_it_con find(const K &key) const { return accessor.find(key); }
list_it find(const K &key) { return accessor.find(key); }
bool contains(const K &key) const { return this->find(key) != this->end(); }
bool remove(const K &key) { return accessor.remove(key); }
};
Accessor access() { return Accessor(&skiplist); }
const Accessor access() const { return Accessor(&skiplist); }
private:
list skiplist;
};

View File

@@ -0,0 +1,70 @@
#pragma once
#include "data_structures/concurrent/skiplist.hpp"
#include "utils/total_ordering.hpp"
using std::pair;
template <typename K, typename T>
class ConcurrentMultiMap
{
typedef Item<K, T> item_t;
typedef SkipList<item_t> list;
typedef typename SkipList<item_t>::Iterator list_it;
typedef typename SkipList<item_t>::ConstIterator list_it_con;
typedef typename SkipList<item_t>::template MultiIterator<K> list_it_multi;
public:
ConcurrentMultiMap() {}
class Accessor : public AccessorBase<item_t>
{
friend class ConcurrentMultiMap<K, T>;
using AccessorBase<item_t>::AccessorBase;
private:
using AccessorBase<item_t>::accessor;
public:
list_it insert(const K &key, const T &data)
{
return accessor.insert_non_unique(item_t(key, data));
}
list_it insert(const K &key, T &&data)
{
return accessor.insert_non_unique(
item_t(key, std::forward<T>(data)));
}
list_it insert(K &&key, T &&data)
{
return accessor.insert_non_unique(
item_t(std::forward<K>(key), std::forward<T>(data)));
}
list_it_multi find_multi(const K &key)
{
return accessor.find_multi(key);
}
list_it_con find(const K &key) const { return accessor.find(key); }
list_it find(const K &key) { return accessor.find(key); }
bool contains(const K &key) const
{
return this->find(key) != this->end();
}
bool remove(const K &key) { return accessor.remove(key); }
};
Accessor access() { return Accessor(&skiplist); }
const Accessor access() const { return Accessor(&skiplist); }
private:
list skiplist;
};

View File

@@ -0,0 +1,50 @@
#pragma once
#include "data_structures/concurrent/skiplist.hpp"
template <class T>
class ConcurrentMultiSet
{
typedef SkipList<T> list;
typedef typename SkipList<T>::Iterator list_it;
typedef typename SkipList<T>::ConstIterator list_it_con;
public:
ConcurrentMultiSet() {}
class Accessor : public AccessorBase<T>
{
friend class ConcurrentMultiSet;
using AccessorBase<T>::AccessorBase;
private:
using AccessorBase<T>::accessor;
public:
list_it insert(const T &item) { return accessor.insert_non_unique(item); }
list_it insert(T &&item)
{
return accessor.insert_non_unique(std::forward<T>(item));
}
list_it_con find(const T &item) const { return accessor.find(item); }
list_it find(const T &item) { return accessor.find(item); }
bool contains(const T &item) const
{
return this->find(item) != this->end();
}
bool remove(const T &item) { return accessor.remove(item); }
};
Accessor access() { return Accessor(&skiplist); }
const Accessor access() const { return Accessor(&skiplist); }
private:
list skiplist;
};

View File

@@ -0,0 +1,54 @@
#pragma once
#include "data_structures/concurrent/common.hpp"
#include "data_structures/concurrent/skiplist.hpp"
template <class T>
class ConcurrentSet
{
typedef SkipList<T> list;
typedef typename SkipList<T>::Iterator list_it;
typedef typename SkipList<T>::ConstIterator list_it_con;
public:
ConcurrentSet() {}
class Accessor : public AccessorBase<T>
{
friend class ConcurrentSet;
using AccessorBase<T>::AccessorBase;
private:
using AccessorBase<T>::accessor;
public:
std::pair<list_it, bool> insert(const T &item)
{
return accessor.insert(item);
}
std::pair<list_it, bool> insert(T &&item)
{
return accessor.insert(std::forward<T>(item));
}
list_it_con find(const T &item) const { return accessor.find(item); }
list_it find(const T &item) { return accessor.find(item); }
bool contains(const T &item) const
{
return this->find(item) != this->end();
}
bool remove(const T &item) { return accessor.remove(item); }
};
Accessor access() { return Accessor(&skiplist); }
const Accessor access() const { return Accessor(&skiplist); }
private:
list skiplist;
};

View File

@@ -0,0 +1,881 @@
#pragma once
#include <algorithm>
#include <cassert>
#include <memory>
#include "utils/placeholder.hpp"
#include "utils/random/fast_binomial.hpp"
#include "threading/sync/lockable.hpp"
#include "threading/sync/spinlock.hpp"
#include "data_structures/concurrent/skiplist_gc.hpp"
/* @brief Concurrent lock-based skiplist with fine grained locking
*
* From Wikipedia:
* "A skip list is a data structure that allows fast search within an
* ordered sequence of elements. Fast search is made possible by
* maintaining a linked hierarchy of subsequences, each skipping over
* fewer elements. Searching starts in the sparsest subsequence until
* two consecutive elements have been found, one smaller and one
* larger than or equal to the element searched for."
*
* [_]---------------->[+]----------->[_]
* [_]->[+]----------->[+]------>[+]->[_]
* [_]->[+]------>[+]->[+]------>[+]->[_]
* [_]->[+]->[+]->[+]->[+]->[+]->[+]->[_]
* head 1 2 4 5 8 9 nil
*
* The logarithmic properties are maintained by randomizing the height for
* every new node using the binomial distribution
* p(k) = (1/2)^k for k in [1...H].
*
* The implementation is based on the work described in the paper
* "A Provably Correct Scalable Concurrent Skip List"
* URL: https://www.cs.tau.ac.il/~shanir/nir-pubs-web/Papers/OPODIS2006-BA.pdf
*
* The proposed implementation is in Java so the authors don't worry about
* garbage collection, but obviously we have to. This implementation uses
* lazy garbage collection. When all clients stop using the skiplist, we can
* be sure that all logically removed nodes are not visible to anyone so
* we can safely remove them. The idea of counting active clients implies
* the use of a intermediary structure (called Accessor) when accessing the
* skiplist.
*
* The implementation has an interface which closely resembles the functions
* with arguments and returned types frequently used by the STL.
*
* Example usage:
* Skiplist<T> skiplist;
*
* {
* auto accessor = skiplist.access();
*
* // inserts item into the skiplist and returns
* // <iterator, bool> pair. iterator points to the newly created
* // node and the boolean member evaluates to true denoting that the
* // insertion was successful
* accessor.insert(item1);
*
* // nothing gets inserted because item1 already exist in the skiplist
* // returned iterator points to the existing element and the return
* // boolean evaluates to false denoting the failed insertion
* accessor.insert(item1);
*
* // returns an iterator to the element item1
* auto it = accessor.find(item1);
*
* // returns an empty iterator. it == accessor.end()
* auto it = accessor.find(item2);
*
* // iterate over all items
* for(auto it = accessor.begin(); it != accessor.end(); ++it)
* cout << *it << endl;
*
* // range based for loops also work
* for(auto& e : accessor)
* cout << e << endl;
*
* accessor.remove(item1); // returns true
* accessor.remove(item1); // returns false because key1 doesn't exist
* }
*
* // accessor out of scope, garbage collection might occur
*
* For detailed operations available, please refer to the Accessor class
* inside the public section of the SkipList class.
*
* @tparam T Type to use as the item
* @tparam H Maximum node height. Determines the effective number of nodes
* the skiplist can hold in order to preserve it's log2 properties
* @tparam lock_t Lock type used when locking is needed during the creation
* and deletion of nodes.
*/
template <class T, size_t H = 32, class lock_t = SpinLock>
class SkipList : private Lockable<lock_t>
{
public:
// computes the height for the new node from the interval [1...H]
// with p(k) = (1/2)^k for all k from the interval
static thread_local FastBinomial<H> rnd;
/* @brief Wrapper class for flags used in the implementation
*
* MARKED flag is used to logically delete a node.
* FULLY_LINKED is used to mark the node as fully inserted, i.e. linked
* at all layers in the skiplist up to the node height
*/
struct Flags
{
enum node_flags : uint8_t
{
MARKED = 0x01,
FULLY_LINKED = 0x10,
};
bool is_marked() const { return flags.load() & MARKED; }
void set_marked() { flags.fetch_or(MARKED); }
bool is_fully_linked() const { return flags.load() & FULLY_LINKED; }
void set_fully_linked() { flags.fetch_or(FULLY_LINKED); }
private:
std::atomic<uint8_t> flags{0};
};
class Node : Lockable<lock_t>
{
public:
friend class SkipList;
const uint8_t height;
Flags flags;
T &value() { return data.get(); }
const T &value() const { return data.get(); }
static Node *sentinel(uint8_t height)
{
// we have raw memory and we need to construct an object
// of type Node on it
return new (allocate(height)) Node(height);
}
static Node *create(const T &item, uint8_t height)
{
return create(item, height);
}
static Node *create(T &&item, uint8_t height)
{
auto node = allocate(height);
// we have raw memory and we need to construct an object
// of type Node on it
return new (node) Node(std::forward<T>(item), height);
}
static void destroy(Node *node)
{
node->~Node();
std::free(node);
}
Node *forward(size_t level) const { return tower[level].load(); }
void forward(size_t level, Node *next) { tower[level].store(next); }
private:
Node(uint8_t height) : height(height)
{
// here we assume, that the memory for N towers (N = height) has
// been allocated right after the Node structure so we need to
// initialize that memory
for (auto i = 0; i < height; ++i)
new (&tower[i]) std::atomic<Node *>{nullptr};
}
Node(T &&data, uint8_t height) : Node(height)
{
this->data.set(std::forward<T>(data));
}
~Node()
{
for (auto i = 0; i < height; ++i)
tower[i].~atomic();
}
static Node *allocate(uint8_t height)
{
// [ Node ][Node*][Node*][Node*]...[Node*]
// | | | | |
// | 0 1 2 height-1
// |----------------||-----------------------------|
// space for Node space for tower pointers
// structure right after the Node
// structure
auto size = sizeof(Node) + height * sizeof(std::atomic<Node *>);
auto node = static_cast<Node *>(std::malloc(size));
return node;
}
Placeholder<T> data;
// this creates an array of the size zero. we can't put any sensible
// value here since we don't know what size it will be untill the
// node is allocated. we could make it a Node** but then we would
// have two memory allocations, one for node and one for the forward
// list. this way we avoid expensive malloc/free calls and also cache
// thrashing when following a pointer on the heap
std::atomic<Node *> tower[0];
};
public:
template <class It>
class IteratorBase : public Crtp<It>
{
protected:
IteratorBase(Node *node) : node(node) {}
Node *node{nullptr};
public:
IteratorBase() = default;
IteratorBase(const IteratorBase &) = default;
T &operator*()
{
assert(node != nullptr);
return node->value();
}
T *operator->()
{
assert(node != nullptr);
return &node->value();
}
operator T &()
{
assert(node != nullptr);
return node->value();
}
It &operator++()
{
assert(node != nullptr);
node = node->forward(0);
return this->derived();
}
bool has_next()
{
assert(node != nullptr);
return node->forward(0) != nullptr;
}
It &operator++(int) { return operator++(); }
friend bool operator==(const It &a, const It &b)
{
return a.node == b.node;
}
friend bool operator!=(const It &a, const It &b) { return !(a == b); }
};
class ConstIterator : public IteratorBase<ConstIterator>
{
friend class SkipList;
ConstIterator(Node *node) : IteratorBase<ConstIterator>(node) {}
public:
ConstIterator() = default;
ConstIterator(const ConstIterator &) = default;
const T &operator*()
{
return IteratorBase<ConstIterator>::operator*();
}
const T *operator->()
{
return IteratorBase<ConstIterator>::operator->();
}
operator const T &()
{
return IteratorBase<ConstIterator>::operator T &();
}
};
class Iterator : public IteratorBase<Iterator>
{
friend class SkipList;
Iterator(Node *node) : IteratorBase<Iterator>(node) {}
public:
Iterator() = default;
Iterator(const Iterator &) = default;
};
template <class K>
class MultiIterator : public Crtp<MultiIterator<K>>
{
friend class SkipList;
MultiIterator(const K &data) : data(data), skiplist(nullptr)
{
succs[0] = nullptr;
};
MultiIterator(SkipList *skiplist, const K &data)
: data(data), skiplist(skiplist)
{
while (true) {
auto level = find_path(skiplist, H - 1, data, preds, succs);
if (level == -1) {
succs[0] = nullptr;
} else if (succs[0] != succs[succs[0]->height - 1] ||
!succs[level]->flags.is_fully_linked()) {
usleep(250);
continue;
}
break;
}
}
public:
MultiIterator(const MultiIterator &) = default;
T &operator*()
{
assert(succs[0] != nullptr);
return succs[0]->value();
}
T *operator->()
{
assert(succs[0] != nullptr);
return &succs[0]->value();
}
operator T &()
{
assert(succs[0] != nullptr);
return succs[0]->value();
}
bool has_next()
{
assert(succs[0] != nullptr);
return succs[0].forward(0) != nullptr;
}
bool has_value() { return succs[0] != nullptr; }
MultiIterator &operator++()
{
assert(succs[0] != nullptr);
// This whole method can be optimized if it's valid to expect height
// of 1 on same key elements.
for (int i = succs[0]->height - 1; i >= 0; i--) {
preds[i] = succs[i];
succs[i] = preds[i]->forward(i);
}
if (succs[0] != nullptr) {
if (succs[0]->value() != data) {
succs[0] = nullptr;
} else {
while (succs[0] != succs[succs[0]->height - 1] ||
!succs[0]->flags.is_fully_linked()) {
usleep(250);
for (int i = succs[0]->height - 1; i >= 0; i--) {
succs[i] = preds[i]->forward(i);
}
}
}
}
return this->derived();
}
MultiIterator &operator++(int) { return operator++(); }
friend bool operator==(const MultiIterator &a, const MultiIterator &b)
{
return a.succs[0] == b.succs[0];
}
friend bool operator!=(const MultiIterator &a, const MultiIterator &b)
{
return !(a == b);
}
bool is_removed()
{
assert(succs[0] != nullptr);
return succs[0]->flags.is_marked();
}
// True if this call successfuly removed value. ITERATOR IS'T ADVANCED.
// False may mean that data has already been removed.
bool remove()
{
assert(succs[0] != nullptr);
return skiplist->template remove<K>(
data, preds, succs,
SkipList<T>::template MultiIterator<K>::update_path);
}
private:
static int update_path(SkipList *skiplist, int start, const K &item,
Node *preds[], Node *succs[])
{
// One optimization here would be to wait for is_fully_linked to be
// true. That way that doesnt have to be done in constructor and
// ++ operator.
int level_found = succs[0]->height - 1;
assert(succs[0] == succs[level_found]);
// for (int i = level_found; i >= 0; i--) {
// // Someone has done something
// if (preds[i]->forward(i) != succs[i]) {
for (auto it = MultiIterator<K>(skiplist, item); it.has_value();
it++) {
if (it.succs[0] == succs[0]) { // Found it
std::copy(it.preds, it.preds + H, preds);
std::copy(it.succs, it.succs + H, succs);
return level_found;
}
}
// Someone removed it
// assert(succs[0]->flags.is_marked());
return -1;
// }
// }
// // Everything is fine
// return level_found;
}
const K &data;
SkipList *skiplist;
Node *preds[H], *succs[H];
};
SkipList() : header(Node::sentinel(H)) {}
~SkipList()
{
// Someone could be using this map through an Accessor.
Node *now = header;
header = nullptr;
while (now != nullptr) {
Node *next = now->forward(0);
Node::destroy(now);
now = next;
}
}
friend class Accessor;
class Accessor
{
friend class SkipList;
Accessor(SkipList *skiplist) : skiplist(skiplist)
{
assert(skiplist != nullptr);
skiplist->gc.add_ref();
}
public:
Accessor(const Accessor &) = delete;
Accessor(Accessor &&other) : skiplist(other.skiplist)
{
other.skiplist = nullptr;
}
~Accessor()
{
if (skiplist == nullptr) return;
skiplist->gc.release_ref();
}
Iterator begin() { return skiplist->begin(); }
ConstIterator begin() const { return skiplist->cbegin(); }
ConstIterator cbegin() const { return skiplist->cbegin(); }
Iterator end() { return skiplist->end(); }
ConstIterator end() const { return skiplist->cend(); }
ConstIterator cend() const { return skiplist->cend(); }
template <class K>
MultiIterator<K> end(const K &data)
{
return skiplist->mend(data);
}
template <class K>
MultiIterator<K> mend(const K &data)
{
return skiplist->template mend<K>(data);
}
std::pair<Iterator, bool> insert(const T &item)
{
return skiplist->insert(item, preds, succs);
}
std::pair<Iterator, bool> insert(T &&item)
{
return skiplist->insert(std::forward<T>(item), preds, succs);
}
Iterator insert_non_unique(const T &item)
{
return skiplist->insert_non_unique(item, preds, succs);
}
Iterator insert_non_unique(T &&item)
{
return skiplist->insert_non_unique(std::forward<T>(item), preds,
succs);
}
template <class K>
MultiIterator<K> find_multi(const K &item) const
{
return MultiIterator<K>(this->skiplist, item);
}
template <class K>
ConstIterator find(const K &item) const
{
return static_cast<const SkipList &>(*skiplist).find(item);
}
template <class K>
Iterator find(const K &item)
{
return skiplist->find(item);
}
template <class K>
bool contains(const K &item) const
{
return this->find(item) != this->end();
}
template <class K>
bool remove(const K &item)
{
return skiplist->remove(item, preds, succs,
SkipList<T>::template find_path<K>);
}
size_t size() const { return skiplist->size(); }
private:
SkipList *skiplist;
Node *preds[H], *succs[H];
};
Accessor access() { return Accessor(this); }
const Accessor access() const { return Accessor(this); }
private:
using guard_t = std::unique_lock<lock_t>;
Iterator begin() { return Iterator(header->forward(0)); }
ConstIterator begin() const { return ConstIterator(header->forward(0)); }
ConstIterator cbegin() const { return ConstIterator(header->forward(0)); }
Iterator end() { return Iterator(); }
ConstIterator end() const { return ConstIterator(); }
ConstIterator cend() const { return ConstIterator(); }
template <class K>
MultiIterator<K> end(const K &data)
{
return MultiIterator<K>(data);
}
template <class K>
MultiIterator<K> mend(const K &data)
{
return MultiIterator<K>(data);
}
size_t size() const { return count.load(); }
template <class K>
static bool greater(const K &item, const Node *const node)
{
return node && item > node->value();
}
template <class K>
static bool less(const K &item, const Node *const node)
{
return (node == nullptr) || item < node->value();
}
// Returns first occurence of item if there exists one.
template <class K>
ConstIterator find(const K &item) const
{
return const_cast<SkipList *>(this)->find_node<ConstIterator, K>(item);
}
// Returns first occurence of item if there exists one.
template <class K>
Iterator find(const K &item)
{
return find_node<Iterator, K>(item);
}
template <class It, class K>
It find_node(const K &item)
{
Node *node, *pred = header;
int h = static_cast<int>(pred->height) - 1;
while (true) {
// try to descend down first the next key on this layer overshoots
for (; h >= 0 && less(item, node = pred->forward(h)); --h) {
}
// if we overshoot at every layer, item doesn't exist
if (h < 0) return It();
// the item is farther to the right, continue going right as long
// as the key is greater than the current node's key
while (greater(item, node))
pred = node, node = node->forward(h);
// check if we have a hit. if not, we need to descend down again
if (!less(item, node) && !node->flags.is_marked()) return It(node);
}
}
template <class K>
static int find_path(SkipList *skiplist, int start, const K &item,
Node *preds[], Node *succs[])
{
int level_found = -1;
Node *pred = skiplist->header;
for (int level = start; level >= 0; --level) {
Node *node = pred->forward(level);
while (greater(item, node))
pred = node, node = pred->forward(level);
if (level_found == -1 && !less(item, node)) level_found = level;
preds[level] = pred;
succs[level] = node;
}
return level_found;
}
template <bool ADDING>
static bool lock_nodes(uint8_t height, guard_t guards[], Node *preds[],
Node *succs[])
{
Node *prepred, *pred, *succ = nullptr;
bool valid = true;
for (int level = 0; valid && level < height; ++level) {
pred = preds[level], succ = succs[level];
if (pred != prepred)
guards[level] = pred->acquire_unique(), prepred = pred;
valid = !pred->flags.is_marked() && pred->forward(level) == succ;
if (ADDING)
valid = valid && (succ == nullptr || !succ->flags.is_marked());
}
return valid;
}
Iterator insert_non_unique(T &&data, Node *preds[], Node *succs[])
{
while (true) {
// TODO: before here was data.first
auto level = find_path(this, H - 1, data, preds, succs);
auto height = 1;
if (level != -1) {
auto found = succs[level];
if (found->flags.is_marked()) continue;
if (!found->flags.is_fully_linked()) {
usleep(250);
continue;
}
// TODO Optimization for avoiding degrading of skiplist to list.
// Somehow this operation is errornus.
// if (found->height == 1) { // To avoid linearization new
// element
// // will be at least 2 height and
// will
// // be added in front.
// height = rnd();
// // if (height == 1) height = 2;
// } else {
// Only level 0 will be used so the rest is irrelevant.
preds[0] = found;
succs[0] = found->forward(0);
// This maybe isn't necessary
auto next = succs[0];
if (next != nullptr) {
if (next->flags.is_marked()) continue;
if (!next->flags.is_fully_linked()) {
usleep(250);
continue;
}
}
} else {
height = rnd();
// Optimization which doesn't add any extra locking.
if (height == 1)
height = 2; // Same key list will be skipped more often.
}
guard_t guards[H];
// try to acquire the locks for predecessors up to the height of
// the new node. release the locks and try again if someone else
// has the locks
if (!lock_nodes<true>(height, guards, preds, succs)) continue;
return insert_here(std::forward<T>(data), preds, succs, height,
guards);
}
}
// Insert unique data
std::pair<Iterator, bool> insert(T &&data, Node *preds[], Node *succs[])
{
while (true) {
// TODO: before here was data.first
auto level = find_path(this, H - 1, data, preds, succs);
if (level != -1) {
auto found = succs[level];
if (found->flags.is_marked()) continue;
while (!found->flags.is_fully_linked())
usleep(250);
return {Iterator{succs[level]}, false};
}
auto height = rnd();
guard_t guards[H];
// try to acquire the locks for predecessors up to the height of
// the new node. release the locks and try again if someone else
// has the locks
if (!lock_nodes<true>(height, guards, preds, succs)) continue;
return {insert_here(std::forward<T>(data), preds, succs, height,
guards),
true};
}
}
// Inserts data to specified locked location.
Iterator insert_here(T &&data, Node *preds[], Node *succs[], int height,
guard_t guards[])
{
// you have the locks, create a new node
auto new_node = Node::create(std::forward<T>(data), height);
// link the predecessors and successors, e.g.
//
// 4 HEAD ... P ------------------------> S ... NULL
// 3 HEAD ... ... P -----> NEW ---------> S ... NULL
// 2 HEAD ... ... P -----> NEW -----> S ... ... NULL
// 1 HEAD ... ... ... P -> NEW -> S ... ... ... NULL
for (uint8_t level = 0; level < height; ++level) {
new_node->forward(level, succs[level]);
preds[level]->forward(level, new_node);
}
new_node->flags.set_fully_linked();
count.fetch_add(1);
return Iterator{new_node};
}
static bool ok_delete(Node *node, int level)
{
return node->flags.is_fully_linked() && node->height - 1 == level &&
!node->flags.is_marked();
}
// Remove item found with fp with arguments skiplist,preds and succs.
// fp has to fill preds and succs which reflect location of item or return
// -1 as in not found otherwise returns level on which the item was first
// found.
template <class K>
bool remove(const K &item, Node *preds[], Node *succs[],
int (*fp)(SkipList *, int, const K &, Node *[], Node *[]))
{
Node *node = nullptr;
guard_t node_guard;
bool marked = false;
int height = 0;
while (true) {
auto level = fp(this, H - 1, item, preds, succs);
if (!marked && (level == -1 || !ok_delete(succs[level], level)))
return false;
if (!marked) {
node = succs[level];
height = node->height;
node_guard = node->acquire_unique();
if (node->flags.is_marked()) return false;
node->flags.set_marked();
marked = true;
}
guard_t guards[H];
if (!lock_nodes<false>(height, guards, preds, succs)) continue;
for (int level = height - 1; level >= 0; --level)
preds[level]->forward(level, node->forward(level));
// TODO: review and test
gc.collect(node);
count.fetch_sub(1);
return true;
}
}
// number of elements
std::atomic<size_t> count{0};
Node *header;
SkiplistGC<Node> gc;
};
template <class T, size_t H, class lock_t>
thread_local FastBinomial<H> SkipList<T, H, lock_t>::rnd;

View File

@@ -0,0 +1,61 @@
#pragma once
// TODO: remove from here and from the project
#include <iostream>
#include "memory/freelist.hpp"
#include "memory/lazy_gc.hpp"
#include "threading/sync/spinlock.hpp"
#include "logging/default.hpp"
template <class T, class lock_t = SpinLock>
class SkiplistGC : public LazyGC<SkiplistGC<T, lock_t>, lock_t>
{
public:
SkiplistGC() : logger(logging::log->logger("SkiplistGC")) {}
// release_ref method should be called by a thread
// when the thread finish it job over object
// which has to be lazy cleaned
// if thread counter becames zero, all objects in the local_freelist
// are going to be deleted
// the only problem with this approach is that
// GC may never be called, but for now we can deal with that
void release_ref()
{
std::vector<T *> local_freelist;
// take freelist if there is no more threads
{
auto lock = this->acquire_unique();
assert(this->count > 0);
--this->count;
if (this->count == 0) {
freelist.swap(local_freelist);
}
}
if (local_freelist.size() > 0) {
logger.trace("GC started");
logger.trace("Local list size: {}", local_freelist.size());
long long counter = 0;
// destroy all elements from local_freelist
for (auto element : local_freelist) {
if (element->flags.is_marked()) {
T::destroy(element);
counter++;
}
}
logger.trace("Number of destroyed elements: {}", counter);
}
}
void collect(T *node) { freelist.add(node); }
protected:
Logger logger;
private:
FreeList<T> freelist;
};

View File

@@ -9,12 +9,18 @@ class Db
public:
using sptr = std::shared_ptr<Db>;
Db() = default;
Db(const std::string& name) : name_(name) {}
Db(const Db& db) = delete;
Graph graph;
tx::Engine tx_engine;
// only for test purposes
std::string identifier()
std::string& name()
{
return "memgraph";
return name_;
}
private:
std::string name_;
};

1
include/io/network/.gitignore vendored Normal file
View File

@@ -0,0 +1 @@
test

View File

@@ -0,0 +1,46 @@
#pragma once
#include <cstring>
#include <netdb.h>
#include "io/network/network_error.hpp"
#include "utils/underlying_cast.hpp"
namespace io
{
class AddrInfo
{
AddrInfo(struct addrinfo* info) : info(info) {}
public:
~AddrInfo()
{
freeaddrinfo(info);
}
static AddrInfo get(const char* addr, const char* port)
{
struct addrinfo hints;
memset(&hints, 0, sizeof(struct addrinfo));
hints.ai_family = AF_UNSPEC; // IPv4 and IPv6
hints.ai_socktype = SOCK_STREAM; // TCP socket
hints.ai_flags = AI_PASSIVE;
struct addrinfo* result;
auto status = getaddrinfo(addr, port, &hints, &result);
if(status != 0)
throw NetworkError(gai_strerror(status));
return AddrInfo(result);
}
operator struct addrinfo*() { return info; }
private:
struct addrinfo* info;
};
}

View File

@@ -0,0 +1,35 @@
#pragma once
#include "io/network/stream_reader.hpp"
namespace io
{
template <class Derived, class Stream>
class Client : public StreamReader<Derived, Stream>
{
public:
bool connect(const std::string& host, const std::string& port)
{
return connect(host.c_str(), port.c_str());
}
bool connect(const char* host, const char* port)
{
auto socket = io::Socket::connect(host, port);
if(!socket.is_open())
return false;
socket.set_non_blocking();
auto& stream = this->derived().on_connect(std::move(socket));
stream.event.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
this->add(stream);
return true;
}
};
}

View File

@@ -0,0 +1,54 @@
#pragma once
#include <malloc.h>
#include <sys/epoll.h>
#include "io/network/socket.hpp"
#include "utils/likely.hpp"
namespace io
{
class EpollError : BasicException
{
public:
using BasicException::BasicException;
};
class Epoll
{
public:
using Event = struct epoll_event;
Epoll(int flags)
{
epoll_fd = epoll_create1(flags);
if(UNLIKELY(epoll_fd == -1))
throw EpollError("Can't create epoll file descriptor");
}
template <class Stream>
void add(Stream& stream, Event* event)
{
auto status = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, stream, event);
if(UNLIKELY(status))
throw EpollError("Can't add an event to epoll listener.");
}
int wait(Event* events, int max_events, int timeout)
{
return epoll_wait(epoll_fd, events, max_events, timeout);
}
int id() const
{
return epoll_fd;
}
private:
int epoll_fd;
};
}

View File

@@ -0,0 +1,74 @@
#pragma once
#include "io/network/epoll.hpp"
#include "utils/crtp.hpp"
namespace io
{
template <class Derived, size_t max_events = 64, int wait_timeout = -1>
class EventListener : public Crtp<Derived>
{
public:
using Crtp<Derived>::derived;
EventListener(uint32_t flags = 0) : listener(flags) {}
void wait_and_process_events()
{
// TODO hardcoded a wait timeout because of thread joining
// when you shutdown the server. This should be wait_timeout of the
// template parameter and should almost never change from that.
// thread joining should be resolved using a signal that interrupts
// the system call.
// waits for an event/multiple events and returns a maximum of
// max_events and stores them in the events array. it waits for
// wait_timeout milliseconds. if wait_timeout is achieved, returns 0
auto n = listener.wait(events, max_events, 200);
// go through all events and process them in order
for(int i = 0; i < n; ++i)
{
auto& event = events[i];
// hangup event
if(UNLIKELY(event.events & EPOLLRDHUP))
{
this->derived().on_close_event(event);
continue;
}
// there was an error on the server side
if(UNLIKELY(!(event.events & EPOLLIN) ||
event.events & (EPOLLHUP | EPOLLERR)))
{
this->derived().on_error_event(event);
continue;
}
// we have some data waiting to be read
this->derived().on_data_event(event);
}
// this will be optimized out :D
if(wait_timeout < 0)
return;
// if there was events, continue to wait on new events
if(n != 0)
return;
// wait timeout occurred and there were no events. if wait_timeout
// is -1 there will never be any timeouts so client should provide
// an empty function. in that case the conditional above and the
// function call will be optimized out by the compiler
this->derived().on_wait_timeout();
}
protected:
Epoll listener;
Epoll::Event events[max_events];
};
}

View File

@@ -0,0 +1,25 @@
#pragma once
#include "io/network/socket.hpp"
namespace io
{
class TcpConnection
{
};
class EventLoop
{
public:
EventLoop()
{
}
private:
Even
};
}

View File

@@ -0,0 +1,16 @@
#pragma once
#include <stdexcept>
#include "utils/exceptions/basic_exception.hpp"
namespace io
{
class NetworkError : public BasicException
{
public:
using BasicException::BasicException;
};
}

View File

@@ -0,0 +1,93 @@
#pragma once
#include "tls.hpp"
#include "io/network/socket.hpp"
#include "tls_error.hpp"
#include "utils/types/byte.hpp"
#include <iostream>
namespace io
{
class SecureSocket
{
public:
SecureSocket(Socket&& socket, const Tls::Context& tls)
: socket(std::forward<Socket>(socket))
{
ssl = SSL_new(tls);
SSL_set_fd(ssl, this->socket);
SSL_set_accept_state(ssl);
if(SSL_accept(ssl) <= 0)
ERR_print_errors_fp(stderr);
}
SecureSocket(SecureSocket&& other)
{
*this = std::forward<SecureSocket>(other);
}
SecureSocket& operator=(SecureSocket&& other)
{
socket = std::move(other.socket);
ssl = other.ssl;
other.ssl = nullptr;
return *this;
}
~SecureSocket()
{
if(ssl == nullptr)
return;
std::cout << "DELETING SSL" << std::endl;
SSL_free(ssl);
}
int error(int status)
{
return SSL_get_error(ssl, status);
}
int write(const std::string& str)
{
return write(str.c_str(), str.size());
}
int write(const byte* data, size_t len)
{
return SSL_write(ssl, data, len);
}
int write(const char* data, size_t len)
{
return SSL_write(ssl, data, len);
}
int read(char* buffer, size_t len)
{
return SSL_read(ssl, buffer, len);
}
operator int()
{
return socket;
}
operator Socket&()
{
return socket;
}
private:
Socket socket;
SSL* ssl {nullptr};
};
}

View File

@@ -0,0 +1,61 @@
#pragma once
#include <openssl/ssl.h>
#include "io/network/stream_reader.hpp"
#include "logging/default.hpp"
namespace io
{
using namespace memory::literals;
template <class Derived, class Stream>
class SecureStreamReader : public StreamReader<Derived, Stream>
{
public:
struct Buffer
{
char* ptr;
size_t len;
};
SecureStreamReader(uint32_t flags = 0)
: StreamReader<Derived, Stream>(flags) {}
void on_data(Stream& stream)
{
while(true)
{
// allocate the buffer to fill the data
auto buf = this->derived().on_alloc(stream);
// read from the buffer at most buf.len bytes
auto len = stream.socket.read(buf.ptr, buf.len);
if(LIKELY(len > 0))
{
buf.len = len;
return this->derived().on_read(stream, buf);
}
auto err = stream.socket.error(len);
// the socket is not ready for reading yet
if(err == SSL_ERROR_WANT_READ ||
err == SSL_ERROR_WANT_WRITE ||
err == SSL_ERROR_WANT_X509_LOOKUP)
{
return;
}
// the socket notified a close event
if(err == SSL_ERROR_ZERO_RETURN)
return stream.close();
// some other error occurred, check errno
return this->derived().on_error(stream);
}
}
};
}

View File

@@ -0,0 +1,43 @@
#pragma once
#include "io/network/stream_reader.hpp"
namespace io
{
template <class Derived>
class Server : public EventListener<Derived>
{
public:
Server(Socket&& socket) : socket(std::forward<Socket>(socket))
{
event.data.fd = this->socket;
event.events = EPOLLIN | EPOLLET;
this->listener.add(this->socket, &event);
}
void on_close_event(Epoll::Event& event)
{
::close(event.data.fd);
}
void on_error_event(Epoll::Event& event)
{
::close(event.data.fd);
}
void on_data_event(Epoll::Event& event)
{
if(UNLIKELY(socket != event.data.fd))
return;
this->derived().on_connect();
}
protected:
Epoll::Event event;
Socket socket;
};
}

View File

@@ -0,0 +1,198 @@
#pragma once
#include <stdexcept>
#include <cstring>
#include <cstdio>
#include <cassert>
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <errno.h>
#include "io/network/addrinfo.hpp"
#include "utils/likely.hpp"
#include "logging/default.hpp"
#include <iostream>
namespace io
{
class Socket
{
protected:
Socket(int family, int socket_type, int protocol)
{
socket = ::socket(family, socket_type, protocol);
}
public:
using byte = uint8_t;
Socket(int socket = -1) : socket(socket) {}
Socket(const Socket&) = delete;
Socket(Socket&& other)
{
*this = std::forward<Socket>(other);
}
~Socket()
{
if(socket == -1)
return;
#ifndef NDEBUG
logging::debug("DELETING SOCKET");
#endif
::close(socket);
}
void close()
{
::close(socket);
socket = -1;
}
Socket& operator=(Socket&& other)
{
this->socket = other.socket;
other.socket = -1;
return *this;
}
bool is_open()
{
return socket != -1;
}
static Socket connect(const std::string& addr, const std::string& port)
{
return connect(addr.c_str(), port.c_str());
}
static Socket connect(const char* addr, const char* port)
{
auto info = AddrInfo::get(addr, port);
for(struct addrinfo* it = info; it != nullptr; it = it->ai_next)
{
auto s = Socket(it->ai_family, it->ai_socktype, it->ai_protocol);
if(!s.is_open())
continue;
if(::connect(s, it->ai_addr, it->ai_addrlen) == 0)
return s;
}
throw NetworkError("Unable to connect to socket");
}
static Socket bind(const std::string& addr, const std::string& port)
{
return bind(addr.c_str(), port.c_str());
}
static Socket bind(const char* addr, const char* port)
{
auto info = AddrInfo::get(addr, port);
for(struct addrinfo* it = info; it != nullptr; it = it->ai_next)
{
auto s = Socket(it->ai_family, it->ai_socktype, it->ai_protocol);
if(!s.is_open())
continue;
int on = 1;
if(setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)))
continue;
if(::bind(s, it->ai_addr, it->ai_addrlen) == 0)
return s;
}
throw NetworkError("Unable to bind to socket");
}
void set_non_blocking()
{
auto flags = fcntl(socket, F_GETFL, 0);
if(UNLIKELY(flags == -1))
throw NetworkError("Cannot read flags from socket");
flags |= O_NONBLOCK;
auto status = fcntl(socket, F_SETFL, flags);
if(UNLIKELY(status == -1))
throw NetworkError("Cannot set NON_BLOCK flag to socket");
}
void listen(int backlog)
{
auto status = ::listen(socket, backlog);
if(UNLIKELY(status == -1))
throw NetworkError("Cannot listen on socket");
}
Socket accept(struct sockaddr* addr, socklen_t* len)
{
return Socket(::accept(socket, addr, len));
}
operator int() { return socket; }
int id() const
{
return socket;
}
int write(const std::string& str)
{
return write(str.c_str(), str.size());
}
int write(const char* data, size_t len)
{
return write(reinterpret_cast<const byte*>(data), len);
}
int write(const byte* data, size_t len)
{
// TODO: use logger
#ifndef NDEBUG
std::stringstream stream;
for(size_t i = 0; i < len; ++i)
stream << fmt::format("{:02X} ", static_cast<byte>(data[i]));
auto str = stream.str();
logging::debug("[Write {}B] {}", len, str);
#endif
return ::write(socket, data, len);
}
int read(void* buffer, size_t len)
{
return ::read(socket, buffer, len);
}
protected:
Logger logger;
int socket;
};
}

View File

@@ -0,0 +1,12 @@
#pragma once
namespace io
{
class StreamDispatcher
{
};
}

View File

@@ -0,0 +1,43 @@
#pragma once
#include "io/network/event_listener.hpp"
namespace io
{
template <class Derived, class Stream,
size_t max_events = 64, int wait_timeout = -1>
class StreamListener : public EventListener<Derived, max_events, wait_timeout>
{
public:
using EventListener<Derived, max_events, wait_timeout>::EventListener;
void add(Stream& stream)
{
// add the stream to the event listener
this->listener.add(stream.socket, &stream.event);
}
void on_close_event(Epoll::Event& event)
{
this->derived().on_close(to_stream(event));
}
void on_error_event(Epoll::Event& event)
{
this->derived().on_error(to_stream(event));
}
void on_data_event(Epoll::Event& event)
{
this->derived().on_data(to_stream(event));
}
private:
Stream& to_stream(Epoll::Event& event)
{
return *reinterpret_cast<Stream*>(event.data.ptr);
}
};
}

View File

@@ -0,0 +1,87 @@
#pragma once
#include "io/network/stream_listener.hpp"
#include "memory/literals.hpp"
namespace io
{
using namespace memory::literals;
template <class Derived, class Stream>
class StreamReader : public StreamListener<Derived, Stream>
{
public:
struct Buffer
{
char* ptr;
size_t len;
};
StreamReader(uint32_t flags = 0) : StreamListener<Derived, Stream>(flags) {}
bool accept(Socket& socket)
{
// accept a connection from a socket
auto s = socket.accept(nullptr, nullptr);
if(!s.is_open())
return false;
// make the recieved socket non blocking
s.set_non_blocking();
auto& stream = this->derived().on_connect(std::move(s));
// we want to listen to an incoming event which is edge triggered and
// we also want to listen on the hangup event
stream.event.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
// add the connection to the event listener
this->add(stream);
return true;
}
void on_data(Stream& stream)
{
while(true)
{
if(UNLIKELY(!stream.alive()))
{
stream.close();
break;
}
// allocate the buffer to fill the data
auto buf = this->derived().on_alloc(stream);
// read from the buffer at most buf.len bytes
buf.len = stream.socket.read(buf.ptr, buf.len);
// check for read errors
if(buf.len == -1)
{
// this means we have read all available data
if(LIKELY(errno == EAGAIN))
{
break;
}
// some other error occurred, check errno
this->derived().on_error(stream);
break;
}
// end of file, the client has closed the connection
if(UNLIKELY(buf.len == 0))
{
stream.close();
break;
}
this->derived().on_read(stream, buf);
}
}
};
}

View File

@@ -0,0 +1,35 @@
#pragma once
#include "io/network/epoll.hpp"
#include "io/network/socket.hpp"
namespace io
{
namespace tcp
{
template <class Socket>
class Stream
{
public:
Stream(Socket&& socket) : socket(std::move(socket))
{
// save this to epoll event data baton to access later
event.data.ptr = this;
}
Stream(Stream&& stream)
{
socket = std::move(stream.socket);
event = stream.event;
event.data.ptr = this;
}
int id() const { return socket.id(); }
Socket socket;
Epoll::Event event;
};
}
}

View File

@@ -0,0 +1,33 @@
#pragma once
#include <string>
#include <openssl/ssl.h>
#include <openssl/err.h>
namespace io
{
class Tls
{
public:
class Context
{
public:
Context();
~Context();
Context& cert(const std::string& path);
Context& key(const std::string& path);
operator SSL_CTX*() const { return ctx; }
private:
SSL_CTX* ctx;
};
static void initialize();
static void cleanup();
};
}

View File

@@ -0,0 +1,14 @@
#pragma once
#include "utils/exceptions/basic_exception.hpp"
namespace io
{
class TlsError : public BasicException
{
public:
using BasicException::BasicException;
};
}

View File

@@ -0,0 +1,125 @@
#pragma once
#include <cstring>
#include <uv.h>
#include "utils/memory/block_allocator.hpp"
namespace uv
{
template <size_t block_size>
class BlockBuffer
{
static BlockAllocator<block_size> allocator;
struct Block : public uv_buf_t
{
Block()
{
// acquire a new block of memory for this buffer
base = static_cast<char*>(allocator.acquire());
len = 0;
}
~Block()
{
// release the block of memory previously acquired
allocator.release(base);
}
size_t append(const char* data, size_t size)
{
// compute the remaining capacity for this block
auto capacity = block_size - len;
// if the capacity is smaller than the requested size, copy only
// up to the remaining capacity
if(size > capacity)
size = capacity;
std::memcpy(base + len, data, size);
len += size;
// return how much we've copied
return size;
}
};
public:
BlockBuffer()
{
// create the first buffer
buffers.emplace_back();
}
~BlockBuffer()
{
buffers.clear();
}
BlockBuffer(BlockBuffer&) = delete;
BlockBuffer(BlockBuffer&&) = delete;
size_t count() const
{
return buffers.size();
}
void clear()
{
// pop all buffers except for the first one since we need to allocate
// it again anyway so why not keep it in the first place
while(count() > 1)
buffers.pop_back();
// pretend we just allocated our first buffer and set it's length to 0
buffers.back().len = 0;
}
BlockBuffer& operator<<(const std::string& data)
{
append(data);
return *this;
}
void append(const std::string& data)
{
append(data.c_str(), data.size());
}
void append(const char* data, size_t size)
{
while(true)
{
// try to copy as much data as possible
auto copied = buffers.back().append(data, size);
// if we managed to copy everything, we're done
if(copied == size)
break;
// move the pointer past the copied part
data += copied;
// reduce the total size by the number of copied items
size -= copied;
// since we ran out of space, construct a new buffer
buffers.emplace_back();
}
}
operator uv_buf_t*()
{
return buffers.data();
}
private:
std::vector<Block> buffers;
};
template <size_t block_size>
BlockAllocator<block_size> BlockBuffer<block_size>::allocator;
}

10
include/io/uv/core.hpp Normal file
View File

@@ -0,0 +1,10 @@
#pragma once
#include <uv.h>
namespace uv
{
using callback_t = void (*)(uv_handle_t *);
}

View File

@@ -0,0 +1,32 @@
#pragma once
#include <uv.h>
#include "core.hpp"
#include "uvloop.hpp"
namespace uv
{
class TcpStream
{
public:
TcpStream(UvLoop& loop);
template <typename T>
T* data();
template <typename T>
void data(T* value);
void close(callback_t callback);
operator uv_handle_t*();
operator uv_tcp_t*();
operator uv_stream_t*();
private:
uv_tcp_t stream;
};
}

7
include/io/uv/uv.hpp Normal file
View File

@@ -0,0 +1,7 @@
#include "tcpstream.hpp"
#include "uvbuffer.hpp"
#include "uvloop.hpp"
#include "blockbuffer.hpp"
#include "tcpstream.inl"
#include "uvbuffer.inl"

View File

@@ -0,0 +1,16 @@
#pragma once
#include <stdexcept>
#include <string>
namespace uv
{
class UvError : public std::runtime_error
{
public:
UvError(const std::string& message)
: std::runtime_error(message) {}
};
}

View File

@@ -0,0 +1,35 @@
#pragma once
#include <string>
#include <uv.h>
namespace uv
{
class UvBuffer
{
public:
UvBuffer();
UvBuffer(size_t capacity);
UvBuffer(const std::string& data);
~UvBuffer();
size_t size() const noexcept;
size_t length() const noexcept;
void clear();
UvBuffer& append(const std::string& data);
UvBuffer& append(const char* data, size_t n);
UvBuffer& operator<<(const std::string& data);
operator uv_buf_t*();
private:
uv_buf_t buffer;
size_t capacity;
};
}

59
include/io/uv/uvloop.hpp Normal file
View File

@@ -0,0 +1,59 @@
#pragma once
#include <memory>
#include <uv.h>
#include "uv_error.hpp"
namespace uv
{
class UvLoop final
{
public:
using sptr = std::shared_ptr<UvLoop>;
enum Mode {
Default = UV_RUN_DEFAULT,
Once = UV_RUN_ONCE,
NoWait = UV_RUN_NOWAIT
};
UvLoop()
{
uv_loop = uv_default_loop();
if(uv_loop == nullptr)
throw UvError("Failed to initialize libuv event loop");
}
~UvLoop()
{
uv_loop_close(uv_loop);
}
bool run(Mode mode)
{
return uv_run(uv_loop, static_cast<uv_run_mode>(mode));
}
bool alive()
{
return uv_loop_alive(uv_loop);
}
void stop()
{
uv_stop(uv_loop);
}
operator uv_loop_t*()
{
return uv_loop;
}
private:
uv_loop_t* uv_loop;
};
}

View File

@@ -0,0 +1,30 @@
#pragma once
#include "logging/log.hpp"
#include "logging/logger.hpp"
namespace logging
{
extern std::unique_ptr<Log> log;
extern Logger debug_logger;
template <class... Args>
void debug(Args&&... args)
{
debug_logger.debug(std::forward<Args>(args)...);
}
extern Logger info_logger;
template <class... Args>
void info(Args&&... args)
{
info_logger.info(std::forward<Args>(args)...);
}
void init_async();
void init_sync();
}

View File

@@ -0,0 +1,33 @@
#pragma once
#include <string>
struct Trace
{
static std::string text;
static constexpr unsigned level = 0;
};
struct Debug
{
static std::string text;
static constexpr unsigned level = 10;
};
struct Info
{
static std::string text;
static constexpr unsigned level = 20;
};
struct Warn
{
static std::string text;
static constexpr unsigned level = 30;
};
struct Error
{
static std::string text;
static constexpr unsigned level = 40;
};

64
include/logging/log.hpp Normal file
View File

@@ -0,0 +1,64 @@
#pragma once
#include <vector>
#include <string>
#include "utils/datetime/timestamp.hpp"
class Logger;
class Log
{
public:
using uptr = std::unique_ptr<Log>;
class Record
{
public:
using uptr = std::unique_ptr<Record>;
Record() = default;
virtual ~Record() = default;
virtual const Timestamp& when() const = 0;
virtual const std::string& where() const = 0;
virtual unsigned level() const = 0;
virtual const std::string& level_str() const = 0;
virtual const std::string& text() const = 0;
};
class Stream
{
public:
using uptr = std::unique_ptr<Stream>;
Stream() = default;
virtual ~Stream() = default;
virtual void emit(const Record&) = 0;
};
virtual ~Log() = default;
Logger logger(const std::string& name);
void pipe(Stream::uptr&& stream)
{
streams.emplace_back(std::forward<Stream::uptr>(stream));
}
protected:
friend class Logger;
virtual void emit(Record::uptr record) = 0;
void dispatch(const Record& record)
{
for(auto& stream : streams)
stream->emit(record);
}
std::vector<Stream::uptr> streams;
};

114
include/logging/logger.hpp Normal file
View File

@@ -0,0 +1,114 @@
#pragma once
#include <cassert>
#include <fmt/format.h>
#include "logging/log.hpp"
#include "logging/levels.hpp"
class Logger
{
template <class Level>
class Message : public Log::Record
{
public:
Message(Timestamp timestamp, std::string location, std::string message)
: timestamp(timestamp), location(location), message(message) {}
const Timestamp& when() const override
{
return timestamp;
}
const std::string& where() const override
{
return location;
}
unsigned level() const override
{
return Level::level;
}
const std::string& level_str() const override
{
return Level::text;
}
const std::string& text() const override
{
return message;
}
private:
Timestamp timestamp;
std::string location;
std::string message;
};
public:
Logger() = default;
Logger(Log* log, const std::string& name) : log(log), name(name) {}
template <class Level, class... Args>
void emit(Args&&... args)
{
assert(log != nullptr);
auto message = std::make_unique<Message<Level>>(
Timestamp::now(), name, fmt::format(std::forward<Args>(args)...)
);
log->emit(std::move(message));
}
template <class... Args>
void trace(Args&&... args)
{
#ifndef NDEBUG
#ifndef LOG_NO_TRACE
emit<Trace>(std::forward<Args>(args)...);
#endif
#endif
}
template <class... Args>
void debug(Args&&... args)
{
#ifndef NDEBUG
#ifndef LOG_NO_DEBUG
emit<Debug>(std::forward<Args>(args)...);
#endif
#endif
}
template <class... Args>
void info(Args&&... args)
{
#ifndef LOG_NO_INFO
emit<Info>(std::forward<Args>(args)...);
#endif
}
template <class... Args>
void warn(Args&&... args)
{
#ifndef LOG_NO_WARN
emit<Warn>(std::forward<Args>(args)...);
#endif
}
template <class... Args>
void error(Args&&... args)
{
#ifndef LOG_NO_ERROR
emit<Error>(std::forward<Args>(args)...);
#endif
}
private:
Log* log;
std::string name;
};

View File

@@ -0,0 +1,22 @@
#pragma once
#include <thread>
#include "logging/log.hpp"
#include "data_structures/queue/mpsc_queue.hpp"
class AsyncLog : public Log
{
public:
~AsyncLog();
protected:
void emit(Record::uptr) override;
private:
lockfree::MpscQueue<Record> records;
std::atomic<bool> alive {true};
std::thread worker {[this]() { work(); }};
void work();
};

View File

@@ -0,0 +1,11 @@
#pragma once
#include "logging/log.hpp"
#include "threading/sync/lockable.hpp"
#include "threading/sync/futex.hpp"
class SyncLog : public Log, Lockable<Futex>
{
protected:
void emit(Record::uptr) override;
};

View File

@@ -0,0 +1,9 @@
#pragma once
#include "logging/log.hpp"
class Stdout : public Log::Stream
{
public:
void emit(const Log::Record&) override;
};

View File

@@ -0,0 +1,27 @@
#pragma once
#include <vector>
#include "threading/sync/lockable.hpp"
#include "threading/sync/spinlock.hpp"
template <class T, class lock_t = SpinLock>
class FreeList : Lockable<lock_t>
{
public:
void swap(std::vector<T *> &dst) { std::swap(data, dst); }
void add(T *element)
{
auto lock = this->acquire_unique();
data.emplace_back(element);
}
size_t size() const
{
return data.size();
}
private:
std::vector<T *> data;
};

View File

@@ -0,0 +1,26 @@
#pragma once
// TODO: remove from here and from the project
#include <atomic>
#include <iostream>
#include "threading/sync/lockable.hpp"
#include "utils/crtp.hpp"
template <class Derived, class lock_t = SpinLock>
class LazyGC : public Crtp<Derived>, public Lockable<lock_t>
{
public:
// add_ref method should be called by a thread
// when the thread has to do something over
// object which has to be lazy cleaned when
// the thread finish it job
void add_ref()
{
auto lock = this->acquire_unique();
++count;
}
protected:
size_t count{0};
};

39
include/mvcc/cre_exp.hpp Normal file
View File

@@ -0,0 +1,39 @@
#pragma once
#include <atomic>
namespace mvcc
{
template <class T>
class CreExp
{
public:
CreExp() = default;
CreExp(T cre, T exp) : cre_(cre), exp_(exp) {}
T cre(std::memory_order order = std::memory_order_seq_cst) const
{
return cre_.load(order);
}
void cre(T value, std::memory_order order = std::memory_order_seq_cst)
{
cre_.store(value, order);
}
T exp(std::memory_order order = std::memory_order_seq_cst) const
{
return exp_.load(order);
}
void exp(T value, std::memory_order order = std::memory_order_seq_cst)
{
exp_.store(value, order);
}
private:
std::atomic<T> cre_ {0}, exp_ {0};
};
}

111
include/mvcc/hints.hpp Normal file
View File

@@ -0,0 +1,111 @@
#pragma once
#include <atomic>
#include <unistd.h>
#include "transactions/commit_log.hpp"
namespace mvcc
{
// known committed and known aborted for both cre and exp
// this hints are used to quickly check the commit/abort status of the
// transaction that created this record. if these are not set, one should
// consult the commit log to find the status and update the status here
// more info https://wiki.postgresql.org/wiki/Hint_Bits
class Hints
{
public:
union HintBits;
private:
enum Flags : uint8_t {
CRE_CMT = 0x01, // __01
CRE_ABT = 0x02, // __10
EXP_CMT = 0x04, // 01__
EXP_ABT = 0x08 // 10__
};
template <Flags COMMITTED, Flags ABORTED>
class TxHints
{
using type = TxHints<COMMITTED, ABORTED>;
public:
TxHints(std::atomic<uint8_t>& bits) : bits(bits) {}
struct Value
{
bool is_committed() const
{
return bits & COMMITTED;
}
bool is_aborted() const
{
return bits & ABORTED;
}
bool is_unknown() const
{
return !(is_committed() || is_aborted());
}
uint8_t bits;
};
Value load(std::memory_order order = std::memory_order_seq_cst)
{
return Value { bits.load(order) };
}
void set_committed(std::memory_order order = std::memory_order_seq_cst)
{
bits.fetch_or(COMMITTED, order);
}
void set_aborted(std::memory_order order = std::memory_order_seq_cst)
{
bits.fetch_or(ABORTED, order);
}
private:
std::atomic<uint8_t>& bits;
};
struct Cre : public TxHints<CRE_CMT, CRE_ABT>
{
using TxHints::TxHints;
};
struct Exp : public TxHints<EXP_CMT, EXP_ABT>
{
using TxHints::TxHints;
};
public:
Hints() : cre(bits), exp(bits)
{
assert(bits.is_lock_free());
}
union HintBits
{
uint8_t bits;
Cre::Value cre;
Exp::Value exp;
};
HintBits load(std::memory_order order = std::memory_order_seq_cst)
{
return HintBits { bits.load(order) };
}
Cre cre;
Exp exp;
std::atomic<uint8_t> bits { 0 };
};
}

View File

@@ -0,0 +1,14 @@
#pragma once
#include <stdexcept>
namespace mvcc
{
class MvccError : public std::runtime_error
{
public:
using runtime_error::runtime_error;
};
}

137
include/mvcc/record.hpp Normal file
View File

@@ -0,0 +1,137 @@
#pragma once
#include <atomic>
#include <iostream>
#include "transactions/transaction.hpp"
#include "transactions/commit_log.hpp"
#include "transactions/engine.hpp"
#include "mvcc/id.hpp"
#include "mvcc/cre_exp.hpp"
#include "mvcc/version.hpp"
#include "mvcc/hints.hpp"
#include "storage/locking/record_lock.hpp"
// the mvcc implementation used here is very much like postgresql's
// more info: https://momjian.us/main/writings/pgsql/mvcc.pdf
namespace mvcc
{
template <class T>
class Record : public Version<T>
{
public:
// tx.cre is the id of the transaction that created the record
// and tx.exp is the id of the transaction that deleted the record
// these values are used to determine the visibility of the record
// to the current transaction
CreExp<Id> tx;
// cmd.cre is the id of the command in this transaction that created the
// record and cmd.exp is the id of the command in this transaction that
// deleted the record. these values are used to determine the visibility
// of the record to the current command in the running transaction
CreExp<uint8_t> cmd;
Hints hints;
// this lock is used by write queries when they update or delete records
RecordLock lock;
// check if this record is visible to the transaction t
bool visible(const tx::Transaction& t)
{
// TODO check if the record was created by a transaction that has been
// aborted. one might implement this by checking the hints in mvcc
// anc/or consulting the commit log
// Mike Olson says 17 march 1993: the tests in this routine are correct;
// if you think they're not, you're wrong, and you should think about it
// again. i know, it happened to me.
return ((tx.cre() == t.id && // inserted by the current transaction
cmd.cre() <= t.cid && // before this command, and
(tx.exp() == Id(0) || // the row has not been deleted, or
(tx.exp() == t.id && // it was deleted by the current
// transaction
cmd.exp() >= t.cid))) // but not before this command,
|| // or
(cre_committed(tx.cre(), t) && // the record was inserted by a
// committed transaction, and
(tx.exp() == Id(0) || // the record has not been deleted, or
(tx.exp() == t.id && // the row is being deleted by this
// transaction
cmd.exp() >= t.cid) || // but it's not deleted "yet", or
(tx.exp() != t.id && // the row was deleted by another
// transaction
!exp_committed(tx.exp(), t) // that has not been committed
))));
}
void mark_created(const tx::Transaction& t)
{
tx.cre(t.id);
cmd.cre(t.cid);
}
void mark_deleted(const tx::Transaction& t)
{
tx.exp(t.id);
cmd.exp(t.cid);
}
bool exp_committed(const Id& id, const tx::Transaction& t)
{
return committed(hints.exp, id, t);
}
bool exp_committed(const tx::Transaction& t)
{
return committed(hints.exp, tx.exp(), t);
}
bool cre_committed(const Id& id, const tx::Transaction& t)
{
return committed(hints.cre, id, t);
}
bool cre_committed(const tx::Transaction& t)
{
return committed(hints.cre, tx.cre(), t);
}
protected:
template <class U>
bool committed(U& hints, const Id& id, const tx::Transaction& t)
{
// you certainly can't see the transaction with id greater than yours
// as that means it started after this transaction and if it committed,
// it committed after this transaction had started.
if(id > t.id)
return false;
auto hint_bits = hints.load();
// if hints are set, return if xid is committed
if(!hint_bits.is_unknown())
return hint_bits.is_committed();
// if hints are not set:
// - the creating transaction is still in progress (examine snapshot)
if(t.snapshot.is_active(id))
return false;
// - you are the first one to check since it ended, consult commit log
auto info = t.engine.clog.fetch_info(id);
if(info.is_committed())
return hints.set_committed(), true;
assert(info.is_aborted());
return hints.set_aborted(), false;
}
};
}

View File

@@ -0,0 +1,15 @@
#pragma once
#include <stdexcept>
class SerializationError : public std::runtime_error
{
static constexpr const char* default_message = "Can't serialize due to\
concurrent operation(s)";
public:
SerializationError() : runtime_error(default_message) {}
SerializationError(const std::string& message)
: runtime_error(message) {}
};

41
include/mvcc/version.hpp Normal file
View File

@@ -0,0 +1,41 @@
#pragma once
#include <atomic>
namespace mvcc
{
template <class T>
class Version
{
public:
Version() = default;
Version(T* older) : older(older) {}
~Version()
{
delete older.load(std::memory_order_seq_cst);
}
// return a pointer to an older version stored in this record
T* next(std::memory_order order = std::memory_order_seq_cst)
{
return older.load(order);
}
const T* next(std::memory_order order = std::memory_order_seq_cst) const
{
return older.load(order);
}
// set the older version of this record
void next(T* value, std::memory_order order = std::memory_order_seq_cst)
{
older.store(value, order);
}
private:
std::atomic<T*> older {nullptr};
};
}

View File

@@ -0,0 +1,175 @@
#pragma once
#include "threading/sync/lockable.hpp"
#include "transactions/transaction.hpp"
#include "memory/lazy_gc.hpp"
#include "mvcc/serialization_error.hpp"
#include "storage/locking/record_lock.hpp"
namespace mvcc
{
template <class T>
class VersionList : public LazyGC<VersionList<T>>
{
friend class Accessor;
public:
using uptr = std::unique_ptr<VersionList<T>>;
using item_t = T;
VersionList(Id id) : id(id) {}
VersionList(const VersionList &) = delete;
/* @brief Move constructs the version list
* Note: use only at the beginning of the "other's" lifecycle since this
* constructor doesn't move the RecordLock, but only the head pointer
*/
VersionList(VersionList &&other) : id(other.id)
{
this->head = other.head.load();
other.head = nullptr;
}
~VersionList() { delete head.load(); }
friend std::ostream &operator<<(std::ostream &stream,
const VersionList<T> &vlist)
{
stream << "VersionList" << std::endl;
auto record = vlist.head.load();
while (record != nullptr) {
stream << "-- " << *record << std::endl;
record = record->next();
}
return stream;
}
auto gc_lock_acquire() { return std::unique_lock<RecordLock>(lock); }
void vacuum() {}
T *find(const tx::Transaction &t) const
{
auto r = head.load(std::memory_order_seq_cst);
// nullptr
// |
// [v1] ...
// |
// [v2] <------+
// | |
// [v3] <------+
// | | Jump backwards until you find a first visible
// [VerList] ----+ version, or you reach the end of the list
//
while (r != nullptr && !r->visible(t))
r = r->next(std::memory_order_seq_cst);
return r;
}
T *insert(tx::Transaction &t)
{
assert(head == nullptr);
// create a first version of the record
// TODO replace 'new' with something better
auto v1 = new T();
// mark the record as created by the transaction t
v1->mark_created(t);
head.store(v1, std::memory_order_seq_cst);
return v1;
}
T *update(tx::Transaction &t)
{
assert(head != nullptr);
auto record = find(t);
// check if we found any visible records
if (!record) return nullptr;
return update(record, t);
}
T *update(T *record, tx::Transaction &t)
{
assert(record != nullptr);
// TODO: VALIDATE NEXT IF BLOCK
if (record->tx.cre() == t.id) {
// THEN ONLY THIS TRANSACTION CAN SEE THIS DATA WHICH MENS THAT IT
// CAN CHANGE IT.
return record;
}
lock_and_validate(record, t);
auto updated = new T();
updated->data = record->data;
updated->mark_created(t);
record->mark_deleted(t);
updated->next(record, std::memory_order_seq_cst);
head.store(updated, std::memory_order_seq_cst);
return updated;
}
bool remove(tx::Transaction &t)
{
assert(head != nullptr);
auto record = find(t);
if (!record) return false;
lock_and_validate(record, t);
return remove(record, t), true;
}
bool remove(T *record, tx::Transaction &t)
{
assert(record != nullptr);
lock_and_validate(record, t);
record->mark_deleted(t);
return true;
}
const Id id;
private:
void lock_and_validate(T *record, tx::Transaction &t)
{
assert(record != nullptr);
assert(record == find(t));
// take a lock on this node
t.take_lock(lock);
// if the record hasn't been deleted yet or the deleting transaction
// has aborted, it's ok to modify it
if (!record->tx.exp() || !record->exp_committed(t)) return;
// if it committed, then we have a serialization conflict
assert(record->hints.load().exp.is_committed());
throw SerializationError();
}
std::atomic<T *> head{nullptr};
RecordLock lock;
};
}
class Vertex;
class Edge;
using VertexRecord = mvcc::VersionList<Vertex>;
// using EdgeRecord = mvcc::VersionList<Edge>;

View File

@@ -0,0 +1,49 @@
#pragma once
#include <string>
#include "exceptions/exceptions.hpp"
#include "logging/default.hpp"
#include "utils/string/join.hpp"
// TODO:
// * all libraries have to be compiled in the server compile time
// * compile command has to be generated
class CodeCompiler
{
public:
CodeCompiler() : logger(logging::log->logger("CodeCompiler")) {}
void compile(const std::string &in_file, const std::string &out_file)
{
// generate compile command
auto compile_command = utils::prints(
"clang++",
// "-std=c++1y -O2 -DNDEBUG", // compile flags
"-std=c++1y", // compile flags // TODO: load from config file
in_file, // input file
"-o", out_file, // ouput file
"-I./include", // include paths (TODO: parameter)
"-I../libs/fmt", // TODO: load from config
"-L./ -lmemgraph_pic",
"-shared -fPIC" // shared library flags
);
// synchronous call
auto compile_status = system(compile_command.c_str());
// if compilation has failed throw exception
if (compile_status == -1) {
throw QueryEngineException("Code compilation error. Generated code "
"is not compilable or compilation "
"settings are wrong");
}
logger.debug("SUCCESS: Query Code Compilation: {} -> {}", in_file,
out_file);
}
protected:
Logger logger;
};

View File

@@ -0,0 +1,70 @@
#pragma once
#include "config/config.hpp"
#include "cypher/ast/ast.hpp"
#include "cypher/compiler.hpp"
#include "query_engine/exceptions/errors.hpp"
#include "template_engine/engine.hpp"
#include "traverser/cpp_traverser.hpp"
#include "utils/string/file.hpp"
#include "logging/default.hpp"
using std::string;
class CodeGenerator
{
public:
CodeGenerator() : logger(logging::log->logger("CodeGenerator")) {}
void generate_cpp(const std::string &query, const uint64_t stripped_hash,
const std::string &path)
{
// TODO: optimize; one time initialization -> be careful that object
// has a state
// TODO: multithread test
CppTraverser cpp_traverser;
// get paths
string template_path = CONFIG(config::TEMPLATE_CPU_CPP_PATH);
string template_file = utils::read_file(template_path.c_str());
// syntax tree generation
try {
tree = compiler.syntax_tree(query);
} catch (const std::runtime_error &e) {
logger.error("Syntax error: {}", query);
throw QueryEngineException(std::string(e.what()));
}
cpp_traverser.reset();
// code generation
try {
tree.root->accept(cpp_traverser);
} catch (const SemanticError &e) {
throw e;
} catch (const std::exception &e) {
logger.error("AST traversal error: {}", std::string(e.what()));
throw QueryEngineException("Unknown code generation error");
}
// save the code
string generated = template_engine.render(
template_file, {{"class_name", "CodeCPU"},
{"stripped_hash", std::to_string(stripped_hash)},
{"query", query},
{"code", cpp_traverser.code}});
logger.trace("generated code: {}", generated);
utils::write_file(generated, path);
}
protected:
Logger logger;
private:
template_engine::TemplateEngine template_engine;
ast::Ast tree;
cypher::Compiler compiler;
};

View File

@@ -0,0 +1,20 @@
#pragma once
#include <cstdint>
enum class ClauseAction : uint32_t
{
Undefined,
CreateNode,
MatchNode,
UpdateNode,
DeleteNode,
CreateRelationship,
MatchRelationship,
UpdateRelationship,
DeleteRelationship,
ReturnNode,
ReturnRelationship,
ReturnPack,
ReturnProjection
};

View File

@@ -0,0 +1,84 @@
#pragma once
#include <vector>
#include "query_engine/code_generator/cypher_state.hpp"
#include "query_engine/code_generator/handlers/all.hpp"
#include "query_engine/code_generator/query_action.hpp"
#include "query_engine/exceptions/exceptions.hpp"
class CppGenerator
{
public:
// !! multithread problem
// two threads shouldn't use this implementation at the same time
// !! TODO: REFACTOR
CppGenerator() : unprocessed_index(0), processed_index(0) { setup(); }
void state(CypherState state) { _cypher_state = state; }
CypherState state() { return _cypher_state; }
std::string generate()
{
std::string code = "";
for (uint64_t i = processed_index; i < unprocessed_index; ++i) {
auto &action = actions.at(i);
auto query_action = action.first;
if (action_functions.find(query_action) == action_functions.end())
throw CppGeneratorException(
"Query Action Function is not defined");
auto &action_data = action.second;
code += action_functions[query_action](_cypher_data, action_data);
++processed_index;
}
return code;
}
QueryActionData &add_action(const QueryAction &query_action)
{
unprocessed_index++;
actions.push_back(std::make_pair(query_action, QueryActionData()));
return action_data();
}
QueryActionData &action_data() { return actions.back().second; }
CypherStateData &cypher_data() { return _cypher_data; }
void clear()
{
processed_index = 0;
unprocessed_index = 0;
actions.clear();
}
private:
// TODO: setup function is going to be called every time
// when object of this class is constructed (optimize this)
void setup()
{
action_functions[QueryAction::TransactionBegin] =
transaction_begin_action;
action_functions[QueryAction::Create] = create_query_action;
action_functions[QueryAction::Match] = match_query_action;
action_functions[QueryAction::Return] = return_query_action;
action_functions[QueryAction::Set] = set_query_action;
action_functions[QueryAction::Delete] = delete_query_action;
action_functions[QueryAction::TransactionCommit] =
transaction_commit_action;
}
uint64_t unprocessed_index;
uint64_t processed_index;
std::vector<std::pair<QueryAction, QueryActionData>> actions;
std::map<QueryAction,
std::function<std::string(CypherStateData &cypher_data,
QueryActionData &action_data)>>
action_functions;
CypherState _cypher_state;
CypherStateData _cypher_data;
};

View File

@@ -0,0 +1,86 @@
#pragma once
#include <cstdint>
#include <map>
#include <string>
// main states that are used while ast is traversed
// in order to generate ActionSequence
enum class CypherState : uint8_t
{
Undefined,
Match,
Where,
Create,
Set,
Return,
Delete
};
enum class EntityStatus : uint8_t
{
NotFound,
Matched,
Created
};
enum class EntityType : uint8_t
{
NotFound,
Node,
Relationship
};
class CypherStateData
{
private:
std::map<std::string, EntityStatus> entity_status;
std::map<std::string, EntityType> entity_type;
// TODO: container that keeps track about c++ variable names
public:
bool exist(const std::string& name) const
{
return entity_status.find(name) != entity_status.end();
}
EntityStatus status(const std::string &name)
{
if (entity_status.find(name) == entity_status.end())
return EntityStatus::NotFound;
return entity_status.at(name);
}
EntityType type(const std::string &name)
{
if (entity_type.find(name) == entity_type.end())
return EntityType::NotFound;
return entity_type.at(name);
}
void node_matched(const std::string &name)
{
entity_type[name] = EntityType::Node;
entity_status[name] = EntityStatus::Matched;
}
void node_created(const std::string &name)
{
entity_type[name] = EntityType::Node;
entity_status[name] = EntityStatus::Created;
}
void relationship_matched(const std::string &name)
{
entity_type[name] = EntityType::Relationship;
entity_status[name] = EntityStatus::Matched;
}
void relationship_created(const std::string &name)
{
entity_type[name] = EntityType::Relationship;
entity_status[name] = EntityStatus::Created;
}
};

View File

@@ -0,0 +1,127 @@
#pragma once
#include <algorithm>
#include <limits>
#include <map>
// TODO: remove
#include "utils/underlying_cast.hpp"
#include <iostream>
// entities are nodes or relationship
namespace entity_search
{
// returns maximum value for given template argument (for give type)
template <typename T>
constexpr T max()
{
return std::numeric_limits<uint64_t>::max();
}
using cost_t = uint64_t;
// TODO: rething
// at least load hard coded values from somewhere
constexpr cost_t internal_id_cost = 10;
constexpr cost_t property_cost = 100;
constexpr cost_t label_cost = 1000;
constexpr cost_t max_cost = max<cost_t>();
template <typename T>
class SearchCost
{
public:
enum class SearchPlace : int
{
internal_id,
label_index,
property_index,
main_storage
};
using costs_t = std::map<SearchPlace, T>;
using cost_pair_t = std::pair<SearchPlace, T>;
SearchCost()
{
costs[SearchPlace::internal_id] = max<T>();
costs[SearchPlace::label_index] = max<T>();
costs[SearchPlace::property_index] = max<T>();
costs[SearchPlace::main_storage] = max<T>();
}
SearchCost(const SearchCost &other) = default;
SearchCost(SearchCost &&other) : costs(std::move(other.costs)) {}
void set(SearchPlace place, T cost) { costs[place] = cost; }
T get(SearchPlace place) const { return costs.at(place); }
SearchPlace min() const
{
auto min_pair = std::min_element(
costs.begin(), costs.end(),
[](const cost_pair_t &l, const cost_pair_t &r) -> bool {
return l.second < r.second;
});
if (min_pair->second == max_cost) return SearchPlace::main_storage;
return min_pair->first;
}
private:
costs_t costs;
};
using search_cost_t = SearchCost<cost_t>;
constexpr auto search_internal_id = search_cost_t::SearchPlace::internal_id;
constexpr auto search_label_index = search_cost_t::SearchPlace::label_index;
constexpr auto search_property_index =
search_cost_t::SearchPlace::property_index;
constexpr auto search_main_storage = search_cost_t::SearchPlace::main_storage;
}
class CypherStateMachine
{
public:
void init_cost(const std::string &entity)
{
entity_search::search_cost_t search_cost;
_search_costs.emplace(entity, search_cost);
}
void search_cost(const std::string &entity,
entity_search::search_cost_t::SearchPlace search_place,
entity_search::cost_t cost)
{
if (_search_costs.find(entity) != _search_costs.end()) {
entity_search::search_cost_t search_cost;
_search_costs.emplace(entity, std::move(search_cost));
}
_search_costs[entity].set(search_place, cost);
}
entity_search::cost_t
search_cost(const std::string &entity,
entity_search::search_cost_t::SearchPlace search_place) const
{
return _search_costs.at(entity).get(search_place);
}
entity_search::search_cost_t::SearchPlace
min(const std::string &entity) const
{
if (_search_costs.find(entity) == _search_costs.end())
return entity_search::search_cost_t::SearchPlace::main_storage;
return _search_costs.at(entity).min();
}
private:
std::map<std::string, entity_search::search_cost_t> _search_costs;
};

View File

@@ -0,0 +1,9 @@
#pragma once
#include "query_engine/code_generator/handlers/create.hpp"
#include "query_engine/code_generator/handlers/match.hpp"
#include "query_engine/code_generator/handlers/return.hpp"
#include "query_engine/code_generator/handlers/set.hpp"
#include "query_engine/code_generator/handlers/delete.hpp"
#include "query_engine/code_generator/handlers/transaction_begin.hpp"
#include "query_engine/code_generator/handlers/transaction_commit.hpp"

View File

@@ -0,0 +1,85 @@
#pragma once
#include "query_engine/code_generator/handlers/includes.hpp"
auto create_query_action =
[](CypherStateData &cypher_data,
const QueryActionData &action_data) -> std::string {
std::string code = "";
for (auto const &kv : action_data.actions) {
if (kv.second == ClauseAction::CreateNode) {
// create node
auto &name = kv.first;
code += code_line(code::create_vertex, name);
// update properties
code += update_properties(action_data, name);
// update labels
auto entity_data = action_data.get_entity_property(name);
for (auto &label : entity_data.tags) {
code += code_line(code::create_label, label);
code += code_line(code::add_label, name, label);
}
// mark node as created
cypher_data.node_created(name);
}
if (kv.second == ClauseAction::CreateRelationship) {
// create relationship
auto name = kv.first;
code += code_line(code::create_edge, name);
// update properties
code += update_properties(action_data, name);
// update tag
auto entity_data = action_data.get_entity_property(name);
for (auto &tag : entity_data.tags) {
code += code_line(code::find_type, tag);
code += code_line(code::set_type, name, tag);
}
// find start and end node
auto &relationships_data = action_data.relationship_data;
if (relationships_data.find(name) == relationships_data.end())
throw CodeGenerationError("Unable to find data for: " + name);
auto &relationship_data = relationships_data.at(name);
auto left_node = relationship_data.nodes.first;
auto right_node = relationship_data.nodes.second;
// TODO: If node isn't already matched or created it has to be
// created here. It is not possible for now.
if (cypher_data.status(left_node) != EntityStatus::Matched) {
throw SemanticError("Create Relationship: node " + left_node +
" can't be found");
}
if (cypher_data.status(right_node) != EntityStatus::Matched) {
throw SemanticError("Create Relationship: node " + right_node +
" can't be found");
}
// define direction
if (relationship_data.direction == Direction::Right) {
code += code_line(code::node_out, left_node, name);
code += code_line(code::node_in, right_node, name);
code += code_line(code::edge_from, name, left_node);
code += code_line(code::edge_to, name, right_node);
} else if (relationship_data.direction == Direction::Left) {
code += code_line(code::node_out, right_node, name);
code += code_line(code::node_in, left_node, name);
code += code_line(code::edge_from, name, right_node);
code += code_line(code::edge_to, name, left_node);
}
// mark relationship as created
cypher_data.relationship_created(name);
}
}
return code;
};

View File

@@ -0,0 +1,22 @@
#pragma once
#include "query_engine/code_generator/handlers/includes.hpp"
auto delete_query_action =
[](CypherStateData &cypher_data,
const QueryActionData &action_data) -> std::string {
std::string code = "";
for (auto const &kv : action_data.actions) {
auto entity = kv.first;
if (kv.second == ClauseAction::DeleteNode) {
code += code_line("// DELETE Node({})", entity);
}
if (kv.second == ClauseAction::DeleteRelationship) {
code += code_line("// DELETE Relationship({})", entity);
}
}
return code;
};

View File

@@ -0,0 +1,36 @@
#pragma once
#include <iostream>
#include <map>
#include <string>
#include <utility>
#include <vector>
#include "query_engine/util.hpp"
#include "query_engine/code_generator/cypher_state.hpp"
#include "query_engine/code_generator/query_action_data.hpp"
#include "query_engine/traverser/code.hpp"
#include "query_engine/exceptions/errors.hpp"
using ParameterIndexKey::Type::InternalId;
using Direction = RelationshipData::Direction;
namespace
{
auto update_properties(const QueryActionData &action_data,
const std::string &name)
{
std::string code = "";
auto entity_data = action_data.get_entity_property(name);
for (auto &property : entity_data.properties) {
auto index =
action_data.parameter_index.at(ParameterIndexKey(name, property));
code += code_line(code::set_property, name, property, index);
}
return code;
}
}

View File

@@ -0,0 +1,63 @@
#pragma once
#include "query_engine/code_generator/handlers/includes.hpp"
namespace
{
bool already_matched(CypherStateData &cypher_data, const std::string &name,
EntityType type)
{
if (cypher_data.type(name) == type &&
cypher_data.status(name) == EntityStatus::Matched)
return true;
else
return false;
}
auto fetch_internal_index(const QueryActionData &action_data,
const std::string &name)
{
return action_data.parameter_index.at(ParameterIndexKey(InternalId, name));
}
}
auto match_query_action =
[](CypherStateData &cypher_data,
const QueryActionData &action_data) -> std::string {
std::string code = "";
for (auto const &kv : action_data.actions) {
// TODO: the same code REFACTOR!
// find node
if (kv.second == ClauseAction::MatchNode) {
auto name = kv.first;
if (already_matched(cypher_data, name, EntityType::Node)) continue;
cypher_data.node_matched(name);
auto place = action_data.csm.min(kv.first);
if (place == entity_search::search_internal_id) {
auto index = fetch_internal_index(action_data, name);
code +=
code_line(code::match_vertex_by_id, name, index);
}
}
// find relationship
if (kv.second == ClauseAction::MatchRelationship) {
auto name = kv.first;
if (already_matched(cypher_data, name, EntityType::Relationship))
continue;
cypher_data.relationship_matched(name);
auto place = action_data.csm.min(kv.first);
if (place == entity_search::search_internal_id) {
auto index = fetch_internal_index(action_data, name);
code += code_line(code::match_edge_by_id, name, index);
}
}
}
return code;
};

View File

@@ -0,0 +1,31 @@
#pragma once
#include "query_engine/code_generator/handlers/includes.hpp"
auto return_query_action =
[](CypherStateData &cypher_data,
const QueryActionData &action_data) -> std::string {
std::string code = "";
const auto &elements = action_data.return_elements;
code += code_line("// number of elements {}", elements.size());
// TODO: call bolt serialization
for (const auto& element : elements) {
auto &entity = element.entity;
if (!cypher_data.exist(entity)) {
throw SemanticError(
fmt::format("{} couldn't be found (RETURN clause).", entity));
}
if (element.is_entity_only()) {
code += code_line(code::write_entity, entity);
} else if (element.is_projection()) {
code += code_line("// TODO: implement projection");
// auto &property = element.property;
// code += code_line(code::print_property, entity, property);
}
}
return code;
};

View File

@@ -0,0 +1,27 @@
#pragma once
#include "query_engine/code_generator/handlers/includes.hpp"
auto set_query_action = [](CypherStateData &cypher_data,
const QueryActionData &action_data) -> std::string {
std::string code = "";
for (auto const &kv : action_data.actions) {
auto name = kv.first;
if (kv.second == ClauseAction::UpdateNode &&
cypher_data.status(name) == EntityStatus::Matched &&
cypher_data.type(name) == EntityType::Node) {
code += update_properties(action_data, name);
}
if (kv.second == ClauseAction::UpdateRelationship &&
cypher_data.status(name) == EntityStatus::Matched &&
cypher_data.type(name) == EntityType::Relationship) {
code += update_properties(action_data, name);
}
}
return code;
};

View File

@@ -0,0 +1,8 @@
#pragma once
#include "query_engine/code_generator/handlers/includes.hpp"
auto transaction_begin_action = [](CypherStateData &,
const QueryActionData &) -> std::string {
return code_line(code::transaction_begin);
};

View File

@@ -0,0 +1,9 @@
#pragma once
#include "query_engine/code_generator/handlers/includes.hpp"
auto transaction_commit_action = [](CypherStateData &,
const QueryActionData &) -> std::string {
return code_line(code::transaction_commit) +
code_line(code::return_true);
};

View File

@@ -0,0 +1,16 @@
#pragma once
#include <cstdint>
// any entity (node or relationship) inside cypher query has an action
// that is associated with that entity
enum class QueryAction : uint32_t
{
TransactionBegin,
Create,
Match,
Set,
Return,
Delete,
TransactionCommit
};

View File

@@ -0,0 +1,161 @@
#pragma once
#include <iostream>
#include <map>
#include <vector>
#include "query_engine/exceptions/exceptions.hpp"
#include "query_engine/code_generator/clause_action.hpp"
#include "query_engine/code_generator/entity_search.hpp"
#include "storage/model/properties/all.hpp"
#include "utils/assert.hpp"
#include "utils/underlying_cast.hpp"
// used for storing data related to an entity (node or relationship)
// data can be:
// * tags: labels or type
// * props: property name, property value
struct EntityData
{
std::vector<std::string> tags;
std::vector<std::string> properties;
void add_tag(const std::string &tag) { tags.push_back(tag); }
void add_property(const std::string &property)
{
properties.push_back(property);
}
};
// used for storing indices of parameters (parameters are stripped before
// compiling process into the array), so somehow the compiler has to know
// how to find appropriate parameter during the compile process
//
// parameter index key can be related to:
// * internal_id and entity_name, e.g.:
// ID(n)=35445 -> ID(n)=2 -> index[PropertyIndexKey(Type::InternalId, n)] =
// 2
// * entity_name and entity_property, e.g.:
// n.name = "test" -> n.name = 3 -> index[PropertyIndexKey(entity_name,
// entity_property)] = 3
struct ParameterIndexKey
{
enum class Type : uint8_t
{
InternalId,
Projection
};
ParameterIndexKey(Type type, const std::string &entity_name)
: type(type), entity_name(entity_name)
{
}
ParameterIndexKey(const std::string &entity_name,
const std::string &entity_property)
: type(Type::Projection), entity_name(entity_name),
entity_property(entity_property)
{
}
const Type type;
const std::string entity_name;
const std::string entity_property;
bool operator<(const ParameterIndexKey &rhs) const
{
runtime_assert(type == rhs.type,
"ParameterIndexKey types should be the same");
if (type == Type::InternalId) return entity_name < rhs.entity_name;
if (entity_name == rhs.entity_name)
return entity_property < rhs.entity_property;
return entity_name < rhs.entity_name;
}
};
struct RelationshipData
{
enum class Direction
{
Left,
Right
};
using nodes_t = std::pair<std::string, std::string>;
RelationshipData(nodes_t nodes, Direction direction)
: nodes(nodes), direction(direction)
{
}
std::pair<std::string, std::string> nodes;
Direction direction;
};
struct ReturnElement
{
ReturnElement(const std::string& entity) : entity(entity) {}
ReturnElement(const std::string& entity, const std::string& property) :
entity(entity), property(property) {};
std::string entity;
std::string property;
bool has_entity() const { return !entity.empty(); }
bool has_property() const { return !property.empty(); }
bool is_entity_only() const { return has_entity() && !has_property(); }
bool is_projection() const { return has_entity() && has_property(); }
};
struct QueryActionData
{
std::map<ParameterIndexKey, uint64_t> parameter_index;
std::map<std::string, ClauseAction> actions;
std::map<std::string, EntityData> entity_data;
std::map<std::string, RelationshipData> relationship_data;
std::vector<ReturnElement> return_elements;
CypherStateMachine csm;
QueryActionData() = default;
QueryActionData(QueryActionData &&other) = default;
void create_entity(const std::string &entity)
{
if (entity_data.find(entity) == entity_data.end())
entity_data.emplace(entity, EntityData());
}
void add_entity_tag(const std::string &entity, const std::string &tag)
{
create_entity(entity);
entity_data.at(entity).add_tag(tag);
}
void add_entitiy_property(const std::string &entity,
const std::string &property)
{
create_entity(entity);
entity_data.at(entity).add_property(property);
}
// TODO: refactor name
auto get_entity_property(const std::string& entity) const
{
if (entity_data.find(entity) == entity_data.end())
throw CppGeneratorException("Entity " + entity + " doesn't exist");
return entity_data.at(entity);
}
void print() const
{
for (auto const &action : actions) {
std::cout << action.first << " " << underlying_cast(action.second)
<< std::endl;
}
}
};

View File

@@ -0,0 +1,19 @@
#pragma once
#include <vector>
#include "utils/assert.hpp"
template <class T>
class Vector : public std::vector<T>
{
public:
using pair = std::pair<T, T>;
pair last_two()
{
runtime_assert(this->size() > 1, "Array size shoud be bigger than 1");
return std::make_pair(*(this->end() - 1), *(this->end() - 2));
}
};

View File

@@ -0,0 +1,19 @@
#pragma once
#include "utils/exceptions/basic_exception.hpp"
// TODO: optimaze exceptions in respect to cypher/errors.hpp
class SemanticError : public BasicException
{
public:
SemanticError(const std::string& what) :
BasicException("Semantic error: " + what) {}
};
class CodeGenerationError : public BasicException
{
public:
CodeGenerationError(const std::string& what) :
BasicException("Code Generation error: " + what) {}
};

View File

@@ -0,0 +1,13 @@
#pragma once
#include "utils/exceptions/basic_exception.hpp"
class QueryEngineException : public BasicException
{
using BasicException::BasicException;
};
class CppGeneratorException : public BasicException
{
using BasicException::BasicException;
};

View File

@@ -0,0 +1,197 @@
#pragma once
#include "database/db.hpp"
#include "query_engine/query_stripper.hpp"
#include "query_engine/util.hpp"
#include "storage/model/properties/property.hpp"
#include "utils/command_line/arguments.hpp"
auto load_queries(Db &db)
{
std::map<uint64_t, std::function<bool(const properties_t &)>> queries;
// CREATE (n {prop: 0}) RETURN n)
auto create_node = [&db](const properties_t &args) {
auto &t = db.tx_engine.begin();
auto vertex_accessor = db.graph.vertices.insert(t);
vertex_accessor.property("prop", args[0]);
t.commit();
return true;
};
queries[11597417457737499503u] = create_node;
auto create_labeled_and_named_node = [&db](const properties_t &args) {
auto &t = db.tx_engine.begin();
auto vertex_accessor = db.graph.vertices.insert(t);
vertex_accessor.property("name", args[0]);
auto &label = db.graph.label_store.find_or_create("LABEL");
vertex_accessor.add_label(label);
cout_properties(vertex_accessor.properties());
t.commit();
return true;
};
auto create_account = [&db](const properties_t &args) {
auto &t = db.tx_engine.begin();
auto vertex_accessor = db.graph.vertices.insert(t);
vertex_accessor.property("id", args[0]);
vertex_accessor.property("name", args[1]);
vertex_accessor.property("country", args[2]);
vertex_accessor.property("created_at", args[3]);
auto &label = db.graph.label_store.find_or_create("ACCOUNT");
vertex_accessor.add_label(label);
cout_properties(vertex_accessor.properties());
t.commit();
return true;
};
auto find_node_by_internal_id = [&db](const properties_t &args) {
auto &t = db.tx_engine.begin();
auto id = static_cast<Int32 &>(*args[0]);
auto vertex_accessor = db.graph.vertices.find(t, Id(id.value));
if (!vertex_accessor) {
cout << "vertex doesn't exist" << endl;
t.commit();
return false;
}
cout_properties(vertex_accessor.properties());
cout << "LABELS:" << endl;
for (auto label_ref : vertex_accessor.labels()) {
cout << label_ref.get() << endl;
}
t.commit();
return true;
};
auto create_edge = [&db](const properties_t &args) {
auto &t = db.tx_engine.begin();
auto v1 = db.graph.vertices.find(t, args[0]->as<Int32>().value);
if (!v1) return t.commit(), false;
auto v2 = db.graph.vertices.find(t, args[1]->as<Int32>().value);
if (!v2) return t.commit(), false;
auto edge_accessor = db.graph.edges.insert(t, v1.vlist, v2.vlist);
v1.vlist->update(t)->data.out.add(edge_accessor.vlist);
v2.vlist->update(t)->data.in.add(edge_accessor.vlist);
auto &edge_type = db.graph.edge_type_store.find_or_create("IS");
edge_accessor.edge_type(edge_type);
t.commit();
cout << edge_accessor.edge_type() << endl;
cout_properties(edge_accessor.properties());
return true;
};
auto find_edge_by_internal_id = [&db](const properties_t &args) {
auto &t = db.tx_engine.begin();
auto edge_accessor = db.graph.edges.find(t, args[0]->as<Int32>().value);
if (!edge_accessor) return t.commit(), false;
// print edge type and properties
cout << "EDGE_TYPE: " << edge_accessor.edge_type() << endl;
auto from = edge_accessor.from();
cout << "FROM:" << endl;
cout_properties(from->find(t)->data.props);
auto to = edge_accessor.to();
cout << "TO:" << endl;
cout_properties(to->find(t)->data.props);
t.commit();
return true;
};
auto update_node = [&db](const properties_t &args) {
auto &t = db.tx_engine.begin();
auto v = db.graph.vertices.find(t, args[0]->as<Int32>().value);
if (!v) return t.commit(), false;
v.property("name", args[1]);
cout_properties(v.properties());
t.commit();
return true;
};
// MATCH (n1), (n2) WHERE ID(n1)=0 AND ID(n2)=1 CREATE (n1)<-[r:IS {age: 25,
// weight: 70}]-(n2) RETURN r
auto create_edge_v2 = [&db](const properties_t &args) {
auto &t = db.tx_engine.begin();
auto n1 = db.graph.vertices.find(t, args[0]->as<Int64>().value);
if (!n1) return t.commit(), false;
auto n2 = db.graph.vertices.find(t, args[1]->as<Int64>().value);
if (!n2) return t.commit(), false;
auto r = db.graph.edges.insert(t, n2.vlist, n1.vlist);
r.property("age", args[2]);
r.property("weight", args[3]);
auto &IS = db.graph.edge_type_store.find_or_create("IS");
r.edge_type(IS);
n2.vlist->update(t)->data.out.add(r.vlist);
n1.vlist->update(t)->data.in.add(r.vlist);
t.commit();
return true;
};
queries[15648836733456301916u] = create_edge_v2;
// MATCH (n) RETURN n
auto match_all_nodes = [&db](const properties_t &args) {
auto &t = db.tx_engine.begin();
auto vertices_accessor = db.graph.vertices.access();
for (auto &it : vertices_accessor) {
auto vertex = it.second.find(t);
if (vertex == nullptr) continue;
cout_properties(vertex->data.props);
}
// TODO
// db.graph.vertices.filter().all(t, handler);
t.commit();
return true;
};
queries[15284086425088081497u] = match_all_nodes;
// MATCH (n:LABEL) RETURN n
auto find_by_label = [&db](const properties_t &args) {
auto &t = db.tx_engine.begin();
auto &label = db.graph.label_store.find_or_create("LABEL");
auto &index_record_collection =
db.graph.vertices.find_label_index(label);
auto accessor = index_record_collection.access();
cout << "VERTICES" << endl;
for (auto &v : accessor) {
cout << v.record->data.props.at("name").as<String>().value << endl;
}
// TODO
// db.graph.vertices.fileter("LABEL").all(t, handler);
return true;
};
queries[4857652843629217005u] = find_by_label;
queries[10597108978382323595u] = create_account;
queries[5397556489557792025u] = create_labeled_and_named_node;
queries[7939106225150551899u] = create_edge;
queries[6579425155585886196u] = create_edge;
queries[11198568396549106428u] = find_node_by_internal_id;
queries[8320600413058284114u] = find_edge_by_internal_id;
queries[6813335159006269041u] = update_node;
return queries;
}

View File

@@ -0,0 +1,16 @@
#pragma once
#include "communication/communication.hpp"
#include "database/db.hpp"
#include "query_engine/query_stripped.hpp"
class ICodeCPU
{
public:
virtual bool run(Db &db, code_args_t &args,
communication::OutputStream &stream) = 0;
virtual ~ICodeCPU() {}
};
using produce_t = ICodeCPU *(*)();
using destruct_t = void (*)(ICodeCPU *);

View File

@@ -0,0 +1,23 @@
#pragma once
#include "query_engine/i_code_cpu.hpp"
#include "dc/dynamic_lib.hpp"
namespace
{
class MemgraphDynamicLib
{
public:
const static std::string produce_name;
const static std::string destruct_name;
using produce = produce_t;
using destruct = destruct_t;
using lib_object = ICodeCPU;
};
const std::string MemgraphDynamicLib::produce_name = "produce";
const std::string MemgraphDynamicLib::destruct_name = "destruct";
using CodeLib = DynamicLib<MemgraphDynamicLib>;
}

Some files were not shown because too many files have changed in this diff Show More