Commit 4ad724fc authored by jan.koester's avatar jan.koester
Browse files

test

parent c7645920

src/client.cpp

+46 −14
Original line number Diff line number Diff line
@@ -345,19 +345,35 @@ void client::store_stripe(uint64_t group_id, uint32_t stripe_index, 
        }

    };



    // Phase 1: store local blocks + open streams and send to remote nodes

    // Phase 1: store local blocks + send to remote nodes in parallel

    {

        // Local node first (no thread needed)

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

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

                send_to_node(i);

            }

        }

        // 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();

    }



    // Phase 1b: retry failed remote nodes once with a fresh connection

    {

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

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

            if (!pending[i].sent && !pending[i].local) {

                pending[i].fail_reason.clear();

                connections_[i].socket.reset();

                connections_[i].connected = false;

            send_to_node(i);

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

            }

        }

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

    }



    // Phase 2: collect responses using waitReadMulti
@@ -514,8 +530,8 @@ bool client::retrieve_stripe(uint64_t group_id, uint32_t stripe_index, 
    std::vector<Pending> pending(n);

    size_t local_ok_count = 0;



    // Phase 1: send FETCH_BLOCK to all nodes (or fetch from local store)

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

    // Phase 1: send FETCH_BLOCK to all nodes (local first, remote in parallel)

    auto fetch_from_node = [&](size_t i) {

        uint32_t bidx = stripe_index * static_cast<uint32_t>(n) + static_cast<uint32_t>(i);

        pending[i].node_index = i;
@@ -523,13 +539,12 @@ bool client::retrieve_stripe(uint64_t group_id, uint32_t stripe_index, 
            pending[i].local = true;

            pending[i].ok = local_store_->fetch(group_id, bidx, pending[i].block)

                            && !pending[i].block.empty();

            if (pending[i].ok) ++local_ok_count;

            continue;

            return;

        }



        try {

            ensure_connected(i);

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

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

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

            auto payload = protocol::make_fetch_block(group_id, bidx);

            auto msg = protocol::encode_message(msg_type::FETCH_BLOCK, payload);
@@ -552,6 +567,23 @@ bool client::retrieve_stripe(uint64_t group_id, uint32_t stripe_index, 
            connections_[i].socket.reset();

            connections_[i].connected = false;

        }

    };



    // Local node first

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

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

            fetch_from_node(i);

            if (pending[i].ok) ++local_ok_count;

        }

    }

    // 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, fetch_from_node, i));

        }

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

    }



    // Phase 2: collect responses using waitReadMulti