protocol.cpp raw

   1  // Copyright (c) 2021 The Limenka developers
   2  // Distributed under the MIT software license, see the accompanying
   3  // file COPYING or http://www.opensource.org/licenses/mit-license.php.
   4  
   5  #include <interfaces/init.h>
   6  #include <ipc/capnp/context.h>
   7  #include <ipc/capnp/init.capnp.h>
   8  #include <ipc/capnp/init.capnp.proxy.h>
   9  #include <ipc/capnp/protocol.h>
  10  #include <ipc/exception.h>
  11  #include <ipc/protocol.h>
  12  #include <kj/async.h>
  13  #include <logging.h>
  14  #include <mp/proxy-io.h>
  15  #include <mp/proxy-types.h>
  16  #include <mp/util.h>
  17  #include <util/threadnames.h>
  18  
  19  #include <assert.h>
  20  #include <errno.h>
  21  #include <future>
  22  #include <memory>
  23  #include <mutex>
  24  #include <optional>
  25  #include <string>
  26  #include <sys/socket.h>
  27  #include <system_error>
  28  #include <thread>
  29  
  30  namespace ipc {
  31  namespace capnp {
  32  namespace {
  33  void IpcLogFn(bool raise, std::string message)
  34  {
  35      LogDebug(BCLog::IPC, "%s\n", message);
  36      if (raise) throw Exception(message);
  37  }
  38  
  39  class CapnpProtocol : public Protocol
  40  {
  41  public:
  42      ~CapnpProtocol() noexcept(true)
  43      {
  44          if (m_loop) {
  45              std::unique_lock<std::mutex> lock(m_loop->m_mutex);
  46              m_loop->removeClient(lock);
  47          }
  48          if (m_loop_thread.joinable()) m_loop_thread.join();
  49          assert(!m_loop);
  50      };
  51      std::unique_ptr<interfaces::Init> connect(int fd, const char* exe_name) override
  52      {
  53          startLoop(exe_name);
  54          return mp::ConnectStream<messages::Init>(*m_loop, fd);
  55      }
  56      void listen(int listen_fd, const char* exe_name, interfaces::Init& init) override
  57      {
  58          startLoop(exe_name);
  59          if (::listen(listen_fd, /*backlog=*/5) != 0) {
  60              throw std::system_error(errno, std::system_category());
  61          }
  62          mp::ListenConnections<messages::Init>(*m_loop, listen_fd, init);
  63      }
  64      void serve(int fd, const char* exe_name, interfaces::Init& init, const std::function<void()>& ready_fn = {}) override
  65      {
  66          assert(!m_loop);
  67          mp::g_thread_context.thread_name = mp::ThreadName(exe_name);
  68          m_loop.emplace(exe_name, &IpcLogFn, &m_context);
  69          if (ready_fn) ready_fn();
  70          mp::ServeStream<messages::Init>(*m_loop, fd, init);
  71          m_loop->loop();
  72          m_loop.reset();
  73      }
  74      void addCleanup(std::type_index type, void* iface, std::function<void()> cleanup) override
  75      {
  76          mp::ProxyTypeRegister::types().at(type)(iface).cleanup_fns.emplace_back(std::move(cleanup));
  77      }
  78      Context& context() override { return m_context; }
  79      void startLoop(const char* exe_name)
  80      {
  81          if (m_loop) return;
  82          std::promise<void> promise;
  83          m_loop_thread = std::thread([&] {
  84              util::ThreadRename("capnp-loop");
  85              m_loop.emplace(exe_name, &IpcLogFn, &m_context);
  86              {
  87                  std::unique_lock<std::mutex> lock(m_loop->m_mutex);
  88                  m_loop->addClient(lock);
  89              }
  90              promise.set_value();
  91              m_loop->loop();
  92              m_loop.reset();
  93          });
  94          promise.get_future().wait();
  95      }
  96      Context m_context;
  97      std::thread m_loop_thread;
  98      std::optional<mp::EventLoop> m_loop;
  99  };
 100  } // namespace
 101  
 102  std::unique_ptr<Protocol> MakeCapnpProtocol() { return std::make_unique<CapnpProtocol>(); }
 103  } // namespace capnp
 104  } // namespace ipc
 105