pipeline/Queue.hpp
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | // <Queue.hpp> -*- C++ -*- | ||
| 2 | |||
| 3 | #pragma once | ||
| 4 | |||
| 5 | #include "simdb/utils/ConcurrentQueue.hpp" | ||
| 6 | #include "simdb/utils/Demangle.hpp" | ||
| 7 | #include <functional> | ||
| 8 | #include <memory> | ||
| 9 | |||
| 10 | namespace simdb::pipeline { | ||
| 11 | |||
| 12 | /*! | ||
| 13 | * \class QueueBase | ||
| 14 | * | ||
| 15 | * \brief Type-erased base for concurrent queues that move data between | ||
| 16 | * pipeline threads. Use Queue<T> for typed access. | ||
| 17 | */ | ||
| 18 | class QueueBase | ||
| 19 | { | ||
| 20 | public: | ||
| 21 | 36 | virtual ~QueueBase() = default; | |
| 22 | /// \brief Return a string representation of the queue's value type (e.g. for diagnostics). | ||
| 23 | virtual std::string stringifiedType() const = 0; | ||
| 24 | /// \brief Return the number of elements currently in the queue. | ||
| 25 | virtual size_t size() const = 0; | ||
| 26 | }; | ||
| 27 | |||
| 28 | /*! | ||
| 29 | * \class Queue | ||
| 30 | * | ||
| 31 | * \brief Typed wrapper around ConcurrentQueue<T> for pipeline input/output | ||
| 32 | * ports. | ||
| 33 | * \tparam T Element type of the queue. | ||
| 34 | */ | ||
| 35 | template <typename T> class Queue : public QueueBase | ||
| 36 | { | ||
| 37 | public: | ||
| 38 | /// \brief Mutable access to the underlying ConcurrentQueue<T>. | ||
| 39 | 55 | ConcurrentQueue<T>& get() { return queue_; } | |
| 40 | |||
| 41 | /// \brief Const access to the underlying ConcurrentQueue<T>. | ||
| 42 | const ConcurrentQueue<T>& get() const { return queue_; } | ||
| 43 | |||
| 44 | /// \brief Return a string representation of the queue's value type (e.g. for diagnostics). | ||
| 45 | ✗ | std::string stringifiedType() const override { return demangle_type<T>(); } | |
| 46 | |||
| 47 | /// \brief Return the number of elements currently in the queue. | ||
| 48 | ✗ | size_t size() const override { return queue_.size(); } | |
| 49 | |||
| 50 | private: | ||
| 51 | ConcurrentQueue<T> queue_; | ||
| 52 | }; | ||
| 53 | |||
| 54 | /// \brief Unique_ptr to a Queue<InputType> (convenience alias). | ||
| 55 | template <typename InputType> using InputQueuePtr = std::unique_ptr<Queue<InputType>>; | ||
| 56 | |||
| 57 | } // namespace simdb::pipeline | ||
| 58 |