Add udp backend

libcaf_net/CMakeLists.txt

@@ -52,6 +52,7 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
   src/multiplexer.cpp
   src/net/backend/test.cpp
   src/net/backend/tcp.cpp
+  src/net/backend/udp.cpp
   src/net/endpoint_manager_queue.cpp
   src/net/middleman.cpp
   src/net/middleman_backend.cpp
@@ -149,4 +150,5 @@ caf_incubator_add_test_suites(caf-net-test
   udp_datagram_socket
   network_socket
   net.backend.tcp
+  net.backend.udp
 )

libcaf_net/caf/net/backend/udp.hpp

@@ -0,0 +1,86 @@
+/******************************************************************************
+ *                       ____    _    _____                                   *
+ *                      / ___|  / \  |  ___|    C++                           *
+ *                     | |     / _ \ | |_       Actor                         *
+ *                     | |___ / ___ \|  _|      Framework                     *
+ *                      \____/_/   \_|_|                                      *
+ *                                                                            *
+ * Copyright 2011-2020 Dominik Charousset                                     *
+ *                                                                            *
+ * Distributed under the terms and conditions of the BSD 3-Clause License or  *
+ * (at your option) under the terms and conditions of the Boost Software      *
+ * License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE.       *
+ *                                                                            *
+ * If you did not receive a copy of the license files, see                    *
+ * http://opensource.org/licenses/BSD-3-Clause and                            *
+ * http://www.boost.org/LICENSE_1_0.txt.                                      *
+ ******************************************************************************/
+
+#pragma once
+
+#include <map>
+#include <mutex>
+
+#include "caf/detail/net_export.hpp"
+#include "caf/error.hpp"
+#include "caf/expected.hpp"
+#include "caf/net/basp/application.hpp"
+#include "caf/net/fwd.hpp"
+#include "caf/net/make_endpoint_manager.hpp"
+#include "caf/net/middleman.hpp"
+#include "caf/net/middleman_backend.hpp"
+#include "caf/net/multiplexer.hpp"
+#include "caf/net/udp_datagram_socket.hpp"
+#include "caf/node_id.hpp"
+
+namespace caf::net::backend {
+
+/// Minimal backend for udp communication.
+class CAF_NET_EXPORT udp : public middleman_backend {
+public:
+  // -- constructors, destructors, and assignment operators --------------------
+
+  udp(middleman& mm);
+
+  ~udp() override;
+
+  // -- interface functions ----------------------------------------------------
+
+  error init() override;
+
+  void stop() override;
+
+  expected<endpoint_manager_ptr> get_or_connect(const uri& locator) override;
+
+  endpoint_manager_ptr peer(const node_id&) override;
+
+  void resolve(const uri& locator, const actor& listener) override;
+
+  strong_actor_ptr make_proxy(node_id nid, actor_id aid) override;
+
+  void set_last_hop(node_id*) override;
+
+  // -- properties -------------------------------------------------------------
+
+  uint16_t port() const noexcept override {
+    return listening_port_;
+  }
+
+  expected<endpoint_manager_ptr> emplace(udp_datagram_socket sock,
+                                         uint16_t port);
+
+private:
+  middleman& mm_;
+
+  endpoint_manager_ptr ep_manager_;
+
+  std::vector<node_id> node_ids_;
+
+  proxy_registry proxies_;
+
+  uint16_t listening_port_;
+
+  std::mutex lock_;
+};
+
+} // namespace caf::net::backend

libcaf_net/caf/net/basp/datagram_application_factory.hpp

@@ -0,0 +1,52 @@
+/******************************************************************************
+ *                       ____    _    _____                                   *
+ *                      / ___|  / \  |  ___|    C++                           *
+ *                     | |     / _ \ | |_       Actor                         *
+ *                     | |___ / ___ \|  _|      Framework                     *
+ *                      \____/_/   \_|_|                                      *
+ *                                                                            *
+ * Copyright 2011-2020 Dominik Charousset                                     *
+ *                                                                            *
+ * Distributed under the terms and conditions of the BSD 3-Clause License or  *
+ * (at your option) under the terms and conditions of the Boost Software      *
+ * License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE.       *
+ *                                                                            *
+ * If you did not receive a copy of the license files, see                    *
+ * http://opensource.org/licenses/BSD-3-Clause and                            *
+ * http://www.boost.org/LICENSE_1_0.txt.                                      *
+ ******************************************************************************/
+
+#pragma once
+
+#include "caf/detail/net_export.hpp"
+#include "caf/error.hpp"
+#include "caf/net/basp/application.hpp"
+#include "caf/net/datagram_adaptor.hpp"
+#include "caf/proxy_registry.hpp"
+
+namespace caf::net::basp {
+
+/// Factory for datagram oriented basp::applications.
+/// @relates transport_worker_dispatcher
+class CAF_NET_EXPORT datagram_application_factory {
+public:
+  using application_type = datagram_adaptor<basp::application>;
+
+  datagram_application_factory(proxy_registry& proxies) : proxies_(proxies) {
+    // nop
+  }
+
+  template <class Parent>
+  error init(Parent&) {
+    return none;
+  }
+
+  application_type make() const {
+    return application_type{basp::application{proxies_}};
+  }
+
+private:
+  proxy_registry& proxies_;
+};
+
+} // namespace caf::net::basp

