pipeline/Flusher.hpp
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | // <Flusher.hpp> -*- C++ -*- | ||
| 2 | |||
| 3 | #pragma once | ||
| 4 | |||
| 5 | #include "simdb/Assert.hpp" | ||
| 6 | #include "simdb/pipeline/Stage.hpp" | ||
| 7 | #include "simdb/sqlite/DatabaseManager.hpp" | ||
| 8 | |||
| 9 | namespace simdb::pipeline { | ||
| 10 | |||
| 11 | /*! | ||
| 12 | * \class Flusher | ||
| 13 | * | ||
| 14 | * \brief Utility to run pipeline stages in order until no more work (processAll | ||
| 15 | * with force). Obtain from Pipeline::createFlusher(). Use when you need | ||
| 16 | * to drain queues or sync stages outside the normal polling loop. | ||
| 17 | */ | ||
| 18 | class Flusher | ||
| 19 | { | ||
| 20 | protected: | ||
| 21 | /// \brief Construct with the ordered list of stages to flush (internal; use Pipeline::createFlusher()). | ||
| 22 | 7 | Flusher(const std::vector<Stage*>& stages) : | |
| 23 | 7 | stages_(stages) | |
| 24 | { | ||
| 25 | 7 | simdb_assert(!stages_.empty(), "No stages given to Flusher"); | |
| 26 | 7 | } | |
| 27 | |||
| 28 | friend class Pipeline; | ||
| 29 | |||
| 30 | public: | ||
| 31 | 7 | virtual ~Flusher() = default; | |
| 32 | |||
| 33 | /// \brief Run all stages with processAll(true) in order until none return PROCEED. | ||
| 34 | /// \return PROCEED if any stage did work, SLEEP otherwise. | ||
| 35 | 1 | virtual PipelineAction flush() | |
| 36 | { | ||
| 37 | 1 | PipelineAction outcome = PipelineAction::SLEEP; | |
| 38 | |||
| 39 | 1 | bool continue_while = true; | |
| 40 | 2 | do | |
| 41 | { | ||
| 42 | 2 | continue_while = false; | |
| 43 | 6 | for (auto stage : stages_) | |
| 44 | { | ||
| 45 | 4 | constexpr auto force = true; | |
| 46 | 4 | if (stage->processAll(force) == PipelineAction::PROCEED) | |
| 47 | { | ||
| 48 | 2 | outcome = PipelineAction::PROCEED; | |
| 49 | 2 | continue_while = true; | |
| 50 | } | ||
| 51 | } | ||
| 52 | } while (continue_while); | ||
| 53 | |||
| 54 | 1 | return outcome; | |
| 55 | } | ||
| 56 | |||
| 57 | private: | ||
| 58 | std::vector<Stage*> stages_; | ||
| 59 | }; | ||
| 60 | |||
| 61 | /*! | ||
| 62 | * \class FlusherWithTransaction | ||
| 63 | * | ||
| 64 | * \brief Flusher that runs flush() inside a single safeTransaction() when any | ||
| 65 | * of the stages are DatabaseStages. Used by Pipeline::createFlusher() | ||
| 66 | * automatically when needed; reduces many small transactions to one. | ||
| 67 | */ | ||
| 68 | class FlusherWithTransaction : public Flusher | ||
| 69 | { | ||
| 70 | private: | ||
| 71 | /// \brief Internal constructor (Pipeline::createFlusher() uses this when a database stage is present). | ||
| 72 | 7 | FlusherWithTransaction(const std::vector<Stage*>& stages, DatabaseManager* db_mgr) : | |
| 73 | Flusher(stages), | ||
| 74 | 7 | db_mgr_(db_mgr) | |
| 75 | { | ||
| 76 | 7 | bool has_db_stage = false; | |
| 77 | 23 | for (auto stage : stages) | |
| 78 | { | ||
| 79 | 16 | if (dynamic_cast<DatabaseStageBase*>(stage)) | |
| 80 | { | ||
| 81 | 7 | has_db_stage = true; | |
| 82 | } | ||
| 83 | } | ||
| 84 | |||
| 85 | 7 | simdb_assert(has_db_stage, "Cannot use FlusherWithTransaction - there are " | |
| 86 | "no database stages!"); | ||
| 87 | 7 | } | |
| 88 | |||
| 89 | friend class Pipeline; | ||
| 90 | |||
| 91 | public: | ||
| 92 | /// \brief Run flush() inside a single BEGIN/COMMIT transaction. | ||
| 93 | /// \return PROCEED if any stage did work, SLEEP otherwise. | ||
| 94 | 1 | PipelineAction flush() override | |
| 95 | { | ||
| 96 | 1 | auto outcome = PipelineAction::SLEEP; | |
| 97 | 2 | db_mgr_->safeTransaction([&]() { outcome = Flusher::flush(); }); | |
| 98 | 1 | return outcome; | |
| 99 | } | ||
| 100 | |||
| 101 | private: | ||
| 102 | DatabaseManager* db_mgr_ = nullptr; | ||
| 103 | }; | ||
| 104 | |||
| 105 | } // namespace simdb::pipeline | ||
| 106 |