Merge pull request #5518 from akashihi/keepalive

Keepalive

CHANGELOG.md

@@ -6,6 +6,7 @@
       - ADDED: new waypoints parameter to the `route` plugin, enabling silent waypoints [#5345](https://github.com/Project-OSRM/osrm-backend/pull/5345)
       - ADDED: data timestamp information in the response (saved in new file `.osrm.timestamp`). [#5115](https://github.com/Project-OSRM/osrm-backend/issues/5115)
       - ADDED: new API parameter - `snapping=any|default` to allow snapping to previously unsnappable edges [#5361](https://github.com/Project-OSRM/osrm-backend/pull/5361)
+      - ADDED: keepalive support to the osrm-routed HTTP server [#5518](https://github.com/Project-OSRM/osrm-backend/pull/5518)
     - Routing:
       - CHANGED: allow routing past `barrier=arch` [#5352](https://github.com/Project-OSRM/osrm-backend/pull/5352)
       - CHANGED: default car weight was reduced to 2000 kg. [#5371](https://github.com/Project-OSRM/osrm-backend/pull/5371)

docs/http.md

@@ -1,3 +1,8 @@
+# OSRM HTTP server
+
+Built-in HTTP server is a basic HTTP/1.0 server that supports 'keep-alive' extension. Persistent connections are limited to 512 requests per
+connection and allow no more then 5 seconds between requests.
+
 ## General options
 
 All OSRM HTTP requests use a common structure.

include/server/connection.hpp

@@ -52,11 +52,17 @@ class Connection : public std::enable_shared_from_this<Connection>
     /// Handle completion of a write operation.
     void handle_write(const boost::system::error_code &e);
 
+    /// Handle read timeout
+    void handle_timeout(boost::system::error_code);
+
+    void handle_shutdown();
+
     std::vector<char> compress_buffers(const std::vector<char> &uncompressed_data,
                                        const http::compression_type compression_type);
 
     boost::asio::io_service::strand strand;
     boost::asio::ip::tcp::socket TCP_socket;
+    boost::asio::deadline_timer timer;
     RequestHandler &request_handler;
     RequestParser request_parser;
     boost::array<char, 8192> incoming_data_buffer;
@@ -65,6 +71,10 @@ class Connection : public std::enable_shared_from_this<Connection>
     std::vector<char> compressed_output;
     // Header compression_header;
     std::vector<boost::asio::const_buffer> output_buffer;
+    // Keep alive support
+    bool keep_alive = false;
+    short processed_requests = 512;
+    short keepalive_timeout = 5; // In seconds
 };
 }
 }

include/server/http/request.hpp

@@ -17,6 +17,7 @@ struct request
     std::string uri;
     std::string referrer;
     std::string agent;
+    std::string connection;
     boost::asio::ip::address endpoint;
 };
 }

src/server/connection.cpp

@@ -2,6 +2,7 @@
 #include "server/request_handler.hpp"
 #include "server/request_parser.hpp"
 
+#include <boost/algorithm/string/predicate.hpp>
 #include <boost/assert.hpp>
 #include <boost/bind.hpp>
 #include <boost/iostreams/filter/gzip.hpp>
@@ -17,7 +18,7 @@ namespace server
 {
 
 Connection::Connection(boost::asio::io_service &io_service, RequestHandler &handler)
-    : strand(io_service), TCP_socket(io_service), request_handler(handler)
+    : strand(io_service), TCP_socket(io_service), timer(io_service), request_handler(handler)
 {
 }
 
@@ -32,6 +33,15 @@ void Connection::start()
                                 this->shared_from_this(),
                                 boost::asio::placeholders::error,
                                 boost::asio::placeholders::bytes_transferred)));
+
+    if (keep_alive)
+    {
+        // Ok, we know it is not a first request, as we switched to keepalive
+        timer.cancel();
+        timer.expires_from_now(boost::posix_time::seconds(keepalive_timeout));
+        timer.async_wait(std::bind(
+            &Connection::handle_timeout, this->shared_from_this(), std::placeholders::_1));
+    }
 }
 
 void Connection::handle_read(const boost::system::error_code &error, std::size_t bytes_transferred)
@@ -41,6 +51,12 @@ void Connection::handle_read(const boost::system::error_code &error, std::size_t
         return;
     }
 
+    if (keep_alive)
+    {
+        timer.cancel();
+        timer.expires_from_now(boost::posix_time::seconds(0));
+    }
+
     // no error detected, let's parse the request
     http::compression_type compression_type(http::no_compression);
     RequestParser::RequestStatus result;
@@ -55,6 +71,17 @@ void Connection::handle_read(const boost::system::error_code &error, std::size_t
         current_request.endpoint = TCP_socket.remote_endpoint().address();
         request_handler.HandleRequest(current_request, current_reply);
 
+        if (boost::iequals(current_request.connection, "close"))
+        {
+            current_reply.headers.emplace_back("Connection", "close");
+        }
+        else
+        {
+            keep_alive = true;
+            current_reply.headers.emplace_back("Connection", "keep-alive");
+            current_reply.headers.emplace_back("Keep-Alive", "timeout=5, max=512");
+        }
+
         // compress the result w/ gzip/deflate if requested
         switch (compression_type)
         {
@@ -116,12 +143,40 @@ void Connection::handle_write(const boost::system::error_code &error)
 {
     if (!error)
     {
-        // Initiate graceful connection closure.
+        if (keep_alive && processed_requests > 0)
-        boost::system::error_code ignore_error;
+        {
-        TCP_socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ignore_error);
+            --processed_requests;
+            current_request = http::request();
+            request_parser = RequestParser();
+            this->start();
+        }
+        else
+        {
+            handle_shutdown();
+        }
     }
 }
 
+/// Handle completion of a timeout timer..
+void Connection::handle_timeout(boost::system::error_code ec)
+{
+    // We can get there for 3 reasons: spurious wakeup by timer.cancel(), which should be ignored
+    // Slow client with a delayed _first_ request, which should be ignored too
+    // Absent next request during waiting time in the keepalive mode - should stop right there.
+    if (ec != boost::asio::error::operation_aborted)
+    {
+        TCP_socket.cancel();
+        handle_shutdown();
+    }
+}
+
+void Connection::handle_shutdown()
+{
+    // Initiate graceful connection closure.
+    boost::system::error_code ignore_error;
+    TCP_socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ignore_error);
+}
+
 std::vector<char> Connection::compress_buffers(const std::vector<char> &uncompressed_data,
                                                const http::compression_type compression_type)
 {

src/server/http/reply.cpp

@@ -103,11 +103,7 @@ boost::asio::const_buffer reply::status_to_buffer(const reply::status_type statu
     return boost::asio::buffer(http_bad_request_string);
 }
 
-reply::reply() : status(ok)
+reply::reply() : status(ok) {}
-{
-    // We do not currently support keep alive. Always set 'Connection: close'.
-    headers.emplace_back("Connection", "close");
-}
 }
 }
 }

src/server/request_parser.cpp

@@ -180,6 +180,11 @@ RequestParser::RequestStatus RequestParser::consume(http::request &current_reque
             current_request.agent = current_header.value;
         }
 
+        if (boost::iequals(current_header.name, "Connection"))
+        {
+            current_request.connection = current_header.value;
+        }
+
         if (input == '\r')
         {
             state = internal_state::expecting_newline_3;