summaryrefslogblamecommitdiffstats
path: root/src/common/threadsafe_queue.h
blob: f553efdc9ac9b26a56a0eead4c945a05ca55020c (plain) (tree)
1
2
3
4
5
6
7
8
9








                                          


                  
                  

                  
                     

                 
                 






                                                
                              



                           
                           
     
 


                                 
 








                                                                  

               


                

               











                                                                         
               




















                                                                
                       







                                               
                                               



                          
                               




                                       
                     
                 
       
                              










                                  


                                                     













                                 


        
                            


                          
// Copyright 2010 Dolphin Emulator Project
// Licensed under GPLv2+
// Refer to the license.txt file included.

#pragma once

// a simple lockless thread-safe,
// single reader, single writer queue

#include <atomic>
#include <cstddef>
#include <mutex>
#include <utility>

namespace Common {
template <typename T>
class SPSCQueue {
public:
    SPSCQueue() {
        write_ptr = read_ptr = new ElementPtr();
    }
    ~SPSCQueue() {
        // this will empty out the whole queue
        delete read_ptr;
    }

    std::size_t Size() const {
        return size.load();
    }

    bool Empty() const {
        return Size() == 0;
    }

    T& Front() const {
        return read_ptr->current;
    }

    template <typename Arg>
    void Push(Arg&& t) {
        // create the element, add it to the queue
        write_ptr->current = std::forward<Arg>(t);
        // set the next pointer to a new element ptr
        // then advance the write pointer
        ElementPtr* new_ptr = new ElementPtr();
        write_ptr->next.store(new_ptr, std::memory_order_release);
        write_ptr = new_ptr;

        ++size;
    }

    void Pop() {
        --size;

        ElementPtr* tmpptr = read_ptr;
        // advance the read pointer
        read_ptr = tmpptr->next.load();
        // set the next element to nullptr to stop the recursive deletion
        tmpptr->next.store(nullptr);
        delete tmpptr; // this also deletes the element
    }

    bool Pop(T& t) {
        if (Empty())
            return false;

        --size;

        ElementPtr* tmpptr = read_ptr;
        read_ptr = tmpptr->next.load(std::memory_order_acquire);
        t = std::move(tmpptr->current);
        tmpptr->next.store(nullptr);
        delete tmpptr;
        return true;
    }

    // not thread-safe
    void Clear() {
        size.store(0);
        delete read_ptr;
        write_ptr = read_ptr = new ElementPtr();
    }

private:
    // stores a pointer to element
    // and a pointer to the next ElementPtr
    class ElementPtr {
    public:
        ElementPtr() {}
        ~ElementPtr() {
            ElementPtr* next_ptr = next.load();

            if (next_ptr)
                delete next_ptr;
        }

        T current;
        std::atomic<ElementPtr*> next{nullptr};
    };

    ElementPtr* write_ptr;
    ElementPtr* read_ptr;
    std::atomic_size_t size{0};
};

// a simple thread-safe,
// single reader, multiple writer queue

template <typename T>
class MPSCQueue {
public:
    std::size_t Size() const {
        return spsc_queue.Size();
    }

    bool Empty() const {
        return spsc_queue.Empty();
    }

    T& Front() const {
        return spsc_queue.Front();
    }

    template <typename Arg>
    void Push(Arg&& t) {
        std::lock_guard<std::mutex> lock(write_lock);
        spsc_queue.Push(t);
    }

    void Pop() {
        return spsc_queue.Pop();
    }

    bool Pop(T& t) {
        return spsc_queue.Pop(t);
    }

    // not thread-safe
    void Clear() {
        spsc_queue.Clear();
    }

private:
    SPSCQueue<T> spsc_queue;
    std::mutex write_lock;
};
} // namespace Common