refactor atomic queue.

lichao

2021-05-06 9eb924f64a56918536d310dd89698aa7e148c727

refactor atomic queue.

 src/robust.h

@@ -312,9 +312,10 @@
        size_type i = 0;
        do {
            auto pos = tail();
            auto cur = buf[pos].load();
            r = Empty(cur) && buf[pos].compare_exchange_strong(cur, Enc(d));
            tail_.compare_exchange_strong(pos, Next(pos));
            if (tail_.compare_exchange_strong(pos, Next(pos))) {
                auto cur = buf[pos].load();
                r = Empty(cur) && buf[pos].compare_exchange_strong(cur, Enc(d));
            }
        } while (try_more && !r && ++i < capacity);
        return r;
    }
@@ -325,9 +326,10 @@
        size_type i = 0;
        do {
            auto pos = head();
            cur = buf[pos].load();
            r = !Empty(cur) && buf[pos].compare_exchange_strong(cur, 0);
            head_.compare_exchange_strong(pos, Next(pos));
            if (head_.compare_exchange_strong(pos, Next(pos))) {
                cur = buf[pos].load();
                r = !Empty(cur) && buf[pos].compare_exchange_strong(cur, 0);
            }
        } while (try_more && !r && ++i < capacity);
        if (r) { d = Dec(cur); }
        return r;

 utest/robust_test.cpp

@@ -18,7 +18,7 @@
BOOST_AUTO_TEST_CASE(QueueTest)
{
    const int nthread = 100;
    const uint64_t nmsg = 1000 * 1000 * 10;
    const uint64_t nmsg = 1000 * 1000 * 100;

    SharedMemory &shm = TestShm();
    shm.Remove();
@@ -34,7 +34,7 @@
    }

#if 1
    typedef AtomicQueue<3> Rcb;
    typedef AtomicQueue<4> Rcb;

    Rcb tmp;
    BOOST_CHECK(tmp.like_empty());