Commit 5a8934aa authored by jan.koester's avatar jan.koester
Browse files

test

parent a99b0196

src/client.cpp

+14 −8
Original line number Diff line number Diff line
@@ -57,13 +57,14 @@ void client::connect_to_node(size_t node_index) { 
    nc.socket = std::make_unique<netplus::quic>();

    nc.socket->connect(nodes_[node_index].address, nodes_[node_index].port, false);



    // Wait for handshake using event-driven wait (epoll/kqueue/IOCP/poll)

    // Wait for handshake using a local socketwait (thread-safe)

    netplus::socketwait local_wait;

    auto hs_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);

    while (int poll_ms = ms_left(hs_deadline)) {

        if (nc.socket->hasPendingWrite()) {

            if (!wait_.waitWrite(*nc.socket, poll_ms)) continue;

            if (!local_wait.waitWrite(*nc.socket, poll_ms)) continue;

        } else {

            if (!wait_.waitRead(*nc.socket, poll_ms)) continue;

            if (!local_wait.waitRead(*nc.socket, poll_ms)) continue;

        }

        nc.socket->pumpNetwork(MSG_DONTWAIT);

        if (nc.socket->getHandshakeDone()) {
@@ -141,6 +142,7 @@ void client::ensure_connected(size_t node_index) { 
std::vector<uint8_t> client::send_recv(size_t node_index,

                                         msg_type type,

                                         const std::vector<uint8_t>& payload) {

    netplus::socketwait local_wait;  // thread-safe local instance

    for (int attempt = 0; attempt < 2; ++attempt) {

        try {

            ensure_connected(node_index);
@@ -158,7 +160,7 @@ std::vector<uint8_t> client::send_recv(size_t node_index, 


        bool sent = false;

        while (int poll_ms = ms_left(sr_deadline)) {

            if (wait_.waitWrite(conn, poll_ms)) {

            if (local_wait.waitWrite(conn, poll_ms)) {

                conn.sendStreamData(stream, msg, true);

                conn.pumpNetwork(MSG_DONTWAIT);

                sent = true;
@@ -179,7 +181,7 @@ std::vector<uint8_t> client::send_recv(size_t node_index, 
        // Wait for response

        std::vector<uint8_t> response;

        while (int poll_ms = ms_left(sr_deadline)) {

            if (!wait_.waitRead(conn, poll_ms)) continue;

            if (!local_wait.waitRead(conn, poll_ms)) continue;

            conn.pumpNetwork(MSG_DONTWAIT);

            if (conn.hasStreamData(stream)) {

                uint8_t hdr[protocol::HEADER_SIZE];
@@ -197,7 +199,7 @@ std::vector<uint8_t> client::send_recv(size_t node_index, 
                while (total_read < resp_len) {

                    int body_ms = ms_left(sr_deadline);

                    if (body_ms <= 0) break;

                    if (!wait_.waitRead(conn, body_ms)) continue;

                    if (!local_wait.waitRead(conn, body_ms)) continue;

                    conn.pumpNetwork(MSG_DONTWAIT);

                    size_t rd = conn.recvStreamData(stream,

                                                     response.data() + protocol::HEADER_SIZE + total_read,
@@ -333,6 +335,8 @@ void client::store_stripe(uint64_t group_id, uint32_t stripe_index, 
            return;

        }



        netplus::socketwait local_wait;  // thread-local wait instance



        try {

            ensure_connected(i);

            if (!connections_[i].connected || !connections_[i].socket) {
@@ -347,7 +351,7 @@ void client::store_stripe(uint64_t group_id, uint32_t stripe_index, 


            // Non-blocking send

            while (int poll_ms = ms_left(stripe_deadline)) {

                if (wait_.waitWrite(conn, poll_ms)) {

                if (local_wait.waitWrite(conn, poll_ms)) {

                    conn.sendStreamData(stream, msg, true);

                    // Pump immediately to flush queued UDP datagrams

                    conn.pumpNetwork(MSG_DONTWAIT);
@@ -576,6 +580,8 @@ bool client::retrieve_stripe(uint64_t group_id, uint32_t stripe_index, 
            return;

        }



        netplus::socketwait local_wait;  // thread-local wait instance



        try {

            ensure_connected(i);

            if (!connections_[i].connected || !connections_[i].socket) return;
@@ -586,7 +592,7 @@ bool client::retrieve_stripe(uint64_t group_id, uint32_t stripe_index, 
            pending[i].stream = stream;



            while (int poll_ms = ms_left(retr_deadline)) {

                if (wait_.waitWrite(conn, poll_ms)) {

                if (local_wait.waitWrite(conn, poll_ms)) {

                    conn.sendStreamData(stream, msg, true);

                    conn.pumpNetwork(MSG_DONTWAIT);

                    pending[i].sent = true;