From 12880ba99096e5742bc5e52db724a212ccf8aadb Mon Sep 17 00:00:00 2001 From: Marko Budiselic Date: Mon, 29 Aug 2016 01:01:42 +0100 Subject: [PATCH 1/3] work in progress engine <-> barrier integration --- CMakeLists.txt | 17 +-- include/barrier/barrier.hpp | 38 +++--- include/barrier/trans.hpp | 7 +- .../bolt/v1/serialization/bolt_serializer.hpp | 3 + .../bolt/v1/serialization/record_stream.hpp | 2 + include/query_engine/code_compiler.hpp | 1 + include/query_engine/code_generator.hpp | 3 +- include/query_engine/i_code_cpu.hpp | 10 +- include/query_engine/program_executor.hpp | 8 +- poc/astar.cpp | 1 + poc/profile.cpp | 1 + src/barrier/barrier.cpp | 109 +++++++++++++----- .../bolt/v1/serialization/bolt_serializer.cpp | 7 ++ src/communication/bolt/v1/states/executor.cpp | 3 + .../template/template_code_cpu.cpp | 17 ++- tests/integration/queries.cpp | 1 + tests/manual/queries.cpp | 1 + 17 files changed, 160 insertions(+), 69 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 01bfbe64f..ebde7a3db 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -144,6 +144,7 @@ endforeach() # ----------------------------------------------------------------------------- # COPY header files required by query engine (query compiler) # TODO: somehow automate (in destination dir should be only required include files) +FILE(COPY ${include_dir}/barrier/barrier.hpp DESTINATION ${build_include_dir}/barrier) FILE(COPY ${include_dir}/database/db.hpp DESTINATION ${build_include_dir}/database) FILE(COPY ${include_dir}/database/db_transaction.hpp DESTINATION ${build_include_dir}/database) FILE(COPY ${include_dir}/database/db_accessor.hpp DESTINATION ${build_include_dir}/database) @@ -455,7 +456,6 @@ set(memgraph_src_files ${src_dir}/storage/record_accessor.cpp ) - # STATIC library used by memgraph executables add_library(memgraph STATIC ${memgraph_src_files}) @@ -463,6 +463,10 @@ add_library(memgraph STATIC ${memgraph_src_files}) add_library(memgraph_pic STATIC ${memgraph_src_files}) set_property(TARGET memgraph_pic PROPERTY POSITION_INDEPENDENT_CODE TRUE) +add_library(barrier STATIC ${memgraph_src_files}) + +add_library(barrier_pic STATIC ${memgraph_src_files}) +set_property(TARGET barrier_pic PROPERTY POSITION_INDEPENDENT_CODE TRUE) # tests if (TESTS) @@ -498,7 +502,7 @@ set(MEMGRAPH_BUILD_NAME # memgraph main executable if (MEMGRAPH) add_executable(${MEMGRAPH_BUILD_NAME} ${src_dir}/memgraph_bolt.cpp) - target_link_libraries(${MEMGRAPH_BUILD_NAME} memgraph) + target_link_libraries(${MEMGRAPH_BUILD_NAME} barrier) target_link_libraries(${MEMGRAPH_BUILD_NAME} Threads::Threads) target_link_libraries(${MEMGRAPH_BUILD_NAME} cypher_lib) if (UNIX) @@ -508,12 +512,3 @@ if (MEMGRAPH) target_link_libraries(${MEMGRAPH_BUILD_NAME} dl) endif (UNIX) endif() - -# # memgraph executable HTTP TODO: DEPRICATED -# add_executable(memgraph_http src/memgraph.cpp) -# add_dependencies(memgraph_http cypher_lib) -# target_link_libraries(memgraph_http Threads::Threads) -# target_link_libraries(memgraph_http pcre) -# target_link_libraries(memgraph_http ${libuv_static_lib}) -# target_link_libraries(memgraph_http ${r3_static_lib}) -# target_link_libraries(memgraph_http ${http_parser_static_lib}) diff --git a/include/barrier/barrier.hpp b/include/barrier/barrier.hpp index d8c088db4..e201b9d61 100644 --- a/include/barrier/barrier.hpp +++ b/include/barrier/barrier.hpp @@ -38,7 +38,7 @@ class EdgePropertyType; // BOLT template -class BoltSerializer; +class RecordStream; // ************ Here should be forward declarations of Unsized barrier classes // COMMON @@ -413,43 +413,41 @@ public: }; template -class BoltSerializer : private Sized<8, 8> +class RecordStream : private Sized<8, 8> { public: template - BoltSerializer(T &&d); + RecordStream(T &&d); - BoltSerializer(const BoltSerializer &other) = default; - BoltSerializer(BoltSerializer &&other) = default; - ~BoltSerializer(); + RecordStream(const RecordStream &other) = default; + RecordStream(RecordStream &&other) = default; + ~RecordStream(); - BoltSerializer &operator=(const BoltSerializer &other) = default; - BoltSerializer &operator=(BoltSerializer &&other) = default; + RecordStream &operator=(const RecordStream &other) = default; + RecordStream &operator=(RecordStream &&other) = default; void write(const VertexAccessor &vertex); - void write(const EdgeAccessor &edge); - void write(const Property &prop); - void write_null(); - void write(const Bool &prop); - void write(const Float &prop); - void write(const Double &prop); - void write(const Int32 &prop); - void write(const Int64 &prop); - void write(const std::string &value); - void write(const String &prop); - template - void handle(const T &prop); + void write_success(); + void write_success_empty(); + void write_ignored(); + void write_fields(const std::vector &fields); + void write_field(const std::string& field); + void write_list_header(size_t size); + void write_record(); + void write_meta(const std::string& type); + void send(); + void chunk(); }; // ************ Here should be definitions of Unsized barrier classes diff --git a/include/barrier/trans.hpp b/include/barrier/trans.hpp index b4fc8a567..af7c8ad76 100644 --- a/include/barrier/trans.hpp +++ b/include/barrier/trans.hpp @@ -1,8 +1,11 @@ #pragma once + #include "barrier/barrier.hpp" // This is the place for imports from memgraph .hpp #include "communication/bolt/v1/serialization/bolt_serializer.hpp" +#include "communication/bolt/v1/serialization/record_stream.hpp" +#include "io/network/socket.hpp" #include "database/db.hpp" #include "database/db_accessor.hpp" #include "storage/edge_type/edge_type.hpp" @@ -165,7 +168,7 @@ TRANSFORM_REF_TEMPLATED(VertexIndex, VertexIndexBase); template TRANSFORM_REF_TEMPLATED(EdgeIndex, EdgeIndexBase); template -TRANSFORM_REF_TEMPLATED(BoltSerializer, ::bolt::BoltSerializer); +TRANSFORM_REF_TEMPLATED(RecordStream, ::bolt::RecordStream); template TRANSFORM_REF_TEMPLATED( @@ -209,7 +212,7 @@ TRANSFORM_VALUE_ONE_RAW(EdgePropertyType, ::EdgePropertyFamily::PropertyType::PropertyTypeKey) template -TRANSFORM_VALUE_ONE_RAW(BoltSerializer, ::bolt::BoltSerializer) +TRANSFORM_VALUE_ONE_RAW(RecordStream, ::bolt::RecordStream) // ********************* SPECIAL CONSTRUCTORS #define VertexPropertyType_constructor(x) \ diff --git a/include/communication/bolt/v1/serialization/bolt_serializer.hpp b/include/communication/bolt/v1/serialization/bolt_serializer.hpp index e34323cdd..73578a121 100644 --- a/include/communication/bolt/v1/serialization/bolt_serializer.hpp +++ b/include/communication/bolt/v1/serialization/bolt_serializer.hpp @@ -8,6 +8,9 @@ #include "storage/model/properties/all.hpp" #include "storage/model/properties/properties.hpp" +#include "storage/label/label.hpp" +#include "storage/edge_type/edge_type.hpp" +#include "storage/vertex_record.hpp" namespace bolt { diff --git a/include/communication/bolt/v1/serialization/record_stream.hpp b/include/communication/bolt/v1/serialization/record_stream.hpp index 7c5f1b46b..93bd7c4e2 100644 --- a/include/communication/bolt/v1/serialization/record_stream.hpp +++ b/include/communication/bolt/v1/serialization/record_stream.hpp @@ -22,6 +22,8 @@ public: logger = logging::log->logger("Record Stream"); } + ~RecordStream() = default; + // TODO: create apstract methods that are not bolt specific --------------- void write_success() { diff --git a/include/query_engine/code_compiler.hpp b/include/query_engine/code_compiler.hpp index 646f00ed8..00ce66ae0 100644 --- a/include/query_engine/code_compiler.hpp +++ b/include/query_engine/code_compiler.hpp @@ -52,6 +52,7 @@ public: "-I../../libs/fmt", "-L./ -L../", "-lmemgraph_pic", + "-lbarrier_pic", "-shared -fPIC" // shared library flags ); diff --git a/include/query_engine/code_generator.hpp b/include/query_engine/code_generator.hpp index 8230b544e..964c824bb 100644 --- a/include/query_engine/code_generator.hpp +++ b/include/query_engine/code_generator.hpp @@ -55,7 +55,8 @@ public: template_file, {{"class_name", "CodeCPU"}, {"stripped_hash", std::to_string(stripped_hash)}, {"query", query}, - {"stream", type_name().to_string()}, + // {"stream", type_name().to_string()}, + {"stream", "RecordStream"}, {"code", cpp_traverser.code}}); // logger.trace("generated code: {}", generated); diff --git a/include/query_engine/i_code_cpu.hpp b/include/query_engine/i_code_cpu.hpp index cd4e92bc1..fd7d7d02f 100644 --- a/include/query_engine/i_code_cpu.hpp +++ b/include/query_engine/i_code_cpu.hpp @@ -1,15 +1,19 @@ #pragma once #include "communication/communication.hpp" -#include "database/db.hpp" -#include "database/db_accessor.hpp" #include "query_engine/query_stripped.hpp" +// #include "database/db.hpp" +// #include "database/db_accessor.hpp" + +// BARRIER! +#include "barrier/barrier.hpp" + template class ICodeCPU { public: - virtual bool run(Db &db, code_args_t &args, + virtual bool run(barrier::Db &db, code_args_t &args, Stream &stream) = 0; virtual ~ICodeCPU() {} }; diff --git a/include/query_engine/program_executor.hpp b/include/query_engine/program_executor.hpp index c74a2720d..3a5ed6614 100644 --- a/include/query_engine/program_executor.hpp +++ b/include/query_engine/program_executor.hpp @@ -11,6 +11,12 @@ // execution // postprocess the results +// BARRIER! +namespace barrier +{ + Db& trans(::Db& ref); +} + template class ProgramExecutor { @@ -22,7 +28,7 @@ public: { try { // TODO: return result of query/code exection - return program.code->run(db, program.stripped.arguments, stream); + return program.code->run(barrier::trans(db), program.stripped.arguments, stream); } catch (...) { // TODO: return more information about the error throw QueryEngineException("code execution error"); diff --git a/poc/astar.cpp b/poc/astar.cpp index 0655294c3..433e39431 100644 --- a/poc/astar.cpp +++ b/poc/astar.cpp @@ -25,6 +25,7 @@ #include "storage/vertices.cpp" #include "storage/vertices.hpp" #include "utils/command_line/arguments.hpp" +#include "communication/bolt/v1/serialization/bolt_serializer.hpp" const int max_score = 1000000; diff --git a/poc/profile.cpp b/poc/profile.cpp index 4faecf66f..f8f320460 100644 --- a/poc/profile.cpp +++ b/poc/profile.cpp @@ -12,6 +12,7 @@ #include #include "import/csv_import.hpp" #include "utils/command_line/arguments.hpp" +#include "communication/bolt/v1/serialization/bolt_serializer.hpp" using namespace std; diff --git a/src/barrier/barrier.cpp b/src/barrier/barrier.cpp index dfb3e9513..7031230bd 100644 --- a/src/barrier/barrier.cpp +++ b/src/barrier/barrier.cpp @@ -499,83 +499,140 @@ OptionPtr> EdgePropertyFamily::index() // ************************* BOLT SERIALIZER template -BoltSerializer::~BoltSerializer() +RecordStream::~RecordStream() { - THIS->~BoltSerializer(); + // TODO: solve this + // THIS->~RecordStream(); } template -void BoltSerializer::write(const VertexAccessor &vertex) +void RecordStream::write(const VertexAccessor &vertex) { HALF_CALL(write(trans(vertex))); } template -void BoltSerializer::write(const EdgeAccessor &edge) +void RecordStream::write(const EdgeAccessor &edge) { HALF_CALL(write(trans(edge))); } template -void BoltSerializer::write(const Property &prop) +void RecordStream::write(const Property &prop) +{ + HALF_CALL(write(prop)); +} + +// template +// void RecordStream::write_null() +// { +// HALF_CALL(write_null()); +// } + +template +void RecordStream::write(const Bool &prop) { HALF_CALL(write(prop)); } template -void BoltSerializer::write_null() -{ - HALF_CALL(write_null()); -} - -template -void BoltSerializer::write(const Bool &prop) +void RecordStream::write(const Float &prop) { HALF_CALL(write(prop)); } template -void BoltSerializer::write(const Float &prop) +void RecordStream::write(const Double &prop) { HALF_CALL(write(prop)); } template -void BoltSerializer::write(const Double &prop) +void RecordStream::write(const Int32 &prop) { HALF_CALL(write(prop)); } template -void BoltSerializer::write(const Int32 &prop) +void RecordStream::write(const Int64 &prop) { HALF_CALL(write(prop)); } template -void BoltSerializer::write(const Int64 &prop) -{ - HALF_CALL(write(prop)); -} - -template -void BoltSerializer::write(const std::string &value) +void RecordStream::write(const std::string &value) { HALF_CALL(write(value)); } template -void BoltSerializer::write(const String &prop) +void RecordStream::write(const String &prop) { HALF_CALL(write(prop)); } template -template -void BoltSerializer::handle(const T &prop) +void RecordStream::write_success() { - HALF_CALL(template handle(prop)); + HALF_CALL(write_success()); } + +template +void RecordStream::write_success_empty() +{ + HALF_CALL(write_success_empty()); +} + +template +void RecordStream::write_ignored() +{ + HALF_CALL(write_ignored()); +} + +template +void RecordStream::write_fields(const std::vector &fields) +{ + HALF_CALL(write_fields(fields)); +} + +template +void RecordStream::write_field(const std::string& field) +{ + HALF_CALL(write_field(field)); +} + +template +void RecordStream::write_list_header(size_t size) +{ + HALF_CALL(write_list_header(size)); +} + +template +void RecordStream::write_record() +{ + HALF_CALL(write_record()); +} + +template +void RecordStream::write_meta(const std::string& type) +{ + HALF_CALL(write_meta(type)); +} + +template +void RecordStream::send() +{ + HALF_CALL(send()); +} + +template +void RecordStream::chunk() +{ + HALF_CALL(chunk()); +} + +template class RecordStream; + } // **************************** ERROR EXAMPLES ****************************** // diff --git a/src/communication/bolt/v1/serialization/bolt_serializer.cpp b/src/communication/bolt/v1/serialization/bolt_serializer.cpp index 4ff4c7f56..b45445897 100644 --- a/src/communication/bolt/v1/serialization/bolt_serializer.cpp +++ b/src/communication/bolt/v1/serialization/bolt_serializer.cpp @@ -1,5 +1,9 @@ #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 "io/network/socket.hpp" #include "storage/edge_x_vertex.hpp" template @@ -28,3 +32,6 @@ void bolt::BoltSerializer::write(const EdgeAccessor &edge) write(*prop.second); } } + +template class bolt::BoltSerializer>>>>; diff --git a/src/communication/bolt/v1/states/executor.cpp b/src/communication/bolt/v1/states/executor.cpp index 6d280f37e..4a8f5e8d4 100644 --- a/src/communication/bolt/v1/states/executor.cpp +++ b/src/communication/bolt/v1/states/executor.cpp @@ -1,6 +1,9 @@ #include "communication/bolt/v1/states/executor.hpp" #include "communication/bolt/v1/messaging/codes.hpp" +// BARRIER! TODO: ATTENTION: HACK!!!!! +#include "barrier/barrier.cpp" + namespace bolt { diff --git a/src/query_engine/template/template_code_cpu.cpp b/src/query_engine/template/template_code_cpu.cpp index 667687d6e..74b1e30b8 100644 --- a/src/query_engine/template/template_code_cpu.cpp +++ b/src/query_engine/template/template_code_cpu.cpp @@ -10,6 +10,10 @@ using std::endl; // query: {{query}} +// BARRIER! +namespace barrier +{ + class {{class_name}} : public ICodeCPU<{{stream}}> { public: @@ -23,13 +27,16 @@ public: ~{{class_name}}() {} }; - -extern "C" ICodeCPU<{{stream}}>* produce() -{ - return new {{class_name}}(); } -extern "C" void destruct(ICodeCPU<{{stream}}>* p) + +extern "C" ICodeCPU* produce() +{ + // BARRIER! + return new barrier::{{class_name}}(); +} + +extern "C" void destruct(ICodeCPU* p) { delete p; } diff --git a/tests/integration/queries.cpp b/tests/integration/queries.cpp index ad649fd75..f8d94f8c9 100644 --- a/tests/integration/queries.cpp +++ b/tests/integration/queries.cpp @@ -4,6 +4,7 @@ #include "database/db.hpp" #include "query_engine/query_stripper.hpp" +#include "communication/bolt/v1/serialization/bolt_serializer.hpp" // #include "storage/edges.cpp" // #include "storage/edges.hpp" // #include "storage/vertices.cpp" diff --git a/tests/manual/queries.cpp b/tests/manual/queries.cpp index 8d46e6e07..3902904d8 100644 --- a/tests/manual/queries.cpp +++ b/tests/manual/queries.cpp @@ -10,6 +10,7 @@ #include "storage/edges.hpp" #include "storage/vertices.cpp" #include "storage/vertices.hpp" +#include "communication/bolt/v1/serialization/bolt_serializer.hpp" using namespace std; From 2218b0e472435a2a437893ada339876c748eaa79 Mon Sep 17 00:00:00 2001 From: Kruno Tomola Fabro Date: Mon, 29 Aug 2016 14:51:31 +0100 Subject: [PATCH 2/3] Merged border and testes two querys. --- include/query_engine/code_generator.hpp | 3 ++- include/query_engine/i_code_cpu.hpp | 3 +-- include/storage/edge_accessor.hpp | 5 +++-- include/storage/edge_x_vertex.hpp | 10 ---------- .../bolt/v1/serialization/bolt_serializer.cpp | 1 - src/storage/edge_accessor.cpp | 10 ++++++++++ 6 files changed, 16 insertions(+), 16 deletions(-) diff --git a/include/query_engine/code_generator.hpp b/include/query_engine/code_generator.hpp index 964c824bb..09d7908cb 100644 --- a/include/query_engine/code_generator.hpp +++ b/include/query_engine/code_generator.hpp @@ -3,11 +3,11 @@ #include "config/config.hpp" #include "cypher/ast/ast.hpp" #include "cypher/compiler.hpp" +#include "logging/default.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" #include "utils/type_discovery.hpp" using std::string; @@ -56,6 +56,7 @@ public: {"stripped_hash", std::to_string(stripped_hash)}, {"query", query}, // {"stream", type_name().to_string()}, + // BARRIER !!!! {"stream", "RecordStream"}, {"code", cpp_traverser.code}}); diff --git a/include/query_engine/i_code_cpu.hpp b/include/query_engine/i_code_cpu.hpp index fd7d7d02f..ae44de8b9 100644 --- a/include/query_engine/i_code_cpu.hpp +++ b/include/query_engine/i_code_cpu.hpp @@ -13,8 +13,7 @@ template class ICodeCPU { public: - virtual bool run(barrier::Db &db, code_args_t &args, - Stream &stream) = 0; + virtual bool run(barrier::Db &db, code_args_t &args, Stream &stream) = 0; virtual ~ICodeCPU() {} }; diff --git a/include/storage/edge_accessor.hpp b/include/storage/edge_accessor.hpp index ddf638494..9fc06765a 100644 --- a/include/storage/edge_accessor.hpp +++ b/include/storage/edge_accessor.hpp @@ -3,6 +3,7 @@ #include "storage/edge.hpp" #include "storage/edge_record.hpp" #include "storage/record_accessor.hpp" +#include "storage/vertex_accessor.hpp" #include "utils/assert.hpp" #include "utils/reference_wrapper.hpp" @@ -23,7 +24,7 @@ public: const EdgeType &edge_type() const; - auto from() const; + VertexAccessor from() const; - auto to() const; + VertexAccessor to() const; }; diff --git a/include/storage/edge_x_vertex.hpp b/include/storage/edge_x_vertex.hpp index 9a92ea1fc..382fe4a0c 100644 --- a/include/storage/edge_x_vertex.hpp +++ b/include/storage/edge_x_vertex.hpp @@ -5,16 +5,6 @@ #include "storage/edge_accessor.hpp" #include "storage/vertex_accessor.hpp" -auto EdgeAccessor::from() const -{ - return VertexAccessor(this->vlist->from(), this->db); -} - -auto EdgeAccessor::to() const -{ - return VertexAccessor(this->vlist->to(), this->db); -} - auto VertexAccessor::out() const { DbTransaction &t = this->db; diff --git a/src/communication/bolt/v1/serialization/bolt_serializer.cpp b/src/communication/bolt/v1/serialization/bolt_serializer.cpp index b45445897..72331143a 100644 --- a/src/communication/bolt/v1/serialization/bolt_serializer.cpp +++ b/src/communication/bolt/v1/serialization/bolt_serializer.cpp @@ -4,7 +4,6 @@ #include "communication/bolt/v1/transport/chunked_encoder.hpp" #include "communication/bolt/v1/transport/socket_stream.hpp" #include "io/network/socket.hpp" -#include "storage/edge_x_vertex.hpp" template void bolt::BoltSerializer::write(const EdgeAccessor &edge) diff --git a/src/storage/edge_accessor.cpp b/src/storage/edge_accessor.cpp index 5bbd2f823..9ac3a75df 100644 --- a/src/storage/edge_accessor.cpp +++ b/src/storage/edge_accessor.cpp @@ -10,3 +10,13 @@ const EdgeType &EdgeAccessor::edge_type() const runtime_assert(this->record->data.edge_type != nullptr, "EdgeType is null"); return *this->record->data.edge_type; } + +VertexAccessor EdgeAccessor::from() const +{ + return VertexAccessor(this->vlist->from(), this->db); +} + +VertexAccessor EdgeAccessor::to() const +{ + return VertexAccessor(this->vlist->to(), this->db); +} From b2ce3d58a4185c7a35ba1dd9cde9941fc03060a4 Mon Sep 17 00:00:00 2001 From: Kruno Tomola Fabro Date: Tue, 30 Aug 2016 00:45:07 +0100 Subject: [PATCH 3/3] Added cleaner. Added multithreading to dbms. Skiplist now supports emplace insert. --- CMakeLists.txt | 3 + include/communication/bolt/v1/bolt.hpp | 7 +- .../concurrent/concurrent_map.hpp | 13 ++- .../data_structures/concurrent/skiplist.hpp | 84 ++++++++++++++++--- include/database/db.hpp | 1 + include/database/db_transaction.hpp | 9 ++ include/dbms/cleaner.hpp | 25 ++++++ include/dbms/dbms.hpp | 32 +++++++ include/mvcc/record.hpp | 6 ++ include/mvcc/version.hpp | 25 +++--- include/mvcc/version_list.hpp | 52 +++++++++++- include/storage/edge_type/edge_type_store.hpp | 6 +- include/storage/edges.hpp | 7 +- .../impl/nonunique_unordered_index.hpp | 7 +- .../indexes/impl/unique_ordered_index.hpp | 4 +- include/storage/indexes/index_base.hpp | 4 +- include/storage/indexes/index_record.hpp | 5 ++ include/storage/label/label_store.hpp | 6 +- include/storage/vertices.hpp | 2 + include/threading/id.hpp | 2 +- include/threading/thread.hpp | 21 ++--- include/transactions/snapshot.hpp | 21 +++++ include/transactions/transaction.hpp | 3 + include/utils/option.hpp | 11 +++ include/utils/placeholder.hpp | 7 ++ src/database/db_transaction.cpp | 74 ++++++++++++++++ src/dbms/cleaner.cpp | 37 ++++++++ src/dbms/dbms.cpp | 31 +++++++ src/dbms/dbms.hpp | 46 ---------- src/storage/edge_type/edge_type_store.cpp | 5 ++ src/storage/edges.cpp | 7 ++ .../impl/nonunique_unordered_index.cpp | 9 +- .../indexes/impl/unique_ordered_index.cpp | 10 ++- src/storage/indexes/index_record.cpp | 7 ++ src/storage/label/label_store.cpp | 2 + src/storage/vertices.cpp | 5 ++ src/threading/thread.cpp | 12 +++ src/transactions/transaction.cpp | 5 ++ 38 files changed, 509 insertions(+), 104 deletions(-) create mode 100644 include/dbms/cleaner.hpp create mode 100644 include/dbms/dbms.hpp create mode 100644 src/dbms/cleaner.cpp create mode 100644 src/dbms/dbms.cpp delete mode 100644 src/dbms/dbms.hpp create mode 100644 src/threading/thread.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index ebde7a3db..a1086b0df 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -402,6 +402,8 @@ EXECUTE_PROCESS( # TODO: create separate static library from bolt code set(memgraph_src_files + ${src_dir}/dbms/dbms.cpp + ${src_dir}/dbms/cleaner.cpp ${src_dir}/utils/string/transform.cpp ${src_dir}/utils/string/join.cpp ${src_dir}/utils/string/file.cpp @@ -416,6 +418,7 @@ set(memgraph_src_files ${src_dir}/communication/bolt/v1/transport/bolt_decoder.cpp ${src_dir}/communication/bolt/v1/transport/buffer.cpp ${src_dir}/communication/bolt/v1/serialization/bolt_serializer.cpp + ${src_dir}/threading/thread.cpp ${src_dir}/mvcc/id.cpp ${src_dir}/storage/vertices.cpp ${src_dir}/storage/edges.cpp diff --git a/include/communication/bolt/v1/bolt.hpp b/include/communication/bolt/v1/bolt.hpp index 850aa5195..0c9453931 100644 --- a/include/communication/bolt/v1/bolt.hpp +++ b/include/communication/bolt/v1/bolt.hpp @@ -1,8 +1,8 @@ #pragma once #include "communication/bolt/v1/states.hpp" -#include "io/network/socket.hpp" #include "dbms/dbms.hpp" +#include "io/network/socket.hpp" namespace bolt { @@ -16,11 +16,10 @@ class Bolt public: Bolt(); - Session* create_session(io::Socket&& socket); - void close(Session* session); + Session *create_session(io::Socket &&socket); + void close(Session *session); States states; Dbms dbms; }; - } diff --git a/include/data_structures/concurrent/concurrent_map.hpp b/include/data_structures/concurrent/concurrent_map.hpp index ae24568f1..b96906cf3 100644 --- a/include/data_structures/concurrent/concurrent_map.hpp +++ b/include/data_structures/concurrent/concurrent_map.hpp @@ -33,7 +33,7 @@ public: std::pair insert(const K &key, T &&data) { - return accessor.insert(item_t(key, std::forward(data))); + return accessor.insert(item_t(key, std::move(data))); } std::pair insert(K &&key, T &&data) @@ -42,6 +42,17 @@ public: item_t(std::forward(key), std::forward(data))); } + template + std::pair emplace(const K &key, + std::tuple first_args, + std::tuple second_args) + { + return accessor.emplace( + key, std::piecewise_construct, + std::forward>(first_args), + std::forward>(second_args)); + } + list_it_con find(const K &key) const { return accessor.find(key); } list_it find(const K &key) { return accessor.find(key); } diff --git a/include/data_structures/concurrent/skiplist.hpp b/include/data_structures/concurrent/skiplist.hpp index 227c6559e..f2450f8c9 100644 --- a/include/data_structures/concurrent/skiplist.hpp +++ b/include/data_structures/concurrent/skiplist.hpp @@ -148,7 +148,7 @@ public: static Node *create(const T &item, uint8_t height) { - return create(item, height); + return create(height, item); } static Node *create(T &&item, uint8_t height) @@ -160,6 +160,16 @@ public: return new (node) Node(std::move(item), height); } + template + static Node *emplace(uint8_t height, Args &&... args) + { + auto node = allocate(height); + + // we have raw memory and we need to construct an object + // of type Node on it + return new (node) Node(height, std::forward(args)...); + } + static void destroy(Node *node) { node->~Node(); @@ -180,6 +190,12 @@ public: new (&tower[i]) std::atomic{nullptr}; } + template + Node(uint8_t height, Args &&... args) : Node(height) + { + this->data.emplace(std::forward(args)...); + } + Node(T &&data, uint8_t height) : Node(height) { this->data.set(std::move(data)); @@ -519,12 +535,19 @@ public: std::pair insert(const T &item) { - return skiplist->insert(item, preds, succs); + return skiplist->insert(preds, succs, item); } std::pair insert(T &&item) { - return skiplist->insert(std::move(item), preds, succs); + return skiplist->insert(preds, succs, std::move(item)); + } + + template + std::pair emplace(K &key, Args &&... args) + { + return skiplist->emplace(preds, succs, key, + std::forward(args)...); } Iterator insert_non_unique(const T &item) @@ -786,13 +809,15 @@ private: // has the locks if (!lock_nodes(height, guards, preds, succs)) continue; - return insert_here(std::forward(data), preds, succs, height, - guards); + return insert_here(Node::create(std::move(data), height), preds, + succs, height, guards); } } // Insert unique data - std::pair insert(T &&data, Node *preds[], Node *succs[]) + // F - type of funct which will create new node if needed. Recieves height + // of node. + std::pair insert(Node *preds[], Node *succs[], T &&data) { while (true) { // TODO: before here was data.first @@ -817,18 +842,53 @@ private: // has the locks if (!lock_nodes(height, guards, preds, succs)) continue; - return {insert_here(std::move(data), preds, succs, height, guards), + return {insert_here(Node::create(std::move(data), height), 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[]) + // Insert unique data + // TODO: This is almost all duplicate code from insert + template + std::pair emplace(Node *preds[], Node *succs[], K &key, + Args &&... args) { - // you have the locks, create a new node - auto new_node = Node::create(std::move(data), height); + while (true) { + // TODO: before here was data.first + auto level = find_path(this, H - 1, key, 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(height, guards, preds, succs)) continue; + + return { + insert_here(Node::emplace(height, std::forward(args)...), + preds, succs, height, guards), + true}; + } + } + + // Inserts data to specified locked location. + Iterator insert_here(Node *new_node, Node *preds[], Node *succs[], + int height, guard_t guards[]) + { + // Node::create(std::move(data), height) // link the predecessors and successors, e.g. // // 4 HEAD ... P ------------------------> S ... NULL diff --git a/include/database/db.hpp b/include/database/db.hpp index f28a1189d..c95def98a 100644 --- a/include/database/db.hpp +++ b/include/database/db.hpp @@ -28,6 +28,7 @@ public: // I - type of function I:const tx::Transaction& -> // std::unique_ptr> // G - type of collection (verrtex/edge) + // TODO: Currently only one index at a time can be created. template bool create_index_on_vertex_property_family(const char *name, G &coll, I &create_index); diff --git a/include/database/db_transaction.hpp b/include/database/db_transaction.hpp index 07972557e..8e38e140c 100644 --- a/include/database/db_transaction.hpp +++ b/include/database/db_transaction.hpp @@ -17,6 +17,7 @@ class DbTransaction friend DbAccessor; public: + DbTransaction(Db &db); DbTransaction(Db &db, tx::Transaction &trans) : db(db), trans(trans) {} // Global transactional algorithms,operations and general methods meant for @@ -24,6 +25,14 @@ public: // This should provide cleaner hierarchy of operations on database. // For example cleaner. + // Cleans edge part of database. MUST be called by one cleaner thread at + // one time. + void clean_edge_section(); + + // Cleans vertex part of database. MUST be called by one cleaner thread at + // one time.. + void clean_vertex_section(); + // Updates indexes of Vertex/Edges in index_updates. True if indexes are // updated successfully. False means that transaction failed. bool update_indexes(); diff --git a/include/dbms/cleaner.hpp b/include/dbms/cleaner.hpp new file mode 100644 index 000000000..4b1ca9453 --- /dev/null +++ b/include/dbms/cleaner.hpp @@ -0,0 +1,25 @@ +#pragma once + +#include "database/db.hpp" + +class Thread; + +// How much sec is a cleaning_cycle in which cleaner will clean at most +// once. +constexpr size_t cleaning_cycle = 60; + +class Cleaning +{ + +public: + Cleaning(ConcurrentMap &dbs); + + ~Cleaning(); + +private: + ConcurrentMap &dbms; + + std::vector> cleaners; + + std::atomic cleaning = {true}; +}; diff --git a/include/dbms/dbms.hpp b/include/dbms/dbms.hpp new file mode 100644 index 000000000..e4c376209 --- /dev/null +++ b/include/dbms/dbms.hpp @@ -0,0 +1,32 @@ +#pragma once + +#include "data_structures/concurrent/concurrent_map.hpp" +#include "database/db.hpp" +#include "dbms/cleaner.hpp" + +class Dbms +{ +public: + Dbms() { create_default(); } + + // returns active database + Db &active(); + + // set active database + // if active database doesn't exist create one + Db &active(const std::string &name); + + // TODO: DELETE action + +private: + // creates default database + Db &create_default() { return active("default"); } + + // dbs container + ConcurrentMap dbs; + + // currently active database + std::atomic active_db; + + Cleaning cleaning = {dbs}; +}; diff --git a/include/mvcc/record.hpp b/include/mvcc/record.hpp index bc2e0ece2..aab7c63e9 100644 --- a/include/mvcc/record.hpp +++ b/include/mvcc/record.hpp @@ -102,6 +102,12 @@ public: return committed(hints.cre, tx.cre(), t); } + // True if record was deleted before id. + bool is_deleted_before(const Id &id) + { + return tx.exp() != Id(0) && tx.exp() < id; + } + // TODO: Test this // True if this record is visible for write. bool is_visible_write(const tx::Transaction &t) diff --git a/include/mvcc/version.hpp b/include/mvcc/version.hpp index 36b61f310..9039a2ffb 100644 --- a/include/mvcc/version.hpp +++ b/include/mvcc/version.hpp @@ -10,32 +10,35 @@ class Version { public: Version() = default; - Version(T* older) : older(older) {} + Version(T *older) : older(older) {} - ~Version() - { - delete older.load(std::memory_order_seq_cst); - } + ~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) + 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 + 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) + void next(T *value, std::memory_order order = std::memory_order_seq_cst) { older.store(value, order); } -private: - std::atomic older {nullptr}; -}; + // sets if as expected + bool cas(T *expected, T *set, + std::memory_order order = std::memory_order_seq_cst) + { + return older.compare_exchange_strong(expected, set, order); + } +private: + std::atomic older{nullptr}; +}; } diff --git a/include/mvcc/version_list.hpp b/include/mvcc/version_list.hpp index 29161fe27..2b7046548 100644 --- a/include/mvcc/version_list.hpp +++ b/include/mvcc/version_list.hpp @@ -11,7 +11,7 @@ namespace mvcc { template -class VersionList : public LazyGC> +class VersionList { friend class Accessor; @@ -51,6 +51,56 @@ public: auto gc_lock_acquire() { return std::unique_lock(lock); } + // Frees all records which are deleted by transaction older than given id. + // EXPECTS THAT THERE IS NO ACTIVE TRANSACTION WITH ID LESS THAN GIVEN ID. + // EXPECTS THAT THERE WON'T BE SIMULATAIUS CALLS FROM DIFFERENT THREADS OF + // THIS METHOD. + // True if this whole version list isn't needed any more. There is still + // possibilty that someone is reading it at this moment but he cant change + // it or get anything from it. + // TODO: Validate this method + bool gc_deleted(const Id &id) + { + auto r = head.load(std::memory_order_seq_cst); + T *bef = nullptr; + + // nullptr + // | + // [v1] ... + // | + // [v2] <------+ + // | | + // [v3] <------+ + // | | Jump backwards until you find a first old deleted + // [VerList] ----+ version, or you reach the end of the list + // + while (r != nullptr && !r->is_deleted_before(id)) { + bef = r; + r = r->next(std::memory_order_seq_cst); + } + + if (bef == nullptr) { + // if r==nullptr he is needed and it is expecting insert. + // if r!=nullptr vertex has been explicitly deleted. It can't be + // updated because for update, visible record is needed and at this + // point whe know that there is no visible record for any + // transaction. Also it cant be inserted because head isn't nullptr. + // Remove also requires visible record. Find wont return any record + // because none is visible. + return r != nullptr; + } else { + if (r != nullptr) { + // Bef is possible visible to some transaction but r is not and + // the implementation of this version list guarantees that + // record r and older records aren't accessed. + bef->next(nullptr, std::memory_order_seq_cst); + delete r; // THIS IS ISSUE IF MULTIPLE THREADS TRY TO DO THIS + } + + return false; + } + } + void vacuum() {} T *find(const tx::Transaction &t) const diff --git a/include/storage/edge_type/edge_type_store.hpp b/include/storage/edge_type/edge_type_store.hpp index 1268f7acf..ac30176b0 100644 --- a/include/storage/edge_type/edge_type_store.hpp +++ b/include/storage/edge_type/edge_type_store.hpp @@ -9,6 +9,10 @@ class EdgeTypeStore { public: + using store_t = ConcurrentMap>; + + store_t::Accessor access(); + const EdgeType &find_or_create(const char *name); bool contains(const char *name); // TODO: const @@ -24,5 +28,5 @@ public: // templetize the two of them private: - ConcurrentMap> edge_types; + store_t edge_types; }; diff --git a/include/storage/edges.hpp b/include/storage/edges.hpp index d171e0d21..b9fccad94 100644 --- a/include/storage/edges.hpp +++ b/include/storage/edges.hpp @@ -19,17 +19,22 @@ using EdgeIndexBase = IndexBase; class Edges { using prop_familys_t = ConcurrentMap; + using store_t = ConcurrentMap; public: + store_t::Accessor access(); + Option find(DbTransaction &t, const Id &id); // Creates new Edge and returns filled EdgeAccessor. EdgeAccessor insert(DbTransaction &t, VertexRecord *from, VertexRecord *to); + prop_familys_t::Accessor property_family_access(); + EdgePropertyFamily &property_family_find_or_create(const std::string &name); private: - ConcurrentMap edges; + store_t edges; // TODO: Because familys wont be removed this could be done with more // efficent // data structure. diff --git a/include/storage/indexes/impl/nonunique_unordered_index.hpp b/include/storage/indexes/impl/nonunique_unordered_index.hpp index 77c76fc75..6cd52ae55 100644 --- a/include/storage/indexes/impl/nonunique_unordered_index.hpp +++ b/include/storage/indexes/impl/nonunique_unordered_index.hpp @@ -9,6 +9,7 @@ template class NonUniqueUnorderedIndex : public IndexBase { public: + using store_t = List>; // typedef T value_type; // typedef K key_type; @@ -33,9 +34,9 @@ public: // Removes for all transactions obsolete Records. // Cleaner has to call this method when he decideds that it is time for - // cleaning. - void clean(DbTransaction &) final; + // cleaning. Id must be id of oldest active transaction. + void clean(const Id &id) final; private: - List> list; + store_t list; }; diff --git a/include/storage/indexes/impl/unique_ordered_index.hpp b/include/storage/indexes/impl/unique_ordered_index.hpp index 69daeecf7..d4519e6d9 100644 --- a/include/storage/indexes/impl/unique_ordered_index.hpp +++ b/include/storage/indexes/impl/unique_ordered_index.hpp @@ -32,8 +32,8 @@ public: // Removes for all transactions obsolete Records. // Cleaner has to call this method when he decideds that it is time for - // cleaning. - void clean(DbTransaction &) final; + // cleaning. Id must be id of oldest active transaction. + void clean(const Id &id) final; private: ConcurrentSet> set; diff --git a/include/storage/indexes/index_base.hpp b/include/storage/indexes/index_base.hpp index 64b9c61fd..be4d98612 100644 --- a/include/storage/indexes/index_base.hpp +++ b/include/storage/indexes/index_base.hpp @@ -56,8 +56,8 @@ public: // Removes for all transactions obsolete Records. // Cleaner has to call this method when he decideds that it is time for - // cleaning. - virtual void clean(DbTransaction &) = 0; + // cleaning. Id must be id of oldest active transaction. + virtual void clean(const Id &id) = 0; // Activates index for readers. void activate(); diff --git a/include/storage/indexes/index_record.hpp b/include/storage/indexes/index_record.hpp index 8381f8292..d970d9bda 100644 --- a/include/storage/indexes/index_record.hpp +++ b/include/storage/indexes/index_record.hpp @@ -1,5 +1,6 @@ #pragma once +#include "mvcc/id.hpp" #include "utils/border.hpp" #include "utils/total_ordering.hpp" @@ -52,6 +53,10 @@ public: bool is_valid(tx::Transaction &t) const; + // True if it can be removed. + bool to_clean(const Id &oldest_active) const; + + // This method is valid only if is_valid is true. const auto access(DbTransaction &db) const; const K key; diff --git a/include/storage/label/label_store.hpp b/include/storage/label/label_store.hpp index 5e83ec061..35ce3166a 100644 --- a/include/storage/label/label_store.hpp +++ b/include/storage/label/label_store.hpp @@ -9,6 +9,10 @@ class LabelStore { public: + using store_t = ConcurrentMap>; + + store_t::Accessor access(); + const Label &find_or_create(const char *name); bool contains(const char *name); // TODO: const @@ -17,5 +21,5 @@ public: // return { Label, is_found } private: - ConcurrentMap> labels; + store_t labels; }; diff --git a/include/storage/vertices.hpp b/include/storage/vertices.hpp index bb9336bc1..f1a528105 100644 --- a/include/storage/vertices.hpp +++ b/include/storage/vertices.hpp @@ -33,6 +33,8 @@ public: VertexPropertyFamily & property_family_find_or_create(const std::string &name); + prop_familys_t::Accessor property_family_access(); + private: vertices_t vertices; // TODO: Because families wont be removed this could be done with more diff --git a/include/threading/id.hpp b/include/threading/id.hpp index 9252b4e28..216140392 100644 --- a/include/threading/id.hpp +++ b/include/threading/id.hpp @@ -2,5 +2,5 @@ namespace this_thread { - thread_local unsigned id = 0; +// thread_local unsigned id = 0; }; diff --git a/include/threading/thread.hpp b/include/threading/thread.hpp index 93c3eeffe..3aefb5cde 100644 --- a/include/threading/thread.hpp +++ b/include/threading/thread.hpp @@ -1,11 +1,11 @@ #pragma once #include -#include #include +#include +#include "threading/id.hpp" #include "utils/underlying_cast.hpp" -#include "id.hpp" class Thread { @@ -28,27 +28,20 @@ public: } Thread() = default; - Thread(const Thread&) = delete; + Thread(const Thread &) = delete; - Thread(Thread&& other) - { - assert(thread_id == UNINITIALIZED); - thread_id = other.thread_id; - thread = std::move(other.thread); - } + Thread(Thread &&other); - void join() { return thread.join(); } + void join(); private: unsigned thread_id = UNINITIALIZED; std::thread thread; template - void start_thread(F&& f) + void start_thread(F &&f) { - this_thread::id = thread_id; + // this_thread::id = thread_id; f(); } }; - -std::atomic Thread::thread_counter {1}; diff --git a/include/transactions/snapshot.hpp b/include/transactions/snapshot.hpp index 04168a959..5c4bdd715 100644 --- a/include/transactions/snapshot.hpp +++ b/include/transactions/snapshot.hpp @@ -3,6 +3,8 @@ #include #include +#include "utils/option.hpp" + namespace tx { @@ -23,6 +25,25 @@ public: return std::binary_search(active.begin(), active.end(), xid); } + // Return id of oldest transaction. None if there is no transactions in + // snapshot. + Option oldest_active() + { + auto n = active.size(); + if (n > 0) { + Id min = active[0]; + for (auto i = 1; i < n; i++) { + if (active[i] < min) { + min = active[i]; + } + } + return Option(min); + + } else { + return Option(); + } + } + void insert(const id_t &id) { active.push_back(id); } void remove(const id_t &id) diff --git a/include/transactions/transaction.hpp b/include/transactions/transaction.hpp index 7c618aadf..864a71f9b 100644 --- a/include/transactions/transaction.hpp +++ b/include/transactions/transaction.hpp @@ -33,6 +33,9 @@ public: // snapshot will be empty. void wait_for_active(); + // Return id of oldest transaction from snapshot. + Id oldest_active(); + // True if id is in snapshot. bool is_active(const Id &id) const; void take_lock(RecordLock &lock); diff --git a/include/utils/option.hpp b/include/utils/option.hpp index 50b2a50cc..56fa51d06 100644 --- a/include/utils/option.hpp +++ b/include/utils/option.hpp @@ -131,6 +131,17 @@ public: return std::move(*data._M_ptr()); } + // Takes if it exists otherwise returns given value. + T take_or(T &&value) + { + if (initialized) { + initialized = false; + return std::move(*data._M_ptr()); + } else { + return std::move(value); + } + } + explicit operator bool() const { return initialized; } private: diff --git a/include/utils/placeholder.hpp b/include/utils/placeholder.hpp index a1eb889c8..31bdf7cb4 100644 --- a/include/utils/placeholder.hpp +++ b/include/utils/placeholder.hpp @@ -43,6 +43,13 @@ public: initialized = true; } + template + void emplace(Args &&... args) + { + new (data._M_addr()) T(args...); + initialized = true; + } + private: __gnu_cxx::__aligned_buffer data; bool initialized = false; diff --git a/src/database/db_transaction.cpp b/src/database/db_transaction.cpp index d55552b06..c7db079e9 100644 --- a/src/database/db_transaction.cpp +++ b/src/database/db_transaction.cpp @@ -1,5 +1,6 @@ #include "database/db_transaction.hpp" +#include "database/db.hpp" #include "storage/edge.hpp" #include "storage/edge_type/edge_type.hpp" #include "storage/label/label.hpp" @@ -10,6 +11,79 @@ return false; \ } +DbTransaction::DbTransaction(Db &db) : db(db), trans(db.tx_engine.begin()) {} + +// Cleaning for indexes in labels and edge_type +template +void clean_indexes(A &&acc, Id oldest_active) +{ + for (auto &l : acc) { + l.second.get()->index().clean(oldest_active); + } +} + +// Cleaning for version lists +template +void clean_version_lists(A &&acc, Id oldest_active) +{ + for (auto &vlist : acc) { + if (vlist.second.gc_deleted(oldest_active)) { + // TODO: Optimization, iterator with remove method. + bool succ = acc.remove(vlist.first); + assert(succ); // There is other cleaner here + } + } +} + +// Cleaning for indexes in properties. +template +void clean_property_indexes(A &&acc, Id oldest_active) +{ + for (auto &family : acc) { + auto oi = family.second->index.get_read(); + if (oi.is_present()) { + oi.get()->clean(oldest_active); + } + } + + // TODO: Code for cleaning other indexes which are not yet coded into + // the database. +} + +// Cleans edge part of database. Should be called by one cleaner thread at +// one time. +void DbTransaction::clean_edge_section() +{ + Id oldest_active = trans.oldest_active(); + + // Clean edge_type index + clean_indexes(db.graph.edge_type_store.access(), oldest_active); + + // Clean family_type_s edge index + clean_property_indexes(db.graph.edges.property_family_access(), + oldest_active); + + // Clean Edge list + clean_version_lists(db.graph.edges.access(), oldest_active); +} + +// Cleans vertex part of database. Should be called by one cleaner thread at +// one time. +void DbTransaction::clean_vertex_section() +{ + Id oldest_active = trans.oldest_active(); + + // Clean label index + clean_indexes(db.graph.label_store.access(), oldest_active); + + // Clean family_type_s vertex index + clean_property_indexes(db.graph.vertices.property_family_access(), + oldest_active); + + // Clean vertex list + clean_version_lists(db.graph.vertices.access(), oldest_active); +} + template bool update_property_indexes(IU &iu, const tx::Transaction &t) { diff --git a/src/dbms/cleaner.cpp b/src/dbms/cleaner.cpp new file mode 100644 index 000000000..7ab29bcac --- /dev/null +++ b/src/dbms/cleaner.cpp @@ -0,0 +1,37 @@ +#include "dbms/cleaner.hpp" + +#include +#include +#include + +#include "database/db_transaction.hpp" +#include "threading/thread.hpp" + +Cleaning::Cleaning(ConcurrentMap &dbs) : dbms(dbs) +{ + cleaners.push_back(std::make_unique([&]() { + std::time_t last_clean = std::time(nullptr); + while (cleaning.load(std::memory_order_acquire)) { + std::time_t now = std::time(nullptr); + + if (now >= last_clean + cleaning_cycle) { + for (auto &db : dbs.access()) { + DbTransaction t(db.second); + t.clean_edge_section(); + t.clean_vertex_section(); + } + last_clean = now; + } else { + std::this_thread::sleep_for(std::chrono::seconds(1)); + } + } + })); +} + +Cleaning::~Cleaning() +{ + cleaning.store(false, std::memory_order_release); + for (auto &t : cleaners) { + t.get()->join(); + } +} diff --git a/src/dbms/dbms.cpp b/src/dbms/dbms.cpp new file mode 100644 index 000000000..6711af348 --- /dev/null +++ b/src/dbms/dbms.cpp @@ -0,0 +1,31 @@ +#include "dbms/dbms.hpp" + +// returns active database +Db &Dbms::active() +{ + Db *active = active_db.load(std::memory_order_acquire); + if (UNLIKELY(active == nullptr)) { + return create_default(); + } else { + return *active; + } +} + +// set active database +// if active database doesn't exist create one +Db &Dbms::active(const std::string &name) +{ + auto acc = dbs.access(); + // create db if it doesn't exist + auto it = acc.find(name); + if (it == acc.end()) { + it = acc.emplace(name, std::forward_as_tuple(name), + std::forward_as_tuple(name)) + .first; + } + + // set and return active db + auto &db = it->second; + active_db.store(&db, std::memory_order_release); + return db; +} diff --git a/src/dbms/dbms.hpp b/src/dbms/dbms.hpp deleted file mode 100644 index d63af620c..000000000 --- a/src/dbms/dbms.hpp +++ /dev/null @@ -1,46 +0,0 @@ -#pragma once - -#include - -#include "database/db.hpp" - -class Dbms -{ -public: - Dbms() { create_default(); } - - // returns active database - Db &active() - { - if (UNLIKELY(active_db == nullptr)) create_default(); - - return *active_db; - } - - // set active database - // if active database doesn't exist create one - Db &active(const std::string &name) - { - // create db if it doesn't exist - if (dbs.find(name) == dbs.end()) { - dbs.emplace(std::piecewise_construct, std::forward_as_tuple(name), - std::forward_as_tuple(name)); - } - - // set and return active db - auto &db = dbs.at(name); - return active_db = &db, *active_db; - } - - // TODO: DELETE action - -private: - // dbs container - std::map dbs; - - // currently active database - Db *active_db; - - // creates default database - void create_default() { active("default"); } -}; diff --git a/src/storage/edge_type/edge_type_store.cpp b/src/storage/edge_type/edge_type_store.cpp index 7dc89a83a..3062eee1d 100644 --- a/src/storage/edge_type/edge_type_store.cpp +++ b/src/storage/edge_type/edge_type_store.cpp @@ -1,5 +1,10 @@ #include "storage/edge_type/edge_type_store.hpp" +EdgeTypeStore::store_t::Accessor EdgeTypeStore::access() +{ + return edge_types.access(); +} + const EdgeType &EdgeTypeStore::find_or_create(const char *name) { auto accessor = edge_types.access(); diff --git a/src/storage/edges.cpp b/src/storage/edges.cpp index e860d8bea..6bd383d04 100644 --- a/src/storage/edges.cpp +++ b/src/storage/edges.cpp @@ -3,6 +3,8 @@ #include "storage/edge_accessor.hpp" #include "utils/iterator/iterator.hpp" +Edges::store_t::Accessor Edges::access() { return edges.access(); } + Option Edges::find(DbTransaction &t, const Id &id) { auto edges_accessor = edges.access(); @@ -35,6 +37,11 @@ EdgeAccessor Edges::insert(DbTransaction &t, VertexRecord *from, return EdgeAccessor(edge, &inserted_edge_record->second, t); } +Edges::prop_familys_t::Accessor Edges::property_family_access() +{ + return prop_familys.access(); +} + EdgePropertyFamily & Edges::property_family_find_or_create(const std::string &name) { diff --git a/src/storage/indexes/impl/nonunique_unordered_index.cpp b/src/storage/indexes/impl/nonunique_unordered_index.cpp index 9bf7ef37c..d96f148ad 100644 --- a/src/storage/indexes/impl/nonunique_unordered_index.cpp +++ b/src/storage/indexes/impl/nonunique_unordered_index.cpp @@ -65,9 +65,14 @@ auto NonUniqueUnorderedIndex::for_range_exact(DbAccessor &t_v, } template -void NonUniqueUnorderedIndex::clean(DbTransaction &) +void NonUniqueUnorderedIndex::clean(const Id &id) { - // TODO: Actual cleaning + auto end = list.end(); + for (auto it = list.begin(); it != end; it++) { + if (it->to_clean(id)) { + it.remove(); + } + } } template class NonUniqueUnorderedIndex; diff --git a/src/storage/indexes/impl/unique_ordered_index.cpp b/src/storage/indexes/impl/unique_ordered_index.cpp index bc046311d..a0de33592 100644 --- a/src/storage/indexes/impl/unique_ordered_index.cpp +++ b/src/storage/indexes/impl/unique_ordered_index.cpp @@ -84,9 +84,15 @@ auto UniqueOrderedIndex::for_range_exact(DbAccessor &t_v, } template -void UniqueOrderedIndex::clean(DbTransaction &) +void UniqueOrderedIndex::clean(const Id &id) { - // TODO: Actual cleaning + auto acc = set.access(); + for (auto ir : acc) { + if (ir.to_clean(id)) { + // TODO: Optimization, iterator with remove method. + acc.remove(ir); + } + } } template class UniqueOrderedIndex; diff --git a/src/storage/indexes/index_record.cpp b/src/storage/indexes/index_record.cpp index c3e42142c..ba7a02397 100644 --- a/src/storage/indexes/index_record.cpp +++ b/src/storage/indexes/index_record.cpp @@ -31,6 +31,13 @@ bool IndexRecord::is_valid(tx::Transaction &t) const return record == vlist->find(t); } +template +bool IndexRecord::to_clean(const Id &oldest_active) const +{ + assert(!empty()); + return record->is_deleted_before(oldest_active); +} + template const auto IndexRecord::access(DbTransaction &db) const { diff --git a/src/storage/label/label_store.cpp b/src/storage/label/label_store.cpp index 3eb5af737..db9fdad7e 100644 --- a/src/storage/label/label_store.cpp +++ b/src/storage/label/label_store.cpp @@ -1,5 +1,7 @@ #include "storage/label/label_store.hpp" +LabelStore::store_t::Accessor LabelStore::access() { return labels.access(); } + const Label &LabelStore::find_or_create(const char *name) { auto accessor = labels.access(); diff --git a/src/storage/vertices.cpp b/src/storage/vertices.cpp index 31be412cc..aae5aa1c9 100644 --- a/src/storage/vertices.cpp +++ b/src/storage/vertices.cpp @@ -37,6 +37,11 @@ VertexAccessor Vertices::insert(DbTransaction &t) return VertexAccessor(vertex, &inserted_vertex_record->second, t); } +Vertices::prop_familys_t::Accessor Vertices::property_family_access() +{ + return prop_familys.access(); +} + VertexPropertyFamily & Vertices::property_family_find_or_create(const std::string &name) { diff --git a/src/threading/thread.cpp b/src/threading/thread.cpp new file mode 100644 index 000000000..b6fe76486 --- /dev/null +++ b/src/threading/thread.cpp @@ -0,0 +1,12 @@ +#include "threading/thread.hpp" + +Thread::Thread(Thread &&other) +{ + assert(thread_id == UNINITIALIZED); + thread_id = other.thread_id; + thread = std::move(other.thread); +} + +void Thread::join() { return thread.join(); } + +std::atomic Thread::thread_counter{1}; diff --git a/src/transactions/transaction.cpp b/src/transactions/transaction.cpp index aab629276..d8ac4673b 100644 --- a/src/transactions/transaction.cpp +++ b/src/transactions/transaction.cpp @@ -31,6 +31,11 @@ bool Transaction::is_active(const Id &id) const return snapshot.is_active(id); } +Id Transaction::oldest_active() +{ + return snapshot.oldest_active().take_or(Id(id)); +} + void Transaction::take_lock(RecordLock &lock) { locks.take(&lock, id); } void Transaction::commit() { engine.commit(*this); }