diff --git a/src/communication/bolt/v1/encoder/chunked_encoder_buffer.hpp b/src/communication/bolt/v1/encoder/chunked_encoder_buffer.hpp index 7626ca4c2..7c3665ed8 100644 --- a/src/communication/bolt/v1/encoder/chunked_encoder_buffer.hpp +++ b/src/communication/bolt/v1/encoder/chunked_encoder_buffer.hpp @@ -8,6 +8,7 @@ #include #include "communication/bolt/v1/constants.hpp" +#include "utils/assert.hpp" #include "utils/bswap.hpp" namespace communication::bolt { diff --git a/src/communication/bolt/v1/session.hpp b/src/communication/bolt/v1/session.hpp index d00c7580e..7e813ed6c 100644 --- a/src/communication/bolt/v1/session.hpp +++ b/src/communication/bolt/v1/session.hpp @@ -23,6 +23,21 @@ namespace communication::bolt { +/** + * Bolt SessionData + * + * This class is responsible for holding references to Dbms and QueryEngine + * that are passed through the network server and worker to the session. + * + * @tparam OutputStream type of output stream (could be a bolt output stream or + * a test output stream) + */ +template +struct SessionData { + Dbms dbms; + QueryEngine query_engine; +}; + /** * Bolt Session * @@ -37,8 +52,8 @@ class Session { using StreamBuffer = io::network::StreamBuffer; public: - Session(Socket &&socket, Dbms &dbms, QueryEngine &query_engine) - : socket_(std::move(socket)), dbms_(dbms), query_engine_(query_engine) { + Session(Socket &&socket, SessionData &data) + : socket_(std::move(socket)), dbms_(data.dbms), query_engine_(data.query_engine) { event_.data.ptr = this; } diff --git a/src/communication/server.hpp b/src/communication/server.hpp index 0b4c8ad40..012117d91 100644 --- a/src/communication/server.hpp +++ b/src/communication/server.hpp @@ -9,9 +9,6 @@ #include #include -#include "database/dbms.hpp" -#include "query/engine.hpp" - #include "communication/worker.hpp" #include "io/network/event_listener.hpp" #include "utils/assert.hpp" @@ -36,17 +33,17 @@ namespace communication { * @tparam OutputStream the server has to get the output stream as a template parameter because the output stream is templated * @tparam Socket the input/output socket that should be used + * @tparam SessionData the class with objects that will be forwarded to the session */ -template +template class Server - : public io::network::EventListener> { + : public io::network::EventListener> { using Event = io::network::Epoll::Event; public: - Server(Socket &&socket, Dbms &dbms, QueryEngine &query_engine) + Server(Socket &&socket, SessionData &session_data) : socket_(std::forward(socket)), - dbms_(dbms), - query_engine_(query_engine) { + session_data_(session_data) { event_.data.fd = socket_; // TODO: EPOLLET is hard to use -> figure out how should EPOLLET be used @@ -61,8 +58,8 @@ class Server workers_.reserve(n); for (size_t i = 0; i < n; ++i) { workers_.push_back( - std::make_unique>( - dbms_, query_engine_)); + std::make_unique>( + session_data_)); workers_.back()->Start(alive_); } std::cout << "Server is fully armed and operational" << std::endl; @@ -113,14 +110,13 @@ class Server void OnErrorEvent(Event &event) { close(event.data.fd); } private: - std::vector::uptr> workers_; + std::vector::uptr> workers_; std::atomic alive_{true}; int idx_{0}; Socket socket_; - Dbms &dbms_; - QueryEngine &query_engine_; Event event_; + SessionData &session_data_; }; } // namespace communication diff --git a/src/communication/worker.hpp b/src/communication/worker.hpp index 5f91080e2..281395ae2 100644 --- a/src/communication/worker.hpp +++ b/src/communication/worker.hpp @@ -9,10 +9,6 @@ #include -#include "database/dbms.hpp" -#include "query/engine.hpp" - -#include "communication/bolt/v1/session.hpp" #include "io/network/network_error.hpp" #include "io/network/stream_reader.hpp" @@ -33,19 +29,19 @@ namespace communication { * @tparam OutputStream the worker has to get the output stream as a template parameter because the output stream is templated * @tparam Socket the input/output socket that should be used + * @tparam SessionData the class with objects that will be forwarded to the session */ -template +template class Worker - : public io::network::StreamReader, + : public io::network::StreamReader, Session> { using StreamBuffer = io::network::StreamBuffer; public: - using uptr = std::unique_ptr>; + using uptr = std::unique_ptr>; - Worker(Dbms &dbms, QueryEngine &query_engine) - : dbms_(dbms), query_engine_(query_engine) {} + Worker(SessionData &session_data) : session_data_(session_data) {} Session &OnConnect(Socket &&socket) { DLOG(INFO) << "Accepting connection on socket " << socket.id(); @@ -53,7 +49,7 @@ class Worker // TODO fix session lifecycle handling // dangling pointers are not cool :) // TODO attach currently active Db - return *(new Session(std::forward(socket), dbms_, query_engine_)); + return *(new Session(std::forward(socket), session_data_)); } void OnError(Session &session) { @@ -101,7 +97,6 @@ class Worker } private: - Dbms &dbms_; - QueryEngine &query_engine_; + SessionData &session_data_; }; } diff --git a/src/memgraph_bolt.cpp b/src/memgraph_bolt.cpp index 0c36fb718..93dbe36a1 100644 --- a/src/memgraph_bolt.cpp +++ b/src/memgraph_bolt.cpp @@ -4,9 +4,6 @@ #include #include -#include "database/dbms.hpp" -#include "query/engine.hpp" - #include "communication/bolt/v1/session.hpp" #include "communication/server.hpp" @@ -26,8 +23,9 @@ using session_t = communication::bolt::Session; using result_stream_t = communication::bolt::ResultStream>>; +using session_data_t = communication::bolt::SessionData; using bolt_server_t = - communication::Server; + communication::Server; DEFINE_string(interface, "0.0.0.0", "Default interface on which to listen."); DEFINE_string(port, "7687", "Default port on which to listen."); @@ -123,11 +121,11 @@ int main(int argc, char **argv) { LOG(FATAL) << "Cannot listen on socket!"; } - Dbms dbms; - QueryEngine query_engine; + // Initialize bolt session data (Dbms and QueryEngine). + session_data_t session_data; // Initialize server. - bolt_server_t server(std::move(socket), dbms, query_engine); + bolt_server_t server(std::move(socket), session_data); // register SIGTERM handler SignalHandler::register_handler(Signal::Terminate, diff --git a/tests/concurrent/network_common.hpp b/tests/concurrent/network_common.hpp index 691ec796d..512aaf3eb 100644 --- a/tests/concurrent/network_common.hpp +++ b/tests/concurrent/network_common.hpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include @@ -11,10 +12,8 @@ #include "communication/bolt/v1/decoder/buffer.hpp" #include "communication/server.hpp" -#include "database/dbms.hpp" #include "io/network/epoll.hpp" #include "io/network/socket.hpp" -#include "query/engine.hpp" static constexpr const int SIZE = 60000; static constexpr const int REPLY = 10; @@ -24,10 +23,11 @@ using socket_t = io::network::Socket; class TestOutputStream {}; +class TestData {}; + class TestSession { public: - TestSession(socket_t &&socket, Dbms &dbms, - QueryEngine &query_engine) + TestSession(socket_t &&socket, TestData &data) : socket_(std::move(socket)) { event_.data.ptr = this; } @@ -65,7 +65,7 @@ class TestSession { }; using test_server_t = - communication::Server; + communication::Server; void server_start(void *serverptr, int num) { ((test_server_t *)serverptr)->Start(num); diff --git a/tests/concurrent/network_read_hang.cpp b/tests/concurrent/network_read_hang.cpp index 424dc1195..9eb2cb062 100644 --- a/tests/concurrent/network_read_hang.cpp +++ b/tests/concurrent/network_read_hang.cpp @@ -15,10 +15,8 @@ #include "communication/bolt/v1/decoder/buffer.hpp" #include "communication/server.hpp" -#include "database/dbms.hpp" #include "io/network/epoll.hpp" #include "io/network/socket.hpp" -#include "query/engine.hpp" static constexpr const char interface[] = "127.0.0.1"; @@ -27,10 +25,11 @@ using socket_t = io::network::Socket; class TestOutputStream {}; +class TestData {}; + class TestSession { public: - TestSession(socket_t &&socket, Dbms &dbms, - QueryEngine &query_engine) + TestSession(socket_t &&socket, TestData &data) : socket_(std::move(socket)) { event_.data.ptr = this; } @@ -53,7 +52,7 @@ class TestSession { }; using test_server_t = - communication::Server; + communication::Server; test_server_t *serverptr; std::atomic run{true}; @@ -90,9 +89,8 @@ TEST(Network, SocketReadHangOnConcurrentConnections) { printf("ADDRESS: %s, PORT: %d\n", ep.address(), ep.port()); // initialize server - Dbms dbms; - QueryEngine query_engine; - test_server_t server(std::move(socket), dbms, query_engine); + TestData data; + test_server_t server(std::move(socket), data); serverptr = &server; // start server diff --git a/tests/concurrent/network_server.cpp b/tests/concurrent/network_server.cpp index 687097889..56c27f52b 100644 --- a/tests/concurrent/network_server.cpp +++ b/tests/concurrent/network_server.cpp @@ -26,9 +26,8 @@ TEST(Network, Server) { printf("ADDRESS: %s, PORT: %d\n", ep.address(), ep.port()); // initialize server - Dbms dbms; - QueryEngine query_engine; - test_server_t server(std::move(socket), dbms, query_engine); + TestData session_data; + test_server_t server(std::move(socket), session_data); serverptr = &server; // start server diff --git a/tests/concurrent/network_session_leak.cpp b/tests/concurrent/network_session_leak.cpp index be33bcff7..afa5c0bc7 100644 --- a/tests/concurrent/network_session_leak.cpp +++ b/tests/concurrent/network_session_leak.cpp @@ -30,9 +30,8 @@ TEST(Network, SessionLeak) { printf("ADDRESS: %s, PORT: %d\n", ep.address(), ep.port()); // initialize server - Dbms dbms; - QueryEngine query_engine; - test_server_t server(std::move(socket), dbms, query_engine); + TestData session_data; + test_server_t server(std::move(socket), session_data); serverptr = &server; // start server diff --git a/tests/unit/bolt_common.hpp b/tests/unit/bolt_common.hpp index 36c079222..3e83804bc 100644 --- a/tests/unit/bolt_common.hpp +++ b/tests/unit/bolt_common.hpp @@ -2,11 +2,11 @@ #include #include #include +#include #include #include -#include "database/dbms.hpp" #include "gtest/gtest.h" /** diff --git a/tests/unit/bolt_encoder.cpp b/tests/unit/bolt_encoder.cpp index fb428fd9a..5f7995a61 100644 --- a/tests/unit/bolt_encoder.cpp +++ b/tests/unit/bolt_encoder.cpp @@ -2,6 +2,7 @@ #include "bolt_testdata.hpp" #include "communication/bolt/v1/encoder/encoder.hpp" +#include "database/dbms.hpp" #include "database/graph_db.hpp" #include "database/graph_db_accessor.hpp" #include "query/typed_value.hpp" diff --git a/tests/unit/bolt_session.cpp b/tests/unit/bolt_session.cpp index bc693481f..88a5c39ed 100644 --- a/tests/unit/bolt_session.cpp +++ b/tests/unit/bolt_session.cpp @@ -11,15 +11,15 @@ DECLARE_bool(interpret); // TODO: This could be done in fixture. // Shortcuts for writing variable initializations in tests #define INIT_VARS \ - Dbms dbms; \ TestSocket socket(10); \ - QueryEngine query_engine; \ - SessionT session(std::move(socket), dbms, query_engine); \ + SessionDataT session_data; \ + SessionT session(std::move(socket), session_data); \ std::vector &output = session.socket_.output; using ResultStreamT = communication::bolt::ResultStream>>; +using SessionDataT = communication::bolt::SessionData; using SessionT = communication::bolt::Session; using StateT = communication::bolt::State;