Initial version of dbms_simulator, partial wsrep provider loading

CMakeLists.txt

@@ -14,10 +14,12 @@ if (NOT WITH_WSREP_API)
 else()
   check_include_file("${WITH_WSREP_API}/wsrep_api.h" HAVE_WSREP_API_HPP)
   include_directories("${WITH_WSREP_API}")
+  link_directories("${WITH_WSREP_API}")
 endif()
 
 find_package(Boost 1.54.0 REQUIRED
   unit_test_framework
+  program_options
   )
 
 # Coverage

src/CMakeLists.txt

@@ -8,7 +8,9 @@ add_library(trrep
   provider.cpp
   client_context.cpp
   server_context.cpp
-  transaction_context.cpp)
+  transaction_context.cpp
+  wsrep_provider_v26.cpp)
+target_link_libraries(trrep wsrep dl)
 
 add_executable(trrep_test
   mock_client_context.cpp
@@ -23,3 +25,7 @@ add_test(NAME trrep_test
   COMMAND ${CMAKE_CURRENT_SOURCE_DIR}/trrep_test
   )
 
+add_executable(dbms_simulator
+  dbms_simulator.cpp)
+target_link_libraries(dbms_simulator trrep ${Boost_PROGRAM_OPTIONS_LIBRARY})
+set_property(TARGET dbms_simulator PROPERTY CXX_STANDARD 14)

src/client_context.hpp

@@ -118,7 +118,7 @@ namespace trrep
         //
         client_context(trrep::mutex& mutex,
                        trrep::server_context& server_context,
-                       client_id id,
+                       const client_id& id,
                        enum mode mode)
             : mutex_(mutex)
             , server_context_(server_context)

src/dbms_simulator.cpp

@@ -0,0 +1,164 @@
+//
+// Copyright (C) 2018 Codership Oy <info@codership.com>
+//
+
+//
+// This file implementes a simple DBMS simulator which
+// will launch one or server threads and replicates transactions
+// through trrep interface.
+//
+
+#include "server_context.hpp"
+#include "client_context.hpp"
+
+#include <boost/program_options.hpp>
+
+#include <iostream>
+#include <memory>
+#include <map>
+#include <atomic>
+
+class dbms_server;
+
+class dbms_simulator
+{
+public:
+    dbms_simulator(size_t n_servers,
+                   size_t n_clients,
+                   const std::string& wsrep_provider,
+                   const std::string& wsrep_provider_options)
+        : servers_()
+        , n_servers_(n_servers)
+        , n_clients_(n_clients)
+        , wsrep_provider_(wsrep_provider)
+        , wsrep_provider_options_(wsrep_provider_options)
+    { }
+    void start();
+    void stop();
+private:
+    std::map<size_t, std::unique_ptr<dbms_server>> servers_;
+    size_t n_servers_;
+    size_t n_clients_;
+    std::string wsrep_provider_;
+    std::string wsrep_provider_options_;
+};
+
+class dbms_client;
+
+class dbms_server : public trrep::server_context
+{
+public:
+    dbms_server(const std::string& name, const std::string& id)
+        : trrep::server_context(name, id, trrep::server_context::rm_async)
+        , mutex_()
+        , last_client_id_(0)
+        , clients_()
+    { }
+
+    void on_connect() { }
+    void on_view() { }
+    void on_sync() { }
+    trrep::client_context* local_client_context();
+private:
+    trrep::default_mutex mutex_;
+    std::atomic<size_t> last_client_id_;
+    std::map<size_t, std::unique_ptr<dbms_client>> clients_;
+};
+
+class dbms_client : public trrep::client_context
+{
+public:
+    dbms_client(dbms_server& server, const trrep::client_id& id,
+                enum trrep::client_context::mode mode)
+        : trrep::client_context(mutex_, server, id, mode)
+        , mutex_()
+    { }
+    bool do_2pc() const { return false; }
+    int apply(trrep::transaction_context&, const trrep::data&)
+    {
+        return 0;
+    }
+    int commit(trrep::transaction_context&)
+    {
+        return 0;
+    }
+    int rollback(trrep::transaction_context&)
+    {
+        return 0;
+    }
+private:
+    trrep::default_mutex mutex_;
+};
+
+trrep::client_context* dbms_server::local_client_context()
+{
+    std::ostringstream id_os;
+    size_t client_id(last_client_id_.fetch_add(1) + 1);
+    trrep::unique_lock<trrep::mutex> lock(mutex_);
+    return clients_.insert(std::make_pair(client_id,
+                                          std::make_unique<dbms_client>(
+                                              *this, client_id,
+                                              trrep::client_context::m_replicating))).first->second.get();
+}
+
+
+void dbms_simulator::start()
+{
+    std::cout << "Provider: " << wsrep_provider_ << "\n";
+    for (size_t i(0); i < n_servers_; ++i)
+    {
+        std::ostringstream name_os;
+        name_os << (i + 1);
+        std::ostringstream id_os;
+        id_os << (i + 1);
+        auto it(servers_.insert(std::make_pair((i + 1),
+                                               std::make_unique<dbms_server>(
+                                                   name_os.str(), id_os.str()))));
+        it.first->second->load_provider(wsrep_provider_, wsrep_provider_options_);
+    }
+}
+
+namespace po = boost::program_options;
+
+int main(int argc, char** argv)
+{
+    try
+    {
+        size_t n_servers;
+        size_t n_clients;
+        std::string wsrep_provider;
+        std::string wsrep_provider_options;
+        po::options_description desc("Allowed options");
+        desc.add_options()
+            ("help", "produce help message")
+            ("wsrep-provider",
+             po::value<std::string>(&wsrep_provider)->required(),
+             "wsrep provider to load")
+            ("wsrep-provider-options",
+             po::value<std::string>(&wsrep_provider_options),
+             "wsrep provider options")
+            ("servers", po::value<size_t>(&n_servers)->required(),
+             "number of servers to start")
+            ("clients", po::value<size_t>(&n_clients)->required(),
+             "number of clients to start per server");
+        po::variables_map vm;
+        po::store(po::parse_command_line(argc, argv, desc), vm);
+        po::notify(vm);
+
+        if (vm.count("help"))
+        {
+            std::cout << desc << "\n";
+            return 1;
+        }
+
+        dbms_simulator sim(n_servers, n_clients, wsrep_provider, wsrep_provider_options);
+        sim.start();
+    }
+    catch (const std::exception& e)
+    {
+        std::cerr << e.what() << "\n";
+        return 1;
+    }
+
+    return 0;
+}

src/server_context.cpp

@@ -60,17 +60,40 @@ namespace
     }
 }
 
