Merge pull request #803 from clemahieu/asio_threading

Holding more shared_ptrs instead of raw ptrs

CMakeLists.txt

@@ -24,7 +24,7 @@
 
 cmake_minimum_required(VERSION 3.0)
 # XXX using 1.8.90 instead of 1.9.0-DEV
-project(nghttp2 VERSION 1.19.0)
+project(nghttp2 VERSION 1.19.90)
 
 # See versioning rule:
 #  http://www.gnu.org/software/libtool/manual/html_node/Updating-version-info.html

configure.ac

@@ -25,7 +25,7 @@ dnl Do not change user variables!
 dnl http://www.gnu.org/software/automake/manual/html_node/Flag-Variables-Ordering.html
 
 AC_PREREQ(2.61)
-AC_INIT([nghttp2], [1.19.0], [t-tujikawa@users.sourceforge.net])
+AC_INIT([nghttp2], [1.20.0-DEV], [t-tujikawa@users.sourceforge.net])
 AC_CONFIG_AUX_DIR([.])
 AC_CONFIG_MACRO_DIR([m4])
 AC_CONFIG_HEADERS([config.h])

doc/nghttpx.h2r

@@ -549,11 +549,11 @@ some cases where the error has occurred before reaching API endpoint
 
 The following section describes available API endpoints.
 
-PUT /api/v1beta1/backendconfig
+POST /api/v1beta1/backendconfig
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 This API replaces the current backend server settings with the
-requested ones.  The request method should be PUT, but POST is also
+requested ones.  The request method should be POST, but PUT is also
 acceptable.  The request body must be nghttpx configuration file
 format.  For configuration file format, see `FILES`_ section.  The
 line separator inside the request body must be single LF (0x0A).

examples/asio-sv.cc

@@ -36,12 +36,25 @@
 
 #include <iostream>
 #include <string>
+#include <thread>
 
 #include <nghttp2/asio_http2_server.h>
 
 using namespace nghttp2::asio_http2;
 using namespace nghttp2::asio_http2::server;
 
