Commit 66d39980 authored by jan.koester's avatar jan.koester
Browse files

tes

parent b48d513a
Loading
Loading
Loading
Loading
Loading
+12 −24
Changes for src/cluster.cpp: 12 added lines, 24 removed lines.
Original line number Diff line number Diff line
@@ -686,29 +686,24 @@ namespace authdb {
            // Don't try to push while critical
            if (critical_) continue;

            // Grab pending items
            PendingDataPush dp_item;
            bool have_dp = false;
            if (!data_push_queue_.empty()) {
                dp_item = std::move(data_push_queue_.front());
            // Drain ALL pending data pushes
            std::vector<PendingDataPush> data_batch;
            while (!data_push_queue_.empty()) {
                data_batch.push_back(std::move(data_push_queue_.front()));
                data_push_queue_.pop_front();
                have_dp = true;
            }

            PendingPush sess_item;
            bool have_sess = false;
            if (!push_queue_.empty()) {
                sess_item = std::move(push_queue_.front());
            // Drain ALL pending session pushes
            std::vector<PendingPush> sess_batch;
            while (!push_queue_.empty()) {
                sess_batch.push_back(std::move(push_queue_.front()));
                push_queue_.pop_front();
                have_sess = true;
            }

            lk.unlock();

            // Launch data push and session push in parallel
            std::future<void> data_future;
            if (have_dp) {
                data_future = std::async(std::launch::async, [this, dp = std::move(dp_item)]() mutable {
            // Push all data items sequentially (each is already a full domain buffer)
            for (auto &dp : data_batch) {
                try {
                    pclient_->store(dp.dgid, dp.buf.data(), dp.buf.size());
                } catch (const std::exception &e) {
@@ -716,12 +711,10 @@ namespace authdb {
                    std::lock_guard<std::mutex> pguard(push_mutex_);
                    data_push_queue_.push_back(std::move(dp));
                }
                });
            }

            std::future<void> sess_future;
            if (have_sess) {
                sess_future = std::async(std::launch::async, [this, item = std::move(sess_item)]() mutable {
            // Push all sessions sequentially (each needs its own group_id for lookup)
            for (auto &item : sess_batch) {
                try {
                    session_client_->store(item.sid_gid, item.data.data(), item.data.size());
                } catch (const std::exception &e) {
@@ -734,12 +727,7 @@ namespace authdb {
                                  << e.what() << std::endl;
                    }
                }
                });
            }

            // Wait for both to finish
            if (data_future.valid()) data_future.get();
            if (sess_future.valid()) sess_future.get();
        }
    }