Commit 3fa7fdbc authored by jan.koester's avatar jan.koester
Browse files

t3wt

parent 5a8934aa

src/client.cpp

+11 −1
Original line number Diff line number Diff line
@@ -338,11 +338,14 @@ void client::store_stripe(uint64_t group_id, uint32_t stripe_index, 
        netplus::socketwait local_wait;  // thread-local wait instance



        try {

            std::cerr << "[PARITY] send_to_node[" << i << "]: connecting..." << std::endl;

            ensure_connected(i);

            if (!connections_[i].connected || !connections_[i].socket) {

                std::cerr << "[PARITY] send_to_node[" << i << "]: connect FAILED" << std::endl;

                pending[i].fail_reason = "connect failed (handshake timeout)";

                return;

            }

            std::cerr << "[PARITY] send_to_node[" << i << "]: connected, sending..." << std::endl;

            auto& conn = *connections_[i].socket;

            auto payload = protocol::make_store_block(group_id, block_index, bdata, blen);

            auto msg = protocol::encode_message(msg_type::STORE_BLOCK, payload);
@@ -356,6 +359,7 @@ void client::store_stripe(uint64_t group_id, uint32_t stripe_index, 
                    // Pump immediately to flush queued UDP datagrams

                    conn.pumpNetwork(MSG_DONTWAIT);

                    pending[i].sent = true;

                    std::cerr << "[PARITY] send_to_node[" << i << "]: sent OK" << std::endl;

                    break;

                }

                conn.pumpNetwork(MSG_DONTWAIT);
@@ -391,13 +395,17 @@ void client::store_stripe(uint64_t group_id, uint32_t stripe_index, 
                send_to_node(i);

            }

        }

        std::cerr << "[PARITY] store_stripe: Phase 1 launching " << n << " async tasks" << std::endl;

        // Remote nodes in parallel

        std::vector<std::future<void>> futures;

        for (size_t i = 0; i < n; i++) {

            if (static_cast<int>(i) == local_node_index_ && local_store_) continue;

            futures.push_back(std::async(std::launch::async, send_to_node, i));

        }

        for (auto& f : futures) f.get();

        for (size_t fi = 0; fi < futures.size(); ++fi) {

            futures[fi].get();

            std::cerr << "[PARITY] store_stripe: async task " << fi << " done" << std::endl;

        }

    }



    // Phase 1b: retry failed remote nodes once with a fresh connection
@@ -419,6 +427,8 @@ void client::store_stripe(uint64_t group_id, uint32_t stripe_index, 
    for (auto& p : pending) {

        if (p.sent && !p.local) ++remaining;

    }

    std::cerr << "[PARITY] store_stripe: Phase 2 collecting " << remaining << " responses" << std::endl;

    }



    // Pump all connections once — data may already be buffered from Phase 1

    for (auto& p : pending) {