+namespace {
+void run_forever(boost::asio::io_service &io_service, size_t num_threads) {
+  std::vector<std::thread> ts;
+  for (size_t i = 0; i < num_threads; ++i) {
+    ts.emplace_back([&io_service]() { io_service.run(); });
+  }
+  for (auto &t : ts) {
+    t.join();
+  }
+}
+} // namespace
+
 int main(int argc, char *argv[]) {
   try {
     // Check command line arguments.
@@ -58,9 +71,9 @@ int main(int argc, char *argv[]) {
     std::string port = argv[2];
     std::size_t num_threads = std::stoi(argv[3]);
 
-    http2 server;
+    boost::asio::io_service io_service;
 
-    server.num_threads(num_threads);
+    http2 server(io_service);
 
     server.handle("/", [](const request &req, const response &res) {
       res.write_head(200, {{"foo", {"bar"}}});
@@ -136,11 +149,16 @@ int main(int argc, char *argv[]) {
       if (server.listen_and_serve(ec, tls, addr, port)) {
         std::cerr << "error: " << ec.message() << std::endl;
       }
+
+      run_forever(io_service, num_threads);
     } else {
       if (server.listen_and_serve(ec, addr, port)) {
         std::cerr << "error: " << ec.message() << std::endl;
       }
+
+      run_forever(io_service, num_threads);
     }
+
   } catch (std::exception &e) {
     std::cerr << "exception: " << e.what() << "\n";
   }

examples/asio-sv2.cc

@@ -43,12 +43,25 @@
 #endif // HAVE_FCNTL_H
 #include <iostream>
 #include <string>
+#include <thread>
 
 #include <nghttp2/asio_http2_server.h>
 
 using namespace nghttp2::asio_http2;
 using namespace nghttp2::asio_http2::server;
 
+namespace {
+void run_forever(boost::asio::io_service &io_service, size_t num_threads) {
+  std::vector<std::thread> ts;
+  for (size_t i = 0; i < num_threads; ++i) {
+    ts.emplace_back([&io_service]() { io_service.run(); });
+  }
+  for (auto &t : ts) {
+    t.join();
+  }
+}
+} // namespace
+
 int main(int argc, char *argv[]) {
   try {
     // Check command line arguments.
@@ -65,9 +78,9 @@ int main(int argc, char *argv[]) {
     std::size_t num_threads = std::stoi(argv[3]);
     std::string docroot = argv[4];
 
-    http2 server;
+    boost::asio::io_service io_service;
 
-    server.num_threads(num_threads);
+    http2 server(io_service);
 
     server.handle("/", [&docroot](const request &req, const response &res) {
       auto path = percent_decode(req.uri().path);
@@ -112,10 +125,14 @@ int main(int argc, char *argv[]) {
       if (server.listen_and_serve(ec, tls, addr, port)) {
         std::cerr << "error: " << ec.message() << std::endl;
       }
+
+      run_forever(io_service, num_threads);
     } else {
       if (server.listen_and_serve(ec, addr, port)) {
         std::cerr << "error: " << ec.message() << std::endl;
       }
+
+      run_forever(io_service, num_threads);
     }
   } catch (std::exception &e) {
     std::cerr << "exception: " << e.what() << "\n";

lib/includes/nghttp2/nghttp2.h

@@ -3566,7 +3566,7 @@ NGHTTP2_EXTERN int nghttp2_session_upgrade2(nghttp2_session *session,
  * Serializes the SETTINGS values |iv| in the |buf|.  The size of the
  * |buf| is specified by |buflen|.  The number of entries in the |iv|
  * array is given by |niv|.  The required space in |buf| for the |niv|
- * entries is ``8*niv`` bytes and if the given buffer is too small, an
+ * entries is ``6*niv`` bytes and if the given buffer is too small, an
  * error is returned.  This function is used mainly for creating a
  * SETTINGS payload to be sent with the ``HTTP2-Settings`` header
  * field in an HTTP Upgrade request.  The data written in |buf| is NOT

src/CMakeLists.txt

@@ -219,7 +219,6 @@ if(ENABLE_ASIO_LIB)
     ssl.cc
     timegm.c
     asio_common.cc
-    asio_io_service_pool.cc
     asio_server_http2.cc
     asio_server_http2_impl.cc
     asio_server.cc

src/HtmlParser.cc

@@ -55,6 +55,21 @@ StringRef get_attr(const xmlChar **attrs, const StringRef &name) {
 }
 } // namespace
 
+namespace {
+ResourceType
+get_resource_type_for_preload_as(const StringRef &attribute_value) {
+  if (util::strieq_l("image", attribute_value)) {
+    return REQ_IMG;
+  } else if (util::strieq_l("style", attribute_value)) {
+    return REQ_CSS;
+  } else if (util::strieq_l("script", attribute_value)) {
+    return REQ_UNBLOCK_JS;
+  } else {
+    return REQ_OTHERS;
+  }
+}
+} // namespace
+
 namespace {
 void add_link(ParserData *parser_data, const StringRef &uri,
               ResourceType res_type) {
@@ -88,6 +103,13 @@ void start_element_func(void *user_data, const xmlChar *src_name,
       add_link(parser_data, href_attr, REQ_OTHERS);
     } else if (util::strieq_l("stylesheet", rel_attr)) {
       add_link(parser_data, href_attr, REQ_CSS);
+    } else if (util::strieq_l("preload", rel_attr)) {
+      auto as_attr = get_attr(attrs, StringRef::from_lit("as"));
+      if (as_attr.empty()) {
+        return;
+      }
+      add_link(parser_data, href_attr,
+               get_resource_type_for_preload_as(as_attr));
     }
   } else if (util::strieq_l("img", name)) {
     auto src_attr = get_attr(attrs, StringRef::from_lit("src"));

src/Makefile.am

@@ -243,7 +243,6 @@ libnghttp2_asio_la_SOURCES = \
 	ssl_compat.h \
 	timegm.c timegm.h \
 	asio_common.cc asio_common.h \
-	asio_io_service_pool.cc asio_io_service_pool.h \
 	asio_server_http2.cc \
 	asio_server_http2_impl.cc asio_server_http2_impl.h \
 	asio_server.cc asio_server.h \

src/asio_client_session_impl.cc

@@ -69,17 +69,17 @@ void session_impl::start_resolve(const std::string &host,
                                  const std::string &service) {
   deadline_.expires_from_now(connect_timeout_);
 
-  auto self = this->shared_from_this();
+  auto self = shared_from_this();
 
   resolver_.async_resolve({host, service},
-                          [this, self](const boost::system::error_code &ec,
+                          [self](const boost::system::error_code &ec,
                                        tcp::resolver::iterator endpoint_it) {
                             if (ec) {
-                              not_connected(ec);
+                              self->not_connected(ec);
                               return;
                             }
 
-                            start_connect(endpoint_it);
+                            self->start_connect(endpoint_it);
                           });
 
   deadline_.async_wait(std::bind(&session_impl::handle_deadline, self));
@@ -597,38 +597,38 @@ void session_impl::do_read() {
 
   auto self = this->shared_from_this();
 
-  read_socket([this, self](const boost::system::error_code &ec,
+  read_socket([self](const boost::system::error_code &ec,
                            std::size_t bytes_transferred) {
     if (ec) {
-      if (!should_stop()) {
+      if (!self->should_stop()) {
-        call_error_cb(ec);
+        self->call_error_cb(ec);
       }
-      stop();
+      self->stop();
       return;
     }
 
     {
-      callback_guard cg(*this);
+      callback_guard cg(*self);
 
       auto rv =
-          nghttp2_session_mem_recv(session_, rb_.data(), bytes_transferred);
+          nghttp2_session_mem_recv(self->session_, self->rb_.data(), bytes_transferred);
 
       if (rv != static_cast<ssize_t>(bytes_transferred)) {
-        call_error_cb(make_error_code(
+        self->call_error_cb(make_error_code(
             static_cast<nghttp2_error>(rv < 0 ? rv : NGHTTP2_ERR_PROTO)));
-        stop();
+        self->stop();
         return;
       }
     }
 
-    do_write();
+    self->do_write();
 
-    if (should_stop()) {
+    if (self->should_stop()) {
-      stop();
+      self->stop();
       return;
     }
 
-    do_read();
+    self->do_read();
   });
 }
 
@@ -695,17 +695,17 @@ void session_impl::do_write() {
   auto self = this->shared_from_this();
 
   write_socket(
-      [this, self](const boost::system::error_code &ec, std::size_t n) {
+      [self](const boost::system::error_code &ec, std::size_t n) {
         if (ec) {
-          call_error_cb(ec);
+          self->call_error_cb(ec);
-          stop();
+          self->stop();
           return;
         }
 
-        wblen_ = 0;
+        self->wblen_ = 0;
-        writing_ = false;
+        self->writing_ = false;
 
-        do_write();
+        self->do_write();
       });
 }
 

src/asio_client_session_tcp_impl.cc

@@ -37,19 +37,20 @@ session_tcp_impl::session_tcp_impl(
 session_tcp_impl::~session_tcp_impl() {}
 
 void session_tcp_impl::start_connect(tcp::resolver::iterator endpoint_it) {
+  auto self = shared_from_this();
   boost::asio::async_connect(socket_, endpoint_it,
-                             [this](const boost::system::error_code &ec,
+                             [self](const boost::system::error_code &ec,
                                     tcp::resolver::iterator endpoint_it) {
-                               if (stopped()) {
+                               if (self->stopped()) {
                                  return;
                                }
 
                                if (ec) {
-                                 not_connected(ec);
+                                 self->not_connected(ec);
                                  return;
                                }
 
-                               connected(endpoint_it);
+                               self->connected(endpoint_it);
                              });
 }
 

src/asio_client_session_tls_impl.cc

@@ -43,37 +43,38 @@ session_tls_impl::session_tls_impl(
 session_tls_impl::~session_tls_impl() {}
 
 void session_tls_impl::start_connect(tcp::resolver::iterator endpoint_it) {
+  auto self = std::static_pointer_cast<session_tls_impl>(shared_from_this());
   boost::asio::async_connect(
-      socket(), endpoint_it, [this](const boost::system::error_code &ec,
+      socket(), endpoint_it, [self](const boost::system::error_code &ec,
                                     tcp::resolver::iterator endpoint_it) {
-        if (stopped()) {
+        if (self->stopped()) {
           return;
         }
 
         if (ec) {
-          not_connected(ec);
+          self->not_connected(ec);
           return;
         }
 
-        socket_.async_handshake(
+		self->socket_.async_handshake(
             boost::asio::ssl::stream_base::client,
-            [this, endpoint_it](const boost::system::error_code &ec) {
+            [self, endpoint_it](const boost::system::error_code &ec) {
-              if (stopped()) {
+              if (self->stopped()) {
                 return;
               }
 
               if (ec) {
-                not_connected(ec);
+                self->not_connected(ec);
                 return;
               }
 
-              if (!tls_h2_negotiated(socket_)) {
+              if (!tls_h2_negotiated(self->socket_)) {
-                not_connected(make_error_code(
+                self->not_connected(make_error_code(
                     NGHTTP2_ASIO_ERR_TLS_NO_APP_PROTO_NEGOTIATED));
                 return;
               }
 
-              connected(endpoint_it);
+              self->connected(endpoint_it);
             });
       });
 }

src/asio_io_service_pool.cc

@@ -1,102 +0,0 @@
-/*
- * nghttp2 - HTTP/2 C Library
- *
- * Copyright (c) 2014 Tatsuhiro Tsujikawa
- *
- * Permission is hereby granted, free of charge, to any person obtaining
- * a copy of this software and associated documentation files (the
- * "Software"), to deal in the Software without restriction, including
- * without limitation the rights to use, copy, modify, merge, publish,
- * distribute, sublicense, and/or sell copies of the Software, and to
- * permit persons to whom the Software is furnished to do so, subject to
- * the following conditions:
- *
- * The above copyright notice and this permission notice shall be
- * included in all copies or substantial portions of the Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
- * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
- * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
- * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
- * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
- * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
- * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
- */
-// We wrote this code based on the original code which has the
-// following license:
-//
-// io_service_pool.cpp
-// ~~~~~~~~~~~~~~~~~~~
-//
-// Copyright (c) 2003-2013 Christopher M. Kohlhoff (chris at kohlhoff dot com)
-//
-// Distributed under the Boost Software License, Version 1.0. (See accompanying
-// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
-//
-#include "asio_io_service_pool.h"
-
-namespace nghttp2 {
-
-namespace asio_http2 {
-
-io_service_pool::io_service_pool(std::size_t pool_size) : next_io_service_(0) {
-  if (pool_size == 0) {
-    throw std::runtime_error("io_service_pool size is 0");
-  }
-
-  // Give all the io_services work to do so that their run() functions will not
-  // exit until they are explicitly stopped.
-  for (std::size_t i = 0; i < pool_size; ++i) {
-    auto io_service = std::make_shared<boost::asio::io_service>();
-    auto work = std::make_shared<boost::asio::io_service::work>(*io_service);
-    io_services_.push_back(io_service);
-    work_.push_back(work);
-  }
-}
-
-void io_service_pool::run(bool asynchronous) {
-  // Create a pool of threads to run all of the io_services.
-  for (std::size_t i = 0; i < io_services_.size(); ++i) {
-    futures_.push_back(std::async(std::launch::async,
-                                  (size_t(boost::asio::io_service::*)(void)) &
-                                      boost::asio::io_service::run,
-                                  io_services_[i]));
-  }
-
-  if (!asynchronous) {
-    join();
-  }
-}
-
-void io_service_pool::join() {
-  // Wait for all threads in the pool to exit.
-  for (auto &fut : futures_) {
-    fut.get();
-  }
-}
-
-void io_service_pool::stop() {
-  // Explicitly stop all io_services.
-  for (auto &iosv : io_services_) {
-    iosv->stop();
-  }
-}
-
-boost::asio::io_service &io_service_pool::get_io_service() {
-  // Use a round-robin scheme to choose the next io_service to use.
-  auto &io_service = *io_services_[next_io_service_];
-  ++next_io_service_;
-  if (next_io_service_ == io_services_.size()) {
-    next_io_service_ = 0;
-  }
-  return io_service;
-}
-
-const std::vector<std::shared_ptr<boost::asio::io_service>> &
-io_service_pool::io_services() const {
-  return io_services_;
-}
-
-} // namespace asio_http2
-
-} // namespace nghttp2

src/asio_io_service_pool.h

@@ -1,95 +0,0 @@
-/*
- * nghttp2 - HTTP/2 C Library
- *
- * Copyright (c) 2014 Tatsuhiro Tsujikawa
- *
- * Permission is hereby granted, free of charge, to any person obtaining
- * a copy of this software and associated documentation files (the
- * "Software"), to deal in the Software without restriction, including
- * without limitation the rights to use, copy, modify, merge, publish,
- * distribute, sublicense, and/or sell copies of the Software, and to
- * permit persons to whom the Software is furnished to do so, subject to
- * the following conditions:
- *
- * The above copyright notice and this permission notice shall be
- * included in all copies or substantial portions of the Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
- * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
- * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
- * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
- * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
- * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
- * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
- */
-// We wrote this code based on the original code which has the
-// following license:
-//
-// io_service_pool.hpp
-// ~~~~~~~~~~~~~~~~~~~
-//
-// Copyright (c) 2003-2013 Christopher M. Kohlhoff (chris at kohlhoff dot com)
-//
-// Distributed under the Boost Software License, Version 1.0. (See accompanying
-// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
-//
-
-#ifndef ASIO_IO_SERVICE_POOL_H
-#define ASIO_IO_SERVICE_POOL_H
-
-#include "nghttp2_config.h"
-
-#include <vector>
-#include <memory>
-#include <future>
-
-#include <boost/noncopyable.hpp>
-#include <boost/thread.hpp>
-
-#include <nghttp2/asio_http2.h>
-
-namespace nghttp2 {
-
-namespace asio_http2 {
-
-/// A pool of io_service objects.
-class io_service_pool : private boost::noncopyable {
-public:
-  /// Construct the io_service pool.
-  explicit io_service_pool(std::size_t pool_size);
-
-  /// Run all io_service objects in the pool.
-  void run(bool asynchronous = false);
-
-  /// Stop all io_service objects in the pool.
-  void stop();
-
-  /// Join on all io_service objects in the pool.
-  void join();
-
-  /// Get an io_service to use.
-  boost::asio::io_service &get_io_service();
-
-  /// Get access to all io_service objects.
-  const std::vector<std::shared_ptr<boost::asio::io_service>> &
-  io_services() const;
-
-private:
-  /// The pool of io_services.
-  std::vector<std::shared_ptr<boost::asio::io_service>> io_services_;
-
-  /// The work that keeps the io_services running.
-  std::vector<std::shared_ptr<boost::asio::io_service::work>> work_;
-
-  /// The next io_service to use for a connection.
-  std::size_t next_io_service_;
-
-  /// Futures to all the io_service objects
-  std::vector<std::future<std::size_t>> futures_;
-};
-
-} // namespace asio_http2
-
-} // namespace nghttp2
-
-#endif // ASIO_IO_SERVICE_POOL_H

src/asio_server.cc

@@ -44,10 +44,10 @@ namespace nghttp2 {
 namespace asio_http2 {
 namespace server {
 
-server::server(std::size_t io_service_pool_size,
+server::server(boost::asio::io_service &service,
 			   const boost::posix_time::time_duration &tls_handshake_timeout,
                const boost::posix_time::time_duration &read_timeout)
-    : io_service_pool_(io_service_pool_size),
+    : service_(service),
 	  tls_handshake_timeout_(tls_handshake_timeout),
       read_timeout_(read_timeout) {}
 
@@ -70,8 +70,6 @@ server::listen_and_serve(boost::system::error_code &ec,
     }
   }
 
-  io_service_pool_.run(asynchronous);
-
   return ec;
 }
 
@@ -81,7 +79,7 @@ boost::system::error_code server::bind_and_listen(boost::system::error_code &ec,
                                                   int backlog) {
   // Open the acceptor with the option to reuse the address (i.e.
   // SO_REUSEADDR).
-  tcp::resolver resolver(io_service_pool_.get_io_service());
+  tcp::resolver resolver(service_);
   tcp::resolver::query query(address, port);
   auto it = resolver.resolve(query, ec);
   if (ec) {
@@ -90,7 +88,7 @@ boost::system::error_code server::bind_and_listen(boost::system::error_code &ec,
 
   for (; it != tcp::resolver::iterator(); ++it) {
     tcp::endpoint endpoint = *it;
-    auto acceptor = tcp::acceptor(io_service_pool_.get_io_service());
+    auto acceptor = tcp::acceptor(service_);
 
     if (acceptor.open(endpoint.protocol(), ec)) {
       continue;
@@ -126,7 +124,7 @@ void server::start_accept(boost::asio::ssl::context &tls_context,
                           tcp::acceptor &acceptor, serve_mux &mux) {
   auto new_connection = std::make_shared<connection<ssl_socket>>(
       mux, tls_handshake_timeout_, read_timeout_,
-      io_service_pool_.get_io_service(), tls_context);
+      service_, tls_context);
 
   acceptor.async_accept(
       new_connection->socket().lowest_layer(),
@@ -159,8 +157,7 @@ void server::start_accept(boost::asio::ssl::context &tls_context,
 
 void server::start_accept(tcp::acceptor &acceptor, serve_mux &mux) {
   auto new_connection = std::make_shared<connection<tcp::socket>>(
-      mux, tls_handshake_timeout_, read_timeout_,
+      mux, tls_handshake_timeout_, read_timeout_, service_);
-      io_service_pool_.get_io_service());
 
   acceptor.async_accept(
       new_connection->socket(), [this, &acceptor, &mux, new_connection](
@@ -170,25 +167,18 @@ void server::start_accept(tcp::acceptor &acceptor, serve_mux &mux) {
           new_connection->start_read_deadline();
           new_connection->start();
         }
-
+        if (acceptor.is_open()) {
           start_accept(acceptor, mux);
+        }
       });
 }
 
 void server::stop() {
-  io_service_pool_.stop();
   for (auto &acceptor : acceptors_) {
     acceptor.close();
   }
 }
 
-void server::join() { io_service_pool_.join(); }
-
-const std::vector<std::shared_ptr<boost::asio::io_service>> &
-server::io_services() const {
-  return io_service_pool_.io_services();
-}
-
 } // namespace server
 } // namespace asio_http2
 } // namespace nghttp2

src/asio_server.h

@@ -44,11 +44,10 @@
 #include <memory>
 
 #include <boost/noncopyable.hpp>
+#include <boost/asio/io_service.hpp>
 
 #include <nghttp2/asio_http2_server.h>
 
-#include "asio_io_service_pool.h"
-
 namespace nghttp2 {
 
 namespace asio_http2 {
@@ -63,7 +62,7 @@ using ssl_socket = boost::asio::ssl::stream<tcp::socket>;
 
 class server : private boost::noncopyable {
 public:
-  explicit server(std::size_t io_service_pool_size,
+  explicit server(boost::asio::io_service &service,
                   const boost::posix_time::time_duration &tls_handshake_timeout,
                   const boost::posix_time::time_duration &read_timeout);
 
@@ -91,10 +90,7 @@ private:
                                             const std::string &address,
                                             const std::string &port,
                                             int backlog);
-
+  boost::asio::io_service &service_;
-  /// The pool of io_service objects used to perform asynchronous
-  /// operations.
-  io_service_pool io_service_pool_;
 
   /// Acceptor used to listen for incoming connections.
   std::vector<tcp::acceptor> acceptors_;

src/asio_server_http2.cc

@@ -36,7 +36,7 @@ namespace asio_http2 {
 
 namespace server {
 
-http2::http2() : impl_(make_unique<http2_impl>()) {}
+http2::http2(boost::asio::io_service &service) : impl_(make_unique<http2_impl>(service)) {}
 
 http2::~http2() {}
 
@@ -65,8 +65,6 @@ boost::system::error_code http2::listen_and_serve(
   return impl_->listen_and_serve(ec, &tls_context, address, port, asynchronous);
 }
 
-void http2::num_threads(size_t num_threads) { impl_->num_threads(num_threads); }
-
 void http2::backlog(int backlog) { impl_->backlog(backlog); }
 
 void http2::tls_handshake_timeout(const boost::posix_time::time_duration &t) {
@@ -83,13 +81,6 @@ bool http2::handle(std::string pattern, request_cb cb) {
 
 void http2::stop() { impl_->stop(); }
 
-void http2::join() { return impl_->join(); }
-
-const std::vector<std::shared_ptr<boost::asio::io_service>> &
-http2::io_services() const {
-  return impl_->io_services();
-}
-
 } // namespace server
 
 } // namespace asio_http2

src/asio_server_http2_impl.cc

@@ -37,8 +37,8 @@ namespace asio_http2 {
 
 namespace server {
 
-http2_impl::http2_impl()
+http2_impl::http2_impl(boost::asio::io_service &service)
-    : num_threads_(1),
+    : service_(service),
       backlog_(-1),
       tls_handshake_timeout_(boost::posix_time::seconds(60)),
       read_timeout_(boost::posix_time::seconds(60)) {}
@@ -47,13 +47,11 @@ boost::system::error_code http2_impl::listen_and_serve(
     boost::system::error_code &ec, boost::asio::ssl::context *tls_context,
     const std::string &address, const std::string &port, bool asynchronous) {
   server_.reset(
-      new server(num_threads_, tls_handshake_timeout_, read_timeout_));
+      new server(service_, tls_handshake_timeout_, read_timeout_));
   return server_->listen_and_serve(ec, tls_context, address, port, backlog_,
                                    mux_, asynchronous);
 }
 
-void http2_impl::num_threads(size_t num_threads) { num_threads_ = num_threads; }
-
 void http2_impl::backlog(int backlog) { backlog_ = backlog; }
 
 void http2_impl::tls_handshake_timeout(
@@ -71,13 +69,6 @@ bool http2_impl::handle(std::string pattern, request_cb cb) {
 
 void http2_impl::stop() { return server_->stop(); }
 
-void http2_impl::join() { return server_->join(); }
-
-const std::vector<std::shared_ptr<boost::asio::io_service>> &
-http2_impl::io_services() const {
-  return server_->io_services();
-}
-
 } // namespace server
 
 } // namespace asio_http2

src/asio_server_http2_impl.h

@@ -41,7 +41,7 @@ class server;
 
 class http2_impl {
 public:
-  http2_impl();
+  http2_impl(boost::asio::io_service &service);
   boost::system::error_code listen_and_serve(
       boost::system::error_code &ec, boost::asio::ssl::context *tls_context,
       const std::string &address, const std::string &port, bool asynchronous);
@@ -51,13 +51,10 @@ public:
   void read_timeout(const boost::posix_time::time_duration &t);
   bool handle(std::string pattern, request_cb cb);
   void stop();
-  void join();
-  const std::vector<std::shared_ptr<boost::asio::io_service>> &
-  io_services() const;
 
 private:
   std::unique_ptr<server> server_;
-  std::size_t num_threads_;
+  boost::asio::io_service &service_;
   int backlog_;
   serve_mux mux_;
   boost::posix_time::time_duration tls_handshake_timeout_;

src/h2load.cc

@@ -269,9 +269,7 @@ void conn_timeout_cb(EV_P_ ev_timer *w, int revents) {
 
 namespace {
 bool check_stop_client_request_timeout(Client *client, ev_timer *w) {
-  auto nreq = client->req_todo - client->req_started;
+  if (client->req_left == 0 ||
-
-  if (nreq == 0 ||
       client->streams.size() >= client->session->max_concurrent_streams()) {
     // no more requests to make, stop timer
     ev_timer_stop(client->worker->loop, w);
@@ -330,6 +328,8 @@ Client::Client(uint32_t id, Worker *worker, size_t req_todo)
       reqidx(0),
       state(CLIENT_IDLE),
       req_todo(req_todo),
+      req_left(req_todo),
+      req_inflight(0),
       req_started(0),
       req_done(0),
       id(id),
@@ -467,16 +467,13 @@ int Client::try_again_or_fail() {
 
   if (new_connection_requested) {
     new_connection_requested = false;
-    if (req_started < req_todo) {
+    if (req_left) {
       // At the moment, we don't have a facility to re-start request
       // already in in-flight.  Make them fail.
-      auto req_abandoned = req_started - req_done;
+      worker->stats.req_failed += req_inflight;
+      worker->stats.req_error += req_inflight;
 
-      worker->stats.req_failed += req_abandoned;
+      req_inflight = 0;
-      worker->stats.req_error += req_abandoned;
-      worker->stats.req_done += req_abandoned;
-
-      req_done = req_started;
 
       // Keep using current address
       if (connect() == 0) {
@@ -528,16 +525,18 @@ void Client::disconnect() {
 }
 
 int Client::submit_request() {
-  ++worker->stats.req_started;
   if (session->submit_request() != 0) {
     return -1;
   }
 
+  ++worker->stats.req_started;
+  --req_left;
   ++req_started;
+  ++req_inflight;
 
   // if an active timeout is set and this is the last request to be submitted
   // on this connection, start the active timeout.
-  if (worker->config->conn_active_timeout > 0. && req_started >= req_todo) {
+  if (worker->config->conn_active_timeout > 0. && req_left == 0) {
     ev_timer_start(worker->loop, &conn_active_watcher);
   }
 
@@ -545,40 +544,36 @@ int Client::submit_request() {
 }
 
 void Client::process_timedout_streams() {
-  for (auto &req_stat : worker->stats.req_stats) {
+  for (auto &p : streams) {
+    auto &req_stat = p.second.req_stat;
     if (!req_stat.completed) {
       req_stat.stream_close_time = std::chrono::steady_clock::now();
     }
   }
 
-  auto req_timed_out = req_todo - req_done;
+  worker->stats.req_timedout += req_inflight;
-  worker->stats.req_timedout += req_timed_out;
 
   process_abandoned_streams();
 }
 
 void Client::process_abandoned_streams() {
-  auto req_abandoned = req_todo - req_done;
+  auto req_abandoned = req_inflight + req_left;
 
   worker->stats.req_failed += req_abandoned;
   worker->stats.req_error += req_abandoned;
-  worker->stats.req_done += req_abandoned;
 
-  req_done = req_todo;
+  req_inflight = 0;
+  req_left = 0;
 }
 
 void Client::process_request_failure() {
-  auto req_abandoned = req_todo - req_started;
+  worker->stats.req_failed += req_left;
+  worker->stats.req_error += req_left;
 
-  worker->stats.req_failed += req_abandoned;
+  req_left = 0;
-  worker->stats.req_error += req_abandoned;
-  worker->stats.req_done += req_abandoned;
 
-  req_done += req_abandoned;
+  if (req_inflight == 0) {
-
-  if (req_done == req_todo) {
     terminate_session();
-    return;
   }
 }
 
@@ -711,6 +706,9 @@ void Client::on_status_code(int32_t stream_id, uint16_t status) {
 }
 
 void Client::on_stream_close(int32_t stream_id, bool success, bool final) {
+  ++req_done;
+  --req_inflight;
+
   auto req_stat = get_req_stat(stream_id);
   if (!req_stat) {
     return;
@@ -741,24 +739,20 @@ void Client::on_stream_close(int32_t stream_id, bool success, bool final) {
   }
 
   ++worker->stats.req_done;
-  ++req_done;
 
   worker->report_progress();
   streams.erase(stream_id);
-  if (req_done == req_todo) {
+  if (req_left == 0 && req_inflight == 0) {
     terminate_session();
     return;
   }
 
-  if (!config.timing_script && !final) {
+  if (!config.timing_script && !final && req_left > 0 &&
-    if (req_started < req_todo) {
+      submit_request() != 0) {
-      if (submit_request() != 0) {
     process_request_failure();
-      }
     return;
   }
 }
-}
 
 RequestStat *Client::get_req_stat(int32_t stream_id) {
   auto it = streams.find(stream_id);
@@ -871,8 +865,7 @@ int Client::connection_made() {
   record_connect_time();
 
   if (!config.timing_script) {
-    auto nreq =
+    auto nreq = std::min(req_left, session->max_concurrent_streams());
-        std::min(req_todo - req_started, session->max_concurrent_streams());
     for (; nreq > 0; --nreq) {
       if (submit_request() != 0) {
         process_request_failure();

src/h2load.h

@@ -292,6 +292,11 @@ struct Client {
   ClientState state;
   // The number of requests this client has to issue.
   size_t req_todo;
+  // The number of requests left to issue
+  size_t req_left;
+  // The number of requests currently have started, but not abandoned
+  // or finished.
+  size_t req_inflight;
   // The number of requests this client has issued so far.
   size_t req_started;
   // The number of requests this client has done so far.

src/h2load_http1_session.cc

@@ -100,7 +100,7 @@ int htp_msg_completecb(http_parser *htp) {
     http_parser_pause(htp, 1);
     // Connection is going down.  If we have still request to do,
     // create new connection and keep on doing the job.
-    if (client->req_started < client->req_todo) {
+    if (client->req_left) {
       client->try_new_connection();
     }
 

src/includes/nghttp2/asio_http2_server.h

@@ -132,7 +132,7 @@ class http2_impl;
 
 class http2 {
 public:
-  http2();
+  http2(boost::asio::io_service &service);
   ~http2();
 
   http2(http2 &&other) noexcept;
@@ -190,10 +190,6 @@ public:
   // equivalent .- and ..-free URL.
   bool handle(std::string pattern, request_cb cb);
 
-  // Sets number of native threads to handle incoming HTTP request.
-  // It defaults to 1.
-  void num_threads(size_t num_threads);
-
   // Sets the maximum length to which the queue of pending
   // connections.
   void backlog(int backlog);
@@ -207,13 +203,6 @@ public:
   // Gracefully stop http2 server
   void stop();
 
-  // Join on http2 server and wait for it to fully stop
-  void join();
-
-  // Get access to the io_service objects.
-  const std::vector<std::shared_ptr<boost::asio::io_service>> &
-  io_services() const;
-
 private:
   std::unique_ptr<http2_impl> impl_;
 };

src/shrpx.cc

@@ -1602,12 +1602,12 @@ Connections:
               The  parameters are  delimited  by  ";".  The  available
               parameters       are:      "proto=<PROTO>",       "tls",
               "sni=<SNI_HOST>",         "fall=<N>",        "rise=<N>",
-              "affinity=<METHOD>", and "dns".   The parameter consists
+              "affinity=<METHOD>",  "dns",  and  "frontend-tls".   The
-              of keyword,  and optionally  followed by "="  and value.
+              parameter consists  of keyword, and  optionally followed
-              For example,  the parameter  "proto=h2" consists  of the
+              by "=" and value.  For example, the parameter "proto=h2"
-              keyword  "proto" and  value "h2".   The parameter  "tls"
+              consists  of the  keyword "proto"  and value  "h2".  The
-              consists  of  the  keyword "tls"  without  value.   Each
+              parameter "tls"  consists of  the keyword  "tls" without
-              parameter is described as follows.
+              value.  Each parameter is described as follows.
 
               The backend application protocol  can be specified using
               optional  "proto"   parameter,  and   in  the   form  of
@@ -1664,6 +1664,17 @@ Connections:
               backend   host   name   at  start   up,   or   reloading
               configuration is skipped.
 
+              If "frontend-tls" parameter is used, the matched backend
+              requires frontend TLS connection.   In other words, even
+              if pattern  is matched,  frontend connection is  not TLS
+              protected, the request is  forwarded to one of catch-all
+              backends.   For this  reason,  catch-all backend  cannot
+              have "frontend-tls" parameter.  If  at least one backend
+              has  "frontend-tls" parameter,  this feature  is enabled
+              for all backend servers  sharing the same <PATTERN>.  It
+              is  advised  to  set  "frontend-tls"  parameter  to  all
+              backends explicitly if this feature is desired.
+
               Since ";" and ":" are  used as delimiter, <PATTERN> must
               not  contain these  characters.  Since  ";" has  special
               meaning in shell, the option value must be quoted.

src/shrpx_client_handler.cc

@@ -1008,6 +1008,13 @@ ClientHandler::get_downstream_connection(Downstream *downstream) {
     CLOG(INFO, this) << "Downstream address group_idx: " << group_idx;
   }
 
+  if (groups[group_idx]->shared_addr->require_upstream_tls && !conn_.tls.ssl) {
+    CLOG(INFO, this) << "Downstream address group " << group_idx
+                     << " requires frontend TLS connection.  Send request to "
+                        "catch-all group.";
+    group_idx = catch_all;
+  }
+
   auto &group = groups[group_idx];
   auto &shared_addr = group->shared_addr;
 

src/shrpx_config.cc

@@ -732,6 +732,7 @@ struct DownstreamParams {
   shrpx_session_affinity affinity;
   bool tls;
   bool dns;
+  bool frontend_tls;
 };
 
 namespace {
@@ -807,6 +808,8 @@ int parse_downstream_params(DownstreamParams &out,
       }
     } else if (util::strieq_l("dns", param)) {
       out.dns = true;
+    } else if (util::strieq_l("frontend-tls", param)) {
+      out.frontend_tls = true;
     } else if (!param.empty()) {
       LOG(ERROR) << "backend: " << param << ": unknown keyword";
       return -1;
@@ -899,6 +902,11 @@ int parse_mapping(Config *config, DownstreamAddrConfig &addr,
         if (params.affinity != AFFINITY_NONE) {
           g.affinity = params.affinity;
         }
+        // If at least one backend requires frontend TLS connection,
+        // enable it for all backends sharing the same pattern.
+        if (params.frontend_tls) {
+          g.require_upstream_tls = true;
+        }
         g.addrs.push_back(addr);
         done = true;
         break;
@@ -913,6 +921,7 @@ int parse_mapping(Config *config, DownstreamAddrConfig &addr,
     auto &g = addr_groups.back();
     g.addrs.push_back(addr);
     g.affinity = params.affinity;
+    g.require_upstream_tls = params.frontend_tls;
 
     if (pattern[0] == '*') {
       // wildcard pattern
@@ -1199,6 +1208,7 @@ int read_tls_sct_from_dir(std::vector<uint8_t> &dst, const StringRef &opt,
 }
 } // namespace
 
+#if !LIBRESSL_IN_USE
 namespace {
 // Reads PSK secrets from path, and parses each line.  The result is
 // directly stored into config->tls.psk_secrets.  This function
@@ -1262,7 +1272,9 @@ int parse_psk_secrets(Config *config, const StringRef &path) {
   return 0;
 }
 } // namespace
+#endif // !LIBRESSL_IN_USE
 
+#if !LIBRESSL_IN_USE
 namespace {
 // Reads PSK secrets from path, and parses each line.  The result is
 // directly stored into config->tls.client.psk.  This function returns
@@ -1322,6 +1334,7 @@ int parse_client_psk_secrets(Config *config, const StringRef &path) {
   return 0;
 }
 } // namespace
+#endif // !LIBRESSL_IN_USE
 
 // generated by gennghttpxfun.py
 int option_lookup_token(const char *name, size_t namelen) {
@@ -3284,9 +3297,23 @@ int parse_config(Config *config, int optid, const StringRef &opt,
     return parse_duration(&config->conn.upstream.timeout.idle_read, opt,
                           optarg);
   case SHRPX_OPTID_PSK_SECRETS:
+#if !LIBRESSL_IN_USE
     return parse_psk_secrets(config, optarg);
+#else  // LIBRESSL_IN_USE
+    LOG(WARN)
+        << opt
+        << ": ignored because underlying TLS library does not support PSK";
+    return 0;
+#endif // LIBRESSL_IN_USE
   case SHRPX_OPTID_CLIENT_PSK_SECRETS:
+#if !LIBRESSL_IN_USE
     return parse_client_psk_secrets(config, optarg);
+#else  // LIBRESSL_IN_USE
+    LOG(WARN)
+        << opt
+        << ": ignored because underlying TLS library does not support PSK";
+    return 0;
+#endif // LIBRESSL_IN_USE
   case SHRPX_OPTID_CLIENT_NO_HTTP2_CIPHER_BLACK_LIST:
     config->tls.client.no_http2_cipher_black_list =
         util::strieq_l("yes", optarg);
@@ -3607,6 +3634,12 @@ int configure_downstream_group(Config *config, bool http2_proxy,
     return -1;
   }
 
+  if (addr_groups[catch_all_group].require_upstream_tls) {
+    LOG(FATAL)
+        << "backend: Catch-all backend cannot have frontend-tls parameter";
+    return -1;
+  }
+
   downstreamconf.addr_group_catch_all = catch_all_group;
 
   if (LOG_ENABLED(INFO)) {

src/shrpx_config.h

@@ -430,7 +430,9 @@ struct AffinityHash {
 
 struct DownstreamAddrGroupConfig {
   DownstreamAddrGroupConfig(const StringRef &pattern)
-      : pattern(pattern), affinity(AFFINITY_NONE) {}
+      : pattern(pattern),
+        affinity(AFFINITY_NONE),
+        require_upstream_tls(false) {}
 
   StringRef pattern;
   std::vector<DownstreamAddrConfig> addrs;
@@ -439,6 +441,8 @@ struct DownstreamAddrGroupConfig {
   std::vector<AffinityHash> affinity_hash;
   // Session affinity
   shrpx_session_affinity affinity;
+  // true if this group requires that client connection must be TLS.
+  bool require_upstream_tls;
 };
 
 struct TicketKey {

src/shrpx_https_upstream.cc

@@ -949,8 +949,9 @@ void HttpsUpstream::error_reply(unsigned int status_code) {
   output->append("\r\nServer: ");
   output->append(get_config()->http.server_name);
   output->append("\r\nContent-Length: ");
-  auto cl = util::utos(html.size());
+  std::array<uint8_t, NGHTTP2_MAX_UINT64_DIGITS> intbuf;
-  output->append(cl);
+  output->append(StringRef{std::begin(intbuf),
+                           util::utos(std::begin(intbuf), html.size())});
   output->append("\r\nDate: ");
   auto lgconf = log_config();
   lgconf->update_tstamp(std::chrono::system_clock::now());
@@ -1026,11 +1027,14 @@ int HttpsUpstream::on_downstream_header_complete(Downstream *downstream) {
   auto connect_method = req.method == HTTP_CONNECT;
 
   auto buf = downstream->get_response_buf();
+  std::array<uint8_t, NGHTTP2_MAX_UINT64_DIGITS> intbuf;
 
   buf->append("HTTP/");
-  buf->append(util::utos(req.http_major));
+  buf->append(StringRef{std::begin(intbuf),
+                        util::utos(std::begin(intbuf), req.http_major)});
   buf->append('.');
-  buf->append(util::utos(req.http_minor));
+  buf->append(StringRef{std::begin(intbuf),
+                        util::utos(std::begin(intbuf), req.http_minor)});
   buf->append(' ');
   buf->append(http2::stringify_status(balloc, resp.http_status));
   buf->append(' ');

src/shrpx_ssl.cc

@@ -525,6 +525,7 @@ int sct_parse_cb(SSL *ssl, unsigned int ext_type, const unsigned char *in,
 } // namespace
 #endif // !LIBRESSL_IN_USE && OPENSSL_VERSION_NUMBER >= 0x10002000L
 
+#if !LIBRESSL_IN_USE
 namespace {
 unsigned int psk_server_cb(SSL *ssl, const char *identity, unsigned char *psk,
                            unsigned int max_psk_len) {
@@ -548,7 +549,9 @@ unsigned int psk_server_cb(SSL *ssl, const char *identity, unsigned char *psk,
   return static_cast<unsigned int>(secret.size());
 }
 } // namespace
+#endif // !LIBRESSL_IN_USE
 
+#if !LIBRESSL_IN_USE
 namespace {
 unsigned int psk_client_cb(SSL *ssl, const char *hint, char *identity_out,
                            unsigned int max_identity_len, unsigned char *psk,
@@ -581,6 +584,7 @@ unsigned int psk_client_cb(SSL *ssl, const char *hint, char *identity_out,
   return (unsigned int)secret.size();
 }
 } // namespace
+#endif // !LIBRESSL_IN_USE
 
 struct TLSProtocol {
   StringRef name;
@@ -784,7 +788,9 @@ SSL_CTX *create_ssl_context(const char *private_key_file, const char *cert_file,
   }
 #endif // !LIBRESSL_IN_USE && OPENSSL_VERSION_NUMBER >= 0x10002000L
 
+#if !LIBRESSL_IN_USE
   SSL_CTX_set_psk_server_callback(ssl_ctx, psk_server_cb);
+#endif // !LIBRESSL_IN_USE
 
   auto tls_ctx_data = new TLSContextData();
   tls_ctx_data->cert_file = cert_file;
@@ -919,7 +925,9 @@ SSL_CTX *create_ssl_client_context(
 #endif // HAVE_NEVERBLEED
   }
 
+#if !LIBRESSL_IN_USE
   SSL_CTX_set_psk_client_callback(ssl_ctx, psk_client_cb);
+#endif // !LIBRESSL_IN_USE
 
   // NPN selection callback.  This is required to set SSL_CTX because
   // OpenSSL does not offer SSL_set_next_proto_select_cb.

src/shrpx_worker.cc

@@ -76,7 +76,8 @@ bool match_shared_downstream_addr(
     return false;
   }
 
-  if (lhs->affinity != rhs->affinity) {
+  if (lhs->affinity != rhs->affinity ||
+      lhs->require_upstream_tls != rhs->require_upstream_tls) {
     return false;
   }
 
@@ -191,6 +192,7 @@ void Worker::replace_downstream_config(
     shared_addr->addrs.resize(src.addrs.size());
     shared_addr->affinity = src.affinity;
     shared_addr->affinity_hash = src.affinity_hash;
+    shared_addr->require_upstream_tls = src.require_upstream_tls;
 
     size_t num_http1 = 0;
     size_t num_http2 = 0;

src/shrpx_worker.h

@@ -136,7 +136,8 @@ struct SharedDownstreamAddr {
         next{0},
         http1_pri{},
         http2_pri{},
-        affinity{AFFINITY_NONE} {}
+        affinity{AFFINITY_NONE},
+        require_upstream_tls{false} {}
 
   SharedDownstreamAddr(const SharedDownstreamAddr &) = delete;
   SharedDownstreamAddr(SharedDownstreamAddr &&) = delete;
@@ -149,7 +150,7 @@ struct SharedDownstreamAddr {
   // AFFINITY_IP.
   std::vector<AffinityHash> affinity_hash;
   // List of Http2Session which is not fully utilized (i.e., the
-  // server advertized maximum concurrency is not reached).  We will
+  // server advertised maximum concurrency is not reached).  We will
   // coalesce as much stream as possible in one Http2Session to fully
   // utilize TCP connection.
   //
@@ -171,6 +172,8 @@ struct SharedDownstreamAddr {
   WeightedPri http2_pri;
   // Session affinity
   shrpx_session_affinity affinity;
+  // true if this group requires that client connection must be TLS.
+  bool require_upstream_tls;
 };
 
 struct DownstreamAddrGroup {