Merge pull request #286 from loki-project/dev

Merge Dev to Master

httpserver/command_line.cpp

@@ -70,6 +70,10 @@ void command_line_parser::parse_args(int argc, char* argv[]) {
         return;
     }
 
+    if (options_.testnet && !vm.count("lokid-rpc-port")) {
+        options_.lokid_rpc_port = 38157;
+    }
+
     if (!vm.count("ip") || !vm.count("port")) {
         throw std::runtime_error(
             "Invalid option: address and/or port missing.");

httpserver/command_line.h

@@ -7,7 +7,7 @@ namespace loki {
 
 struct command_line_options {
     uint16_t port;
-    uint16_t lokid_rpc_port = 22023;
+    uint16_t lokid_rpc_port = 22023; // Or 38157 if `testnet`
     bool force_start = false;
     bool print_version = false;
     bool print_help = false;

httpserver/http_connection.cpp

@@ -105,15 +105,15 @@ void LokidClient::make_lokid_request(boost::string_view method,
                                      const nlohmann::json& params,
                                      http_callback_t&& cb) const {
 
-    make_lokid_request(local_ip_, lokid_rpc_port_, method, params,
-                       std::move(cb));
+    make_custom_lokid_request(local_ip_, lokid_rpc_port_, method, params,
+                              std::move(cb));
 }
 
-void LokidClient::make_lokid_request(const std::string& daemon_ip,
-                                     const uint16_t daemon_port,
-                                     boost::string_view method,
-                                     const nlohmann::json& params,
-                                     http_callback_t&& cb) const {
+void LokidClient::make_custom_lokid_request(const std::string& daemon_ip,
+                                            const uint16_t daemon_port,
+                                            boost::string_view method,
+                                            const nlohmann::json& params,
+                                            http_callback_t&& cb) const {
 
     auto req = std::make_shared<request_t>();
 
@@ -242,8 +242,12 @@ void connection_t::do_handshake() {
 }
 
 void connection_t::on_handshake(boost::system::error_code ec) {
+
+    const auto sockfd = stream_.lowest_layer().native_handle();
+    LOKI_LOG(debug, "Open https socket: {}", sockfd);
+    get_net_stats().record_socket_open(sockfd);
     if (ec) {
-        LOKI_LOG(warn, "ssl handshake failed: {}", ec.message());
+        LOKI_LOG(warn, "ssl handshake failed: ec: {} ({})", ec.value(), ec.message());
         deadline_.cancel();
         return;
     }
@@ -433,7 +437,8 @@ void connection_t::process_blockchain_test_req(uint64_t,
 void connection_t::process_swarm_req(boost::string_view target) {
 
 #ifndef DISABLE_SNODE_SIGNATURE
-    if (!validate_snode_request()) {
+    // allow ping request as a quick workaround (and they are cheap)
+    if (!validate_snode_request() && (target != "/swarms/ping_test/v1")) {
         return;
     }
 #endif
@@ -527,6 +532,7 @@ void connection_t::process_swarm_req(boost::string_view target) {
         }
 
     } else if (target == "/swarms/ping_test/v1") {
+        LOKI_LOG(debug, "Received ping_test");
         response_.result(http::status::ok);
     } else if (target == "/swarms/push/v1") {
 
@@ -615,6 +621,12 @@ void connection_t::process_request() {
             response_.result(http::status::ok);
             write_response();
             ioc_.stop();
+        } else if (target == "/sleep") {
+            ioc_.post([]() {
+                LOKI_LOG(warn, "Sleeping for some time...");
+                std::this_thread::sleep_for(std::chrono::seconds(30));
+            });
+            response_.result(http::status::ok);
         }
 #endif
         else {
@@ -1164,10 +1176,14 @@ void connection_t::on_shutdown(boost::system::error_code ec) {
         // http://stackoverflow.com/questions/25587403/boost-asio-ssl-async-shutdown-always-finishes-with-an-error
         ec.assign(0, ec.category());
     }
-    if (ec)
+    if (ec) {
         LOKI_LOG(error, "Could not close ssl stream gracefully, ec: {}",
                  ec.message());
+    }
 
+    const auto sockfd = stream_.lowest_layer().native_handle();
+    LOKI_LOG(debug, "Close https socket: {}", sockfd);
+    get_net_stats().record_socket_close(sockfd);
     stream_.lowest_layer().close();
 }
 
@@ -1224,7 +1240,9 @@ HttpClientSession::HttpClientSession(boost::asio::io_context& ioc,
 
 void HttpClientSession::on_connect() {
 
-    LOKI_LOG(trace, "on connect");
+    const auto sockfd = socket_.native_handle();
+    LOKI_LOG(debug, "Open http socket: {}", sockfd);
+    get_net_stats().record_socket_open(sockfd);
     http::async_write(socket_, *req_,
                       std::bind(&HttpClientSession::on_write,
                                 shared_from_this(), std::placeholders::_1,
@@ -1283,11 +1301,21 @@ void HttpClientSession::start() {
         if (ec) {
             // We should make sure that we print the error a few levels above,
             // where we have more context
-            LOKI_LOG(
-                debug,
-                "[http client]: could not connect to {}:{}, message: {} ({})",
-                endpoint_.address().to_string(), endpoint_.port(), ec.message(),
-                ec.value());
+
+            if (ec == boost::system::errc::connection_refused) {
+                LOKI_LOG(debug,
+                         "[http client]: could not connect to {}:{}, message: "
+                         "{} ({})",
+                         endpoint_.address().to_string(), endpoint_.port(),
+                         ec.message(), ec.value());
+            } else {
+                LOKI_LOG(error,
+                         "[http client]: could not connect to {}:{}, message: "
+                         "{} ({})",
+                         endpoint_.address().to_string(), endpoint_.port(),
+                         ec.message(), ec.value());
+            }
+
             trigger_callback(SNodeError::NO_REACH, nullptr);
             return;
         }
@@ -1307,7 +1335,7 @@ void HttpClientSession::start() {
                 }
             } else {
                 LOKI_LOG(warn, "client socket timed out");
-                self->socket_.close();
+                self->clean_up();
             }
         });
 }
@@ -1320,20 +1348,20 @@ void HttpClientSession::trigger_callback(SNodeError error,
     deadline_timer_.cancel();
 }
 
-/// We execute callback (if haven't already) here to make sure it is called
-HttpClientSession::~HttpClientSession() {
+void HttpClientSession::clean_up() {
 
-    if (!used_callback_) {
-        // If we destroy the session before posting the callback,
-        // it must be due to some error
-        ioc_.post(std::bind(callback_,
-                            sn_response_t{SNodeError::ERROR_OTHER, nullptr}));
+    if (!needs_cleanup) {
+        // This can happen because the deadline timer
+        // triggered and cleaned up the connection already
+        LOKI_LOG(debug, "No need for cleanup");
+        return;
     }
 
-    get_net_stats().http_connections_out--;
+    needs_cleanup = false;
 
     if (!socket_.is_open()) {
-        LOKI_LOG(debug, "Socket is already closed");
+        /// This should never happen!
+        LOKI_LOG(critical, "Socket is already closed");
         return;
     }
 
@@ -1349,12 +1377,32 @@ HttpClientSession::~HttpClientSession() {
                  ec.message());
     }
 
+    const auto sockfd = socket_.native_handle();
     socket_.close(ec);
 
     if (ec) {
-        LOKI_LOG(error, "On close socket [{}: {}]", ec.value(), ec.message());
+        LOKI_LOG(error, "Closing socket {} failed [{}: {}]", sockfd, ec.value(), ec.message());
+    } else {
+        LOKI_LOG(debug, "Close http socket: {}", sockfd);
+        get_net_stats().record_socket_close(sockfd);
+    }
+}
+
+/// We execute callback (if haven't already) here to make sure it is called
+HttpClientSession::~HttpClientSession() {
+
+    if (!used_callback_) {
+        // If we destroy the session before posting the callback,
+        // it must be due to some error
+        ioc_.post(std::bind(callback_,
+                            sn_response_t{SNodeError::ERROR_OTHER, nullptr}));
     }
 
+    get_net_stats().http_connections_out--;
+
+    this->clean_up();
+
+
 }
 
 } // namespace loki

httpserver/http_connection.h

@@ -70,11 +70,11 @@ class LokidClient {
     void make_lokid_request(boost::string_view method,
                             const nlohmann::json& params,
                             http_callback_t&& cb) const;
-    void make_lokid_request(const std::string& daemon_ip,
-                            const uint16_t daemon_port,
-                            boost::string_view method,
-                            const nlohmann::json& params,
-                            http_callback_t&& cb) const;
+    void make_custom_lokid_request(const std::string& daemon_ip,
+                                   const uint16_t daemon_port,
+                                   boost::string_view method,
+                                   const nlohmann::json& params,
+                                   http_callback_t&& cb) const;
 };
 
 constexpr auto SESSION_TIME_LIMIT = std::chrono::seconds(30);
@@ -104,6 +104,7 @@ class HttpClientSession
     response_t res_;
 
     bool used_callback_ = false;
+    bool needs_cleanup = true;
 
     void on_connect();
 
@@ -114,6 +115,8 @@ class HttpClientSession
     void trigger_callback(SNodeError error,
                           std::shared_ptr<std::string>&& body);
 
+    void clean_up();
+
   public:
     // Resolver and socket require an io_context
     HttpClientSession(boost::asio::io_context& ioc, const tcp::endpoint& ep,

httpserver/https_client.cpp

@@ -79,6 +79,9 @@ HttpsClientSession::HttpsClientSession(
       server_pub_key_b32z(sn_pubkey_b32z) {
 
     get_net_stats().https_connections_out++;
+
+    static uint64_t connection_count = 0;
+    this->connection_idx = connection_count++;
 }
 
 void HttpsClientSession::start() {
@@ -128,7 +131,12 @@ void HttpsClientSession::start() {
 }
 
 void HttpsClientSession::on_connect() {
-    LOKI_LOG(trace, "on connect");
+    LOKI_LOG(trace, "on connect, connection idx: {}", this->connection_idx);
+
+    const auto sockfd = stream_.lowest_layer().native_handle();
+    LOKI_LOG(debug, "Open https socket: {}", sockfd);
+    get_net_stats().record_socket_open(sockfd);
+
     stream_.set_verify_mode(ssl::verify_none);
     stream_.set_verify_callback(
         [this](bool preverified, ssl::verify_context& ctx) -> bool {
@@ -260,6 +268,10 @@ void HttpsClientSession::on_shutdown(boost::system::error_code ec) {
                  ec.message());
     }
 
+    const auto sockfd = stream_.lowest_layer().native_handle();
+    LOKI_LOG(debug, "Close https socket: {}", sockfd);
+    get_net_stats().record_socket_close(sockfd);
+
     stream_.lowest_layer().close();
 
     // If we get here then the connection is closed gracefully

httpserver/https_client.h

@@ -14,6 +14,9 @@ void make_https_request(boost::asio::io_context& ioc, const std::string& ip,
 class HttpsClientSession
     : public std::enable_shared_from_this<HttpsClientSession> {
 
+    // For debugging purposes mostly
+    uint64_t connection_idx;
+
     using tcp = boost::asio::ip::tcp;
 
     boost::asio::io_context& ioc_;

httpserver/main.cpp

@@ -8,6 +8,7 @@
 #include "service_node.h"
 #include "swarm.h"
 #include "version.h"
+#include "utils.hpp"
 
 #include <boost/filesystem.hpp>
 #include <sodium.h>
@@ -119,6 +120,15 @@ int main(int argc, char* argv[]) {
         return EXIT_FAILURE;
     }
 
+    {
+        const auto fd_limit = util::get_fd_limit();
+        if (fd_limit != -1) {
+            LOKI_LOG(debug, "Open file descriptor limit: {}", fd_limit);
+        } else {
+            LOKI_LOG(debug, "Open descriptor limit: N/A");
+        }
+    }
+
     try {
 
         // ed25519 key

httpserver/net_stats.h

@@ -1,10 +1,33 @@
 #pragma once
 
+#include <set>
+#include "loki_logger.h"
+
 struct net_stats_t {
 
     uint32_t connections_in = 0;
     uint32_t http_connections_out = 0;
     uint32_t https_connections_out = 0;
+
+    std::set<int> open_fds;
+
+    void record_socket_open(int sockfd) {
+#ifdef INTEGRATION_TEST
+        if (open_fds.find(sockfd) != open_fds.end()) {
+            LOKI_LOG(critical, "Already recorded as open: {}!", sockfd);
+        }
+#endif
+        open_fds.insert(sockfd);
+    }
+
+    void record_socket_close(int sockfd) {
+#ifdef INTEGRATION_TEST
+        if (open_fds.find(sockfd) == open_fds.end()) {
+            LOKI_LOG(critical, "Socket is NOT recorded as open: {}", sockfd);
+        }
+#endif
+        open_fds.erase(sockfd);
+    }
 };
 
 inline net_stats_t& get_net_stats() {