Ett alternativ, STL deque

Om det är som så att du behöver stoppa in och ta ut element i båda ändarna av en container, så är std::deque ett givet alternativ. Den har en accesstid i båda ändarna som inte beror av antalet element i containern. I denna artikel beskriver jag mer om hur du kan använda denna, samt visar en typisk tillämpning där den förekommer i implementationen av en thread-safe mesage queue.

#include <deque>
//...
auto q = std::deque<int>{};
auto N = 25;
for (auto k = 1; k <= N; ++k) q.push_front(k);

N = q.size()/2;
for (auto k = 0; k < N; ++k) {
    std::print("{} ", q.front());
    q.pop_front();
}
std::println("\n----");
while (!q.empty()) {
    std::print("{} ", q.back());
    q.pop_back();
}

25 24 23 22 21 20 19 18 17 16 15 14 
----
1 2 3 4 5 6 7 8 9 10 11 12 13 

Try it yourself on Compiler Explorer

Jag har i en tidigare artikel beskrivit användning av std::vector. Denna består internt av ett heap-allokerat minnesblock, som omallokeras i takt med att man stoppar in fler element. Containern std::deque å andra sidan består internt av ett varierande antal lika stora minnesblock, plus en tabell som håller reda på vilket index-intervall, som hör till respektive minnesblock.

Storleken på minnesblocken är inte bestämd av C++ standarden utan implementations-beroende och vald på så sätt att det passar in i CPU cachen. Precis som för std::vector, så har std::deque konstant access-tid, dvs O(1). Att returnera referensen till ett element går till på "i princip" följande sätt:

auto operator[](int index) -> ElementType& {
    auto block_index   = (start_offset + index) / block_size;
    auto element_index = (start_offset + index) % block_size;
    return block_table[block_index][element_index];
}

Till skillnad från vector, så har deque inte en reserve() metod. Dock finns det en resize() metod som fungerar på likartat sätt som för vector. Vilket innebär att man sätter in ett bestämt antal element med ett visst värde eller via dess default-constructor.

auto q = std::deque<unsigned>{};
q.resize(12, 42);
while (!q.empty()) {
    std::print("{} ", q.back());
    q.pop_back();
}
//prints: 42 42 42 42 42 42 42 42 42 42 42 42

Containern fungerar bra som en så kallad FIFO (First In, First Out) kö. Här följer ett litet program där den används för att realisera en thread-safe unbounded message-queue.

#include <deque>
#include <print>
#include <thread>
#include <mutex>
#include <condition_variable>

template<typename T>
struct MessageQueue {
    std::deque<T>            q{};
    std::mutex               entry{};
    std::condition_variable  not_empty{};

    void put(T x) {
        auto guard = std::lock_guard<std::mutex>{entry};
        q.push_back(x);
        not_empty.notify_all();
    }
    
    T get() {
        auto guard = std::unique_lock<std::mutex>{entry};
        not_empty.wait(guard, [this]{ return not q.empty(); });
        auto x = q.front();
        q.pop_front();
        return x;
    }
};

int main() {
    auto mq = MessageQueue<unsigned long>{};

    auto consumer = std::jthread{[&mq] {
        auto count = 0U;
        auto sum   = 0UL;
        for (auto msg = mq.get(); msg > 0; msg = mq.get()) {
            ++count;
            sum += msg;
        }
        std::println("[consumer] count={}, sum={}", count, sum);
    }};

    auto const N = 100'000UL;
    for (auto k = 1UL; k <= N; ++k) mq.put(k);
    mq.put(0UL);
    std::println("[producer] count={}, sum={}", N, N * (N + 1) / 2);
}

[producer] count=100000, sum=5000050000
[consumer] count=100000, sum=5000050000

Try it yourself on Compiler Explorer

Först implementerar jag meddelande-kön, vilken består av en deque och en mutex plus en condition. De två sista behövs för att göra kön thread-safe. Dylika köer kan antingen vara unbounded eller bounded. I det första fallet är det bara consumer som suspenderas när kön är tom. I det andra fallet suspenderas även producer då kön är full.

I put() metoden använder jag en lock_guard för att aktivera mutex lock och som utför mutex unlock, vid function return. Innan return, så signaleras att kön är icke-tom, så en eventuell väntande consumer kan väckas från en suspendering.

I get() metoden behöver jag använda en unique_lock i stället, eftersom denna behövs i samband med wait metoden hos condition variabeln. Att man behöver hålla reda när respektive guard ska användas är en omdebatterad inkonsekvens i API:et, vilket skapades av Anthony Williams. Ett alternativ är att använda unique_lock i båda metoderna.

I main() funktionen skapar jag två threads, den första explicit (consumer) och den andra är implicit i form av main() funktionen själv. Producer skickar en stigande sekvens av heltal, via kön och consumer tar emot dessa och summerar ihop dem.

På slutet skriver consumer ut antal och summa och likaså producer skriver ut antal, samt beräknar summan med Gauss summeringsformel. Talen ska givetvis vara samma om allt funkar som det ska.

För fullständighetens skull visar jag också implementationen av en bounded message-queue. Det är bara några små tillägg som behövs. En extra condition för att representera att kän kan bli full, samt att put ändras till att använda unique_lock och condition wait.

template<typename T, unsigned CAPACITY>
struct MessageQueue {
    std::deque<T>            q{};
    std::mutex               entry{};
    std::condition_variable  not_empty{};
    std::condition_variable  not_full{};

    void put(T x) {
        auto guard = std::unique_lock<std::mutex>{entry};
        not_full.wait(guard, [this]{ return q.size() <= CAPACITY; });
        q.push_back(x);
        not_empty.notify_all();
    }
    
    T get() {
        auto guard = std::unique_lock<std::mutex>{entry};
        not_empty.wait(guard, [this]{ return not q.empty(); });
        auto x = q.front();
        q.pop_front();
        not_full.notify_all();
        return x;
    }
};

int main() {
    auto mq = MessageQueue<unsigned long, 16>{};
    //the rest is the same code as before
}

Try it yourself on Compiler Explorer