-int trrep::server_context::load_provider(const std::string& provider_spec)
+int trrep::server_context::load_provider(const std::string& provider_spec,
+                                         const std::string& provider_options)
 {
+    std::cout << "Loading provider " << provider_spec << "\n";
     if (provider_spec == "mock")
     {
         provider_ = new trrep::mock_provider;
     }
     else
     {
+        std::cerr << "Provider options" << provider_options << "\n";
         struct wsrep_init_args init_args;
+        memset(&init_args, 0, sizeof(init_args));
+        init_args.app_ctx = this;
+        init_args.node_name = name_.c_str();
+        init_args.node_address = "";
+        init_args.node_incoming = "";
+        init_args.data_dir = "./";
+        init_args.options = provider_options.c_str();
+        init_args.proto_ver = 1;
+        init_args.state_id = 0;
+        init_args.state = 0;
+        init_args.logger_cb = 0;
+        init_args.connected_cb = 0;
+        init_args.view_cb = 0;
+        init_args.sst_request_cb = 0;
         init_args.apply_cb = &apply_cb;
-        provider_ = new trrep::wsrep_provider_v26(&init_args);
+        init_args.unordered_cb = 0;
+        init_args.sst_donate_cb = 0;
+        init_args.synced_cb = 0;
+
+        std::cerr << init_args.options << "\n";
+        provider_ = new trrep::wsrep_provider_v26(provider_spec.c_str(),
+                                                  &init_args);
     }
     return 0;
 }

src/server_context.hpp

@@ -62,7 +62,7 @@ namespace trrep
         //
         // @return Zero on success, non-zero on error
         //
-        int load_provider(const std::string&);
+        int load_provider(const std::string&, const std::string& = "");
 
         //
         // Return reference to provider

src/wsrep_provider_v26.cpp

@@ -0,0 +1,23 @@
+//
+// Copyright (C) 2018 Codership Oy <info@codership.com>
+//
+
+#include "wsrep_provider_v26.hpp"
+#include "exception.hpp"
+
+#include <wsrep_api.h>
+
+trrep::wsrep_provider_v26::wsrep_provider_v26(
+    const char* path,
+    struct wsrep_init_args* args)
+    : wsrep_()
+{
+    if (wsrep_load(path, &wsrep_, 0))
+    {
+        throw trrep::runtime_error("Failed to load wsrep library");
+    }
+    if (wsrep_->init(wsrep_, args) != WSREP_OK)
+    {
+        throw trrep::runtime_error("Failed to initialize wsrep provider");
+    }
+}

src/wsrep_provider_v26.hpp

@@ -5,7 +5,7 @@
 #ifndef TRREP_WSREP_PROVIDER_V26_HPP
 #define TRREP_WSREP_PROVIDER_V26_HPP
 
-#include "provider_impl.hpp"
+#include "provider.hpp"
 
 #include <wsrep_api.h>
 
@@ -15,8 +15,8 @@ namespace trrep
     {
     public:
 
-        wsrep_provider_v26(struct wsrep_init_args*)
+        wsrep_provider_v26(const char*, struct wsrep_init_args*);
-        { }
+
         int start_transaction(wsrep_ws_handle_t*) { return 0; }
         int append_key(wsrep_ws_handle_t*, const wsrep_key_t*) { return 0; }
         int append_data(wsrep_ws_handle_t*, const wsrep_buf_t*) { return 0; }
@@ -31,7 +31,8 @@ namespace trrep
         wsrep_status commit_order_enter(wsrep_ws_handle_t*) { return WSREP_OK; }
         int commit_order_leave(wsrep_ws_handle_t*) { return 0; }
         int release(wsrep_ws_handle_t*) { return 0; }
-
+    private:
+        struct wsrep* wsrep_;
     };
 }