Inside Bitcoin

1001    }
1002}
1003
1004HTTPServer::IOReadiness HTTPServer::GenerateWaitSockets() const
1005{
1006    IOReadiness io_readiness;
1007
1008    // If the server is already handling its max connected clients count,
1009    // don't bother checking the listening sockets for new inbound connections.
1010    // Leave them in the kernel's queue until space in the application opens
1011    // up (or the client times out on its own).
1012    if (GetConnectionsCount() < static_cast<size_t>(m_rpcmaxconnections)) {
1013        for (const auto& sock : m_listen) {
1014            io_readiness.events_per_sock.emplace(sock, Sock::Events{Sock::RecvEvent});
1015        }
1016    }
1017
1018    for (const auto& http_client : m_connected) {
1019        // Safely copy the shared pointer to the socket
1020        std::shared_ptr<Sock> sock{http_client->GetSock()};
1021
1022        // Event choice:
1023        //   1. ReadyToSend() (m_send_ready set) -> Send
1024        //      m_send_ready stays set while the send buffer still has data to
1025        //      drain, so we keep sending and do not Recv. This is also how the
1026        //      send-throttle applies backpressure: while the send buffer is
1027        //      full, TryReadRequest() holds a completed request back from a
1028        //      worker, so nothing new is read until send has drained.
1029        //   2. Else, m_req is incomplete and needs more data, or there is no
1030        //      m_req at all and the recv buffer is empty -> Recv
1031        //   3. Else (no parse in progress, leftover bytes in m_recv_buffer) -> 0
1032        //      Stay in the I/O map so TryReadRequest() drains the buffer first.
1033        //      Extra pipelined data waits in the kernel socket buffer
1034        //      (TCP backpressure), not in m_recv_buffer.
1035        //
1036        // Lock-order safety: the convention established by
1037        // MaybeSendBytesFromBuffer() is to take m_send_mutex before m_sock_mutex.
1038        // In this loop GetSock() (above) takes m_sock_mutex and ReadyToSend()
1039        // (below) takes m_send_mutex; both are scoped, so each lock is released
1040        // before the next is taken and they stay separate critical sections.
1041        // Holding m_sock_mutex while acquiring m_send_mutex would invert that
1042        // order and risk a lock-order-inversion deadlock.
1043        Sock::Event event{0};
1044        if (http_client->ReadyToSend()) {
1045            event = Sock::SendEvent;
1046        } else if (http_client->GetRequest() != nullptr || http_client->ReceiveBufferEmpty()) {
1047            // Mid-parse (need more bytes) or buffer empty.
1048            event = Sock::RecvEvent;
1049        }
1050
1051        io_readiness.events_per_sock.emplace(sock, Sock::Events{event});
1052        io_readiness.httpclients_per_sock.emplace(sock, http_client);
1053    }
1054
1055    return io_readiness;
1056}
1057
1058/// \anchor http
1059void HTTPServer::ThreadSocketHandler()
1060{
1061    while (!m_interrupt_net) {
1062        // Check for the readiness of the already connected sockets and the
1063        // listening sockets in one call ("readiness" as in poll(2) or
1064        // select(2)). If none are ready, wait for a short while and return
1065        // empty sets.
1066        auto io_readiness{GenerateWaitSockets()};
1067        if (io_readiness.events_per_sock.empty() ||
1068            // WaitMany() may as well be a static method, the context of the first Sock in the vector is not relevant.
1069            !io_readiness.events_per_sock.begin()->first->WaitMany(SELECT_TIMEOUT,
1070                                                                   io_readiness.events_per_sock)) {
1071            m_interrupt_net.sleep_for(SELECT_TIMEOUT);
1072        }
1073
1074        // Service (send/receive) each of the already connected sockets.
1075        SocketHandlerConnected(io_readiness);
1076
1077        // Accept new connections from listening sockets.
1078        SocketHandlerListening(io_readiness.events_per_sock);
1079
1080        // Disconnect any clients that have been flagged.
1081        DisconnectClients();
1082    }
1083}
1084
1085std::unique_ptr<HTTPRequest> HTTPRemoteClient::TryReadRequest(const std::shared_ptr<HTTPRemoteClient>& client)
1086{
1087    // If we are already handling a request from
1088    // this client, do nothing. We'll check again on the next I/O
1089    // loop iteration.
1090    if (client->m_req_busy) return nullptr;
1091
1092    if (!client->m_req) {
1093        client->m_req = std::make_unique<HTTPRequest>(client);
1094    }
1095
1096    try {
1097        // Read data from the buffer into the current request
1098        client->ReadRequest(*client->m_req);
1099    } catch (const ContentTooLargeError& e) {
1100        LogDebug(
1101            BCLog::HTTP,
1102            "HTTP request body too large from client %s (id=%llu): %s",
1103            client->m_origin,
1104            client->m_id,
1105            e.what());
1106
1107        WriteNoStoreErrorReply(*client->m_req, HTTP_CONTENT_TOO_LARGE);
1108        client->m_disconnect = true;
1109        return nullptr;
1110    } catch (const std::runtime_error& e) {
1111        LogDebug(
1112            BCLog::HTTP,
1113            "Error reading HTTP request from client %s (id=%llu): %s",
1114            client->m_origin,
1115            client->m_id,
1116            e.what());
1117
1118        // We failed to read a complete request from the buffer
1119        WriteNoStoreErrorReply(*client->m_req, HTTP_BAD_REQUEST);
1120        client->m_disconnect = true;
1121        return nullptr;
1122    }
1123
1124    // If the request is ready, hand it to a worker.
1125    if (client->m_req->GetState() == HTTPRequest::State::Complete) {
1126        // Unless this client's send buffer is full: in that case hold the
1127        // parsed request here instead of moving it to a worker. This prevents
1128        // the server from reading any more data from this client until they
1129        // drain their end of the socket, and prevents the server from packing
1130        // more responses into the send buffer.
1131        const size_t buffer_used{WITH_LOCK(
1132            client->m_send_mutex,
1133            return client->m_send_buffer.size();)};
1134        if (buffer_used > MAX_BODY_SIZE) return nullptr;
1135        LogDebug(
1136            BCLog::HTTP,
1137            "Received a %s request for %s from %s (id=%llu)",
1138            RequestMethodString(client->m_req->GetRequestMethod()),
1139            client->m_req->GetURI(),
1140            client->m_origin,
1141            client->m_id);
1142
1143        client->m_req_busy = true;
1144        return std::move(client->m_req);
1145    }
1146
1147    return nullptr;
1148}
1149
1150void HTTPServer::DisconnectClients()
1151{
1152    const auto now{Now<SteadySeconds>()};
1153    size_t erased = std::erase_if(m_connected,
1154                                  [&](auto& client) {
1155                                      return client->MaybeDisconnect(now,
1156                                                                     m_rpcservertimeout,
1157                                                                     /*disconnect_all=*/m_disconnect_all_clients);
1158                                  });
1159    if (erased > 0) {
1160        // Report back to the main thread
1161        m_connected_size.fetch_sub(erased, std::memory_order_relaxed);
1162    }
1163}
1164
1165bool HTTPRemoteClient::MaybeDisconnect(std::chrono::time_point<SteadyClock> now, std::chrono::seconds rpcservertimeout, bool disconnect_all)
1166{
1167    // First check for idle timeout. We reset the timer when we send and receive data,
1168    // but if the server is busy handling a request we should ignore the timeout until
1169    // the reply is sent. If we did erase the shared_ptr<HTTPRemoteClient> reference in m_connected
1170    // while the server is busy with a request, it might be prematurely dropped before
1171    // the response has been sent, or if the HTTPRequest was holding a temporary shared_ptr
1172    // client on a worker thread - it would keep the socket open even after "disconnecting".
1173    const bool is_idle{rpcservertimeout.count() > 0 &&
1174                       now - m_idle_since.load() > rpcservertimeout &&
1175                       !m_req_busy};
1176
1177    // Disconnect this client due to error, end of communication, or idle timeout.
1178    // May drop unsent data if we are closing due to error.
1179    if (m_disconnect || is_idle) {
1180        if (is_idle) {
1181            LogDebug(BCLog::HTTP,
1182                     "HTTP client idle timeout %s (id=%llu)",
1183                     m_origin,
1184                     m_id);
1185        }
1186    } else {
1187        // Disconnect this client because the server is shutting
1188        // down and we need to disconnect all clients...
1189        if (disconnect_all) {
1190            // ...unless we still have data for this client.
1191            if (m_connection_busy) {
1192                // There is still data for this healthy-connected client.
1193                // Continue the I/O loop until all data is sent or an error is encountered.
1194                return false;
1195            } else {
1196                // This is a healthy persistent connection (e.g. keep-alive)
1197                // but it's time to say goodbye.
1198                ;
1199            }
1200        } else {
1201            // No reason to disconnect.
1202            return false;
1203        }
1204    }
1205    // No reason NOT to disconnect, log and remove.
1206    LogDebug(BCLog::HTTP,
1207             "Disconnecting HTTP client %s (id=%llu)",
1208             m_origin,
1209             m_id);
1210    return true;
1211}
1212
1213void HTTPServer::ClearConnectedClients()
1214{
1215    Assume(!m_thread_socket_handler.joinable()); // must be called after JoinSocketsThreads()
1216    if (m_connected.empty()) return;
1217    LogWarning("Force-disconnecting %d HTTP client(s) that did not disconnect gracefully", m_connected.size());
1218    m_connected_size.fetch_sub(m_connected.size(), std::memory_order_relaxed);
1219    m_connected.clear();
1220}
1221
1222void HTTPRemoteClient::ReadRequest(HTTPRequest& req)
1223{
1224    if (m_recv_buffer.empty()) return;
1225
1226    LineReader reader(m_recv_buffer, MAX_HEADERS_SIZE);
1227
1228    try {
1229        switch (req.GetState()) {
1230        case HTTPRequest::State::Init:
1231            if (!req.LoadControlData(reader)) break;
1232            req.SetState(HTTPRequest::State::NeedsHeaders);
1233            [[fallthrough]];
1234
1235        case HTTPRequest::State::NeedsHeaders:
1236            if (!req.LoadHeaders(reader)) break;
1237            req.SetState(HTTPRequest::State::NeedsBody);
1238            [[fallthrough]];
1239
1240        case HTTPRequest::State::NeedsBody:
1241            if (!req.LoadBody(reader)) break;
1242            req.SetState(HTTPRequest::State::Complete);
1243            [[fallthrough]];
1244
1245        case HTTPRequest::State::Complete:
1246            break;
1247
1248        case HTTPRequest::State::Error:
1249            break;
1250        }