libcaf_net/caf/net/datagram_adaptor.hpp

@@ -0,0 +1,86 @@
+/******************************************************************************
+ *                       ____    _    _____                                   *
+ *                      / ___|  / \  |  ___|    C++                           *
+ *                     | |     / _ \ | |_       Actor                         *
+ *                     | |___ / ___ \|  _|      Framework                     *
+ *                      \____/_/   \_|_|                                      *
+ *                                                                            *
+ * Copyright 2011-2020 Dominik Charousset                                     *
+ *                                                                            *
+ * Distributed under the terms and conditions of the BSD 3-Clause License or  *
+ * (at your option) under the terms and conditions of the Boost Software      *
+ * License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE.       *
+ *                                                                            *
+ * If you did not receive a copy of the license files, see                    *
+ * http://opensource.org/licenses/BSD-3-Clause and                            *
+ * http://www.boost.org/LICENSE_1_0.txt.                                      *
+ ******************************************************************************/
+
+#pragma once
+
+#include "caf/byte_buffer.hpp"
+#include "caf/net/basp/header.hpp"
+
+namespace caf::net {
+
+/// Implements an adaption layer for datagram oriented transport protocols.
+template <class Application>
+class datagram_adaptor {
+public:
+  datagram_adaptor(Application application)
+    : application_(std::move(application)) {
+    // nop
+  }
+
+  template <class Parent>
+  error init(Parent& parent) {
+    return application_.init(parent);
+  }
+
+  template <class Parent>
+  error write_message(Parent& parent,
+                      std::unique_ptr<endpoint_manager_queue::message> msg) {
+    return application_.write_message(parent, std::move(msg));
+  }
+
+  template <class Parent>
+  error handle_data(Parent& parent, span<const byte> received) {
+    if (auto err = application_.handle_data(
+          parent, make_span(received.data(), basp::header_size)))
+      return err;
+    auto data = received.data() + basp::header_size;
+    auto size = received.size() - basp::header_size;
+    if (size > 0)
+      return application_.handle_data(parent, make_span(data, size));
+    return none;
+  }
+
+  template <class Parent>
+  void resolve(Parent& parent, string_view path, const actor& listener) {
+    application_.resolve(parent, path, listener);
+  }
+
+  template <class Parent>
+  void new_proxy(Parent& parent, actor_id id) {
+    application_.new_proxy(parent, id);
+  }
+
+  template <class Parent>
+  void local_actor_down(Parent& parent, actor_id id, error reason) {
+    application_.local_actor_down(parent, id, std::move(reason));
+  }
+
+  template <class Parent>
+  void timeout(Parent& parent, std::string tag, uint64_t id) {
+    application_.timeout(parent, std::move(tag), id);
+  }
+
+  void handle_error(sec error) {
+    application_.handle_error(error);
+  }
+
+private:
+  Application application_;
+};
+
+} // namespace caf::net

libcaf_net/caf/net/datagram_transport.hpp

@@ -61,6 +61,8 @@ class datagram_transport : public datagram_transport_base<Factory> {
 
   using buffer_cache_type = typename super::buffer_cache_type;
 
+  using dispatcher_type = transport_worker_dispatcher<Factory, ip_endpoint>;
+
   // -- constructors, destructors, and assignment operators --------------------
 
   datagram_transport(udp_datagram_socket handle, factory_type factory)

libcaf_net/caf/net/defaults.hpp

@@ -36,4 +36,7 @@ CAF_NET_EXPORT extern const size_t max_header_buffers;
 /// Port to listen on for tcp.
 CAF_NET_EXPORT extern const uint16_t tcp_port;
 
+/// Port to listen on for udp.
+CAF_NET_EXPORT extern const uint16_t udp_port;
+
 } // namespace caf::defaults::middleman

libcaf_net/caf/net/doorman.hpp

@@ -121,6 +121,10 @@ class doorman {
     CAF_LOG_ERROR("doorman encounterd error: " << err);
   }
 
+  error emplace(const uri&) {
+    return make_error(sec::runtime_error, "function not implemented");
+  }
+
 private:
   net::tcp_accept_socket acceptor_;
 

libcaf_net/caf/net/transport_worker_dispatcher.hpp

@@ -20,11 +20,14 @@
 
 #include <unordered_map>
 
+#include "caf/ip_endpoint.hpp"
 #include "caf/logger.hpp"
 #include "caf/net/endpoint_manager_queue.hpp"
 #include "caf/net/fwd.hpp"
+#include "caf/net/ip.hpp"
 #include "caf/net/packet_writer_decorator.hpp"
 #include "caf/net/transport_worker.hpp"
+#include "caf/node_id.hpp"
 #include "caf/sec.hpp"
 #include "caf/send.hpp"
 
@@ -65,9 +68,10 @@ class transport_worker_dispatcher {
   error handle_data(Parent& parent, span<const byte> data, id_type id) {
     if (auto worker = find_worker(id))
       return worker->handle_data(parent, data);
-    // TODO: Where to get node_id from here?
-    auto worker = add_new_worker(parent, node_id{}, id);
-    if (worker)
+    auto locator = make_uri("udp://" + to_string(id));
+    if (!locator)
+      return locator.error();
+    if (auto worker = add_new_worker(parent, make_node_id(*locator), id))
       return (*worker)->handle_data(parent, data);
     else
       return std::move(worker.error());
@@ -77,25 +81,33 @@ class transport_worker_dispatcher {
   void write_message(Parent& parent,
                      std::unique_ptr<endpoint_manager_queue::message> msg) {
     auto receiver = msg->receiver;
-    if (!receiver)
+    if (!receiver) {
+      CAF_LOG_ERROR("no receiver was specified");
       return;
+    }
     auto nid = receiver->node();
-    if (auto worker = find_worker(nid)) {
+    auto worker = find_worker(nid);
+    if (!worker)
+      CAF_LOG_ERROR("could not find worker for endpoint");
+    else
       worker->write_message(parent, std::move(msg));
-      return;
-    }
-    // TODO: where to get id_type from here?
-    if (auto worker = add_new_worker(parent, nid, id_type{}))
-      (*worker)->write_message(parent, std::move(msg));
   }
 
   template <class Parent>
   void resolve(Parent& parent, const uri& locator, const actor& listener) {
-    if (auto worker = find_worker(make_node_id(locator)))
-      worker->resolve(parent, locator.path(), listener);
-    else
+    if (auto auth = locator.authority_only()) {
+      if (auto worker = find_worker(make_node_id(*auth))) {
+        worker->resolve(parent, locator.path(), listener);
+      } else {
+        if (auto ret = emplace(parent, locator))
+          (*ret)->resolve(parent, locator.path(), listener);
+        else
+          anon_send(listener, ret.error());
+      }
+    } else {
       anon_send(listener,
-                make_error(sec::runtime_error, "could not resolve node"));
+                make_error(sec::runtime_error, "could not get authority"));
+    }
   }
 
   template <class Parent>
@@ -125,15 +137,29 @@ class transport_worker_dispatcher {
   }
 
   void handle_error(sec error) {
-    for (const auto& p : workers_by_id_) {
-      auto worker = p.second;
-      worker->handle_error(error);
+    for (const auto& p : workers_by_id_)
+      p.second->handle_error(error);
+  }
+
+  template <class Parent>
+  expected<worker_ptr> emplace(Parent& parent, const uri& locator) {
+    auto& auth = locator.authority();
+    ip_address addr;
+    if (auto hostname = get_if<std::string>(&auth.host)) {
+      auto addrs = ip::resolve(*hostname);
+      if (addrs.empty())
+        return sec::remote_lookup_failed;
+      addr = addrs.at(0);
+    } else {
+      addr = *get_if<ip_address>(&auth.host);
     }
+    return add_new_worker(parent, make_node_id(*locator.authority_only()),
+                          ip_endpoint{addr, auth.port});
   }
 
   template <class Parent>
-  expected<worker_ptr> add_new_worker(Parent& parent, node_id node,
-                                      id_type id) {
+  expected<worker_ptr>
+  add_new_worker(Parent& parent, node_id node, id_type id) {
     CAF_LOG_TRACE(CAF_ARG(node) << CAF_ARG(id));
     auto application = factory_.make();
     auto worker = std::make_shared<worker_type>(std::move(application), id);
@@ -156,10 +182,8 @@ class transport_worker_dispatcher {
   template <class Key>
   worker_ptr find_worker_impl(const std::unordered_map<Key, worker_ptr>& map,
                               const Key& key) {
-    if (map.count(key) == 0) {
-      CAF_LOG_DEBUG("could not find worker: " << CAF_ARG(key));
+    if (map.count(key) == 0)
       return nullptr;
-    }
     return map.at(key);
   }
 

libcaf_net/src/defaults.cpp

@@ -26,4 +26,6 @@ const size_t max_header_buffers = 10;
 
 const uint16_t tcp_port = 0;
 
+const uint16_t udp_port = 0;
+
 } // namespace caf::defaults::middleman