Function test_queue_batch

common/test/test_lockfree.cpp:149–226

template <typename LSType, typename LRType, typename QType>
int test_queue_batch(const char *name, QType &queue) {
    std::vector<std::thread> senders, receivers;
    scnt.fill(0);
    rcnt.fill(0);
    auto begin = std::chrono::steady_clock::now();
    for (size_t i = 0; i < receiver_num; i++) {
        receivers.emplace_back([i, &queue] {
            photon::set_cpu_affinity(i);
            int buffer[32];
            size_t size;
            int amount = items_num / receiver_num;
            std::chrono::nanoseconds rspent(std::chrono::nanoseconds(0));
            for (size_t x = 0; x < items_num / receiver_num;) {
                auto tm = std::chrono::high_resolution_clock::now();
                LRType::lock(rlock);
                while (!(size = queue.pop_batch(buffer, std::min(32, amount)))) {
                    LRType::unlock(rlock);
                    CPUPause::pause();
                    LRType::lock(rlock);
                }
                LRType::unlock(rlock);
                if (x) rspent += (std::chrono::high_resolution_clock::now() - tm) / size;
                for (auto y: xrange(size)) {
                    rc[buffer[y]]++;
                    rcnt[i]++;
                }
                x += size;
                amount -= size;
            }
            LOG_DEBUG("` receiver done, ` ns per action", i,
                   rspent.count() / (items_num / receiver_num - 1));
        });
    }
    for (size_t i = 0; i < sender_num; i++) {
        senders.emplace_back([i, &queue] {
            photon::set_cpu_affinity(i);
            std::vector<int> vec;
            vec.resize(items_num / sender_num);
            for (size_t x = 0; x < items_num / sender_num; x++) {
                vec[x] = x;
            }
            size_t size;
            std::chrono::nanoseconds wspent{std::chrono::nanoseconds(0)};
            for (size_t x = 0; x < items_num / sender_num;) {
                auto tm = std::chrono::high_resolution_clock::now();
                LSType::lock(wlock);
                while (!(size = queue.push_batch(&vec[x], std::min(32UL, vec.size() - x)))) {
                    LSType::unlock(wlock);
                    CPUPause::pause();
                    LSType::lock(wlock);
                }
                LSType::unlock(wlock);
                wspent += (std::chrono::high_resolution_clock::now() - tm) / size;
                for (auto y = x; y < x + size; y++) {
                    sc[y] ++;
                    scnt[i] ++;
                }
                x += size;