[FOLD] working but sloppy type erasure

Author: vinniefalco

example/server/main.cpp

@@ -85,19 +85,29 @@ int server_main( int argc, char* argv[] )
             http_proto::install_serializer_service(srv.services(), cfg);
         }
 
-        ports<executor_type> vp(srv.get_executor());
-
         // Create the routes
         // VFALCO this is a hacky WIP
         router_type rr0;
         rr0.get<https_redirect_responder>("/*splat");
         router_type rr;
         rr.get<file_responder>("/*splat", doc_root);
 
+        auto& vp = emplace_part<ports<executor_type>>(srv,
+            srv.get_executor(),
+            1,
+            num_workers,
+            of_type<worker_ssl<executor_type>>,
+            srv,
+            srv.get_executor(),
+            ssl_ctx,
+            rr);
+        vp.emplace(asio::ip::tcp::endpoint(addr, 443), reuse_addr);
+        vp.emplace(asio::ip::tcp::endpoint(addr, 5050), reuse_addr);
+
         //
         // Add the listening ports and workers
         //
-
+#if 0
         {
             // plain (no https) port that does https redirect
             //
@@ -128,6 +138,7 @@ int server_main( int argc, char* argv[] )
                 ssl_ctx,
                 rr);
         }
+#endif
 
         srv.run();
     }

example/server/worker_ssl.hpp

@@ -60,24 +60,19 @@ class worker_ssl : public http_responder<
     using stream_type = ssl_stream<socket_type>;
 
     template<
-        class Executor1,
         class Executor0,
         class = typename std::enable_if<std::is_constructible<Executor, Executor0>::value>::type
     >
     worker_ssl(
+        std::size_t index,
+        any_lambda<void(std::size_t)> do_idle,
         asio_server& srv,
-        asio::basic_socket_acceptor<
-            Protocol, Executor1>& acc,
         Executor0 const& ex,
         asio::ssl::context& ssl_ctx,
         router_type& rr)
         : http_responder<worker_ssl>(srv, rr)
-        , do_accept_(
-            [&acc](worker_ssl& self)
-            {
-                acc.async_accept( self.stream_.next_layer(), self.ep_,
-                    call_mf(&worker_ssl::on_accept, &self));
-            })
+        , index_(index)
+        , do_idle_(std::move(do_idle))
         , ssl_ctx_(ssl_ctx)
         , stream_(ex, ssl_ctx)
     {
@@ -89,28 +84,21 @@ class worker_ssl : public http_responder<
         return stream_.stream();
     }
 
-    /** Run the worker
-
-        The worker will continue to run until
-        @ref stop is called or the worker is destroyed.
-    */
-    void run()
-    {
-        do_accept();
-    }
-
-    /** Stop the worker
-
-        Any outstanding I/O is canceled.
+    /** Cancel all outstanding I/O
     */
-    void stop()
+    void cancel()
     {
         system::error_code ec;
         stream_.next_layer().cancel(ec);
     }
 
-    void do_accept()
+    template<class Executor1>
+    void do_accept(
+        asio::basic_socket_acceptor<Protocol, Executor1>& acc,
+        any_lambda<void()> do_accepted)
     {
+        do_accepted_ = do_accepted;
+
         // Clean up any previous connection.
         {
             system::error_code ec;
@@ -124,7 +112,8 @@ class worker_ssl : public http_responder<
             stream_ = stream_type(std::move(stream_.next_layer()), ssl_ctx_);
         }
 
-        do_accept_(*this);
+        acc.async_accept(stream_.next_layer(), ep_,
+            call_mf(&worker_ssl::on_accept, this));
     }
 
     /** Called when the connected session is ended
@@ -146,12 +135,14 @@ class worker_ssl : public http_responder<
         }
 
         LOG_DBG(this->sect_, this->id(), " ", s, ": ", ec.message());
-        return do_accept();
+        return do_idle();
     }
 
 private:
     void on_accept(system::error_code const& ec)
     {
+        do_accepted_();
+
         if(ec.failed())
             return do_failed("worker_ssl::on_accept", ec);
 
@@ -186,12 +177,19 @@ class worker_ssl : public http_responder<
         stream_.next_layer().shutdown(asio::socket_base::shutdown_both, ec);
         // error ignored
 
-        return do_accept();
+        return do_idle();
+    }
+
+    void do_idle()
+    {
+        do_idle_(index_);
     }
 
 private:
     // order of destruction matters here
-    any_lambda<void(worker_ssl&)> do_accept_;
+    std::size_t index_;
+    any_lambda<void(std::size_t)> do_idle_;
+    any_lambda<void()> do_accepted_;
     asio::ssl::context& ssl_ctx_;
     stream_type stream_;
     typename Protocol::endpoint ep_;

include/boost/http_io/server/ports.hpp

@@ -10,7 +10,11 @@
 #ifndef BOOST_HTTP_IO_SERVER_PORTS_HPP
 #define BOOST_HTTP_IO_SERVER_PORTS_HPP
 
+#include "logger.hpp"
 #include <boost/http_io/detail/config.hpp>
+#include <boost/http_io/server/call_mf.hpp>
+#include <boost/http_io/server/fixed_array.hpp>
+#include <boost/http_io/server/server.hpp>
 #include <boost/asio/basic_socket_acceptor.hpp>
 #include <boost/asio/basic_stream_socket.hpp>
 #include <boost/asio/dispatch.hpp>
@@ -27,6 +31,30 @@ class tcp;
 namespace boost {
 namespace http_io {
 
+template<class T>
+struct of_type_t
+{
+    using type = T;
+};
+
+template<class T>
+constexpr of_type_t<T> of_type;
+
+class ports_base
+{
+public:
+    struct entry
+    {
+        std::size_t need;   // number of accepts we need
+    };
+
+    struct api
+    {
+        std::size_t index;
+    };
+
+};
+
 /** A TCP/IP listening port
 
     This implements a listening port as a server service. An array of workers
@@ -40,48 +68,160 @@ namespace http_io {
 */
 template< class Executor, class Protocol = asio::ip::tcp >
 class ports
+    : public ports_base
+    , public server::part
 {
 public:
-    static constexpr std::size_t max_accept = 128;
-
     using executor_type = Executor;
-    using endpoint_type = typename Protocol::endpoint;
-    using acceptor_type =
-        asio::basic_socket_acceptor<Protocol, Executor>;
+    using protocol_type = Protocol;
+    using endpoint = typename Protocol::endpoint;
+    using acceptor = asio::basic_socket_acceptor<Protocol, Executor>;
 
-    struct entry
+    struct entry_impl : entry
     {
         template<class... Args>
-        explicit entry(Args&&... args)
+        explicit entry_impl(Args&&... args)
             : sock(std::forward<Args>(args)...)
         {
         }
 
-        acceptor_type sock;
-        std::size_t avail;      // free workers available
-        std::size_t accept;     // number currently accepting
-
+        acceptor sock;
     };
 
-    template<class Executor1
-        ,typename = std::enable_if<std::is_constructible<Executor, Executor1>::value>
+    /** Constructor
+
+        @param concurrency The number of threads calling io_context::run
+    */
+    template<
+        class Executor1,
+        class Worker,
+        class... Args,
+        class = typename std::enable_if<std::is_constructible<Executor, Executor1>::value>::type
     >
-    explicit
-    ports(Executor1 const& ex)
+    ports(
+        server&,
+        Executor1 const& ex,
+        unsigned concurrency,
+        std::size_t num_workers,
+        of_type_t<Worker>,
+        Args&&... args)
         : ex_(ex)
+        , concurrency_(concurrency)
+    {
+        fixed_array<Worker> vw(num_workers);
+        while(! vw.is_full())
+            vw.emplace_back(
+                vw.size() + 1,
+                [this](std::size_t i)
+                {
+                    asio::dispatch(ex_,
+                        [this, i]
+                        {
+                            do_idle(i);
+                        });
+                },
+                std::forward<Args>(args)...);
+        vw_ = std::move(vw);
+
+        // tracks linked list of idle workers
+        vi_.resize(num_workers);
+        for(std::size_t i = 0; i < vi_.size() - 1; ++i)
+            vi_[i] = i + 2;
+        vi_.back() = 0;
+
+        do_accept_ = &ports::do_worker_accept<Worker>;
+        do_stop_ = &ports::do_worker_stop<Worker>;
+    }
+
+    template<class Worker>
+    void do_worker_accept(std::size_t i, entry_impl& e)
     {
+        vw_.to_span<Worker>()[i - 1].do_accept(e.sock,
+            [this, &e]
+            {
+                asio::dispatch(ex_,
+                    [this, &e]
+                    {
+                        ++e.need;
+                        LOG_TRC(sect_, "need=", e.need);
+                    });
+            });
     }
 
-    template<class... Args>
-    entry& emplace(Args&&... args)
+    template<class Worker>
+    void do_worker_stop()
     {
-        v_.emplace_back(std::forward<Args>(args)...);
-        return v_.back();
+        for(auto& w : vw_.to_span<Worker>())
+            w.cancel();
+        for(auto& e : ve_)
+        {
+            system::error_code ec;
+            e.sock.cancel(ec);
+        }
+    }
+
+    void do_idle(std::size_t index)
+    {
+        vi_[index - 1] = idle_;
+        idle_ = index;
+        do_accept();
+    }
+
+    /** Add another port
+    */
+    void emplace(
+        endpoint const& ep,
+        bool reuse_addr = true)
+    {
+        ve_.emplace_back(ex_, ep, reuse_addr);
+        ve_.back().need = concurrency_;
     }
 
 private:
+    void run() override
+    {
+        asio::dispatch(ex_, call_mf(&ports::do_accept, this));
+    }
+
+    void stop() override
+    {
+        asio::dispatch(ex_, call_mf(&ports::do_stop, this));
+    }
+
+    void do_accept()
+    {
+        if(idle_ == 0) // no idle workers
+            return;
+        for(auto& e : ve_)
+        {
+            while(e.need > 0)
+            {
+                --e.need;
+                LOG_TRC(sect_, "need=", e.need);
+                auto const i = idle_;
+                idle_ = vi_[idle_ - 1];
+                (this->*do_accept_)(i, e);
+                if(idle_ == 0) // no idle workers
+                    return;
+            }
+        }
+    }
+
+    void do_stop()
+    {
+        (this->*do_stop_)();
+    }
+
+    section sect_;
     Executor ex_;
-    std::vector<entry> v_;
+    unsigned concurrency_;
+    std::vector<entry_impl> ve_;
+    any_fixed_array vw_;
+    std::vector<std::size_t> vi_;
+    std::size_t idle_ = 1;
+
+    void(ports::*do_accept_)(std::size_t, entry_impl&);
+    void(ports::*do_stop_)();
 };
 
 } // http_io

include/boost/http_io/server/workers.hpp

@@ -11,8 +11,8 @@
 #define BOOST_HTTP_IO_SERVER_WORKERS_HPP
 
 #include <boost/http_io/detail/config.hpp>
+#include <boost/http_io/server/fixed_array.hpp>
 #include <boost/http_io/server/server.hpp>
-#include "fixed_array.hpp"
 
 namespace boost {
 namespace http_io {