pipeline/Runnable.hpp
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | // <Runnable.hpp> -*- C++ -*- | ||
| 2 | |||
| 3 | #pragma once | ||
| 4 | |||
| 5 | #include "simdb/Assert.hpp" | ||
| 6 | #include "simdb/sqlite/DatabaseManager.hpp" | ||
| 7 | |||
| 8 | #include <iostream> | ||
| 9 | #include <memory> | ||
| 10 | #include <string> | ||
| 11 | #include <type_traits> | ||
| 12 | #include <vector> | ||
| 13 | |||
| 14 | namespace simdb::pipeline { | ||
| 15 | |||
| 16 | class PipelineManager; | ||
| 17 | class PollingThread; | ||
| 18 | |||
| 19 | /// \brief Outcome of processOne() / processAll(): whether the runnable did work or the thread may sleep. | ||
| 20 | enum PipelineAction { | ||
| 21 | /// Runnable did work (or should keep the thread busy); thread continues without sleeping. | ||
| 22 | PROCEED, | ||
| 23 | /// Runnable had nothing to do; if all runnables on the thread return SLEEP, the thread sleeps. | ||
| 24 | SLEEP | ||
| 25 | }; | ||
| 26 | |||
| 27 | /*! | ||
| 28 | * \class Runnable | ||
| 29 | * | ||
| 30 | * \brief Base class for units of work on a pipeline thread (e.g. Stage). PollingThread | ||
| 31 | * calls processOne() / processAll(); PROCEED means work was done, SLEEP means | ||
| 32 | * none. Supports enable/disable and a human-readable description for reporting. | ||
| 33 | */ | ||
| 34 | class Runnable | ||
| 35 | { | ||
| 36 | public: | ||
| 37 | 32 | virtual ~Runnable() = default; | |
| 38 | |||
| 39 | /// \brief Return the runnable's description (for logging/reports); uses set value or getDescription_(). | ||
| 40 | 32 | std::string getDescription() const { return !description_.empty() ? description_ : getDescription_(); } | |
| 41 | |||
| 42 | /// \brief Set or overwrite the runnable's description. | ||
| 43 | /// \param desc Description string. | ||
| 44 | void setDescription(const std::string& desc) { description_ = desc; } | ||
| 45 | |||
| 46 | /// \brief Process one unit of work; return PROCEED if work was done, SLEEP otherwise. | ||
| 47 | /// \param force If true, run even when there is no input (e.g. for flushing). | ||
| 48 | virtual PipelineAction processOne(bool force) = 0; | ||
| 49 | |||
| 50 | /// \brief Process all pending work; return PROCEED if any work was done, SLEEP otherwise. | ||
| 51 | /// \param force If true, run until no more work (e.g. for flushing). | ||
| 52 | virtual PipelineAction processAll(bool force) = 0; | ||
| 53 | |||
| 54 | /// \brief Print a one-line description to \p os with \p indent spaces (for perf reports). | ||
| 55 | 32 | virtual void print(std::ostream& os, int indent) const | |
| 56 | { | ||
| 57 | 32 | os << std::string(indent, ' ') << getDescription() << "\n"; | |
| 58 | 32 | } | |
| 59 | |||
| 60 | /// \brief Return true if this runnable is enabled (disabled runnables are skipped by the thread). | ||
| 61 | 144799 | bool enabled() const { return enabled_; } | |
| 62 | |||
| 63 | /// \brief Enable or disable this runnable. | ||
| 64 | /// \param enable true to enable, false to disable. | ||
| 65 | 79998 | void enable(bool enable = true) { enabled_ = enable; } | |
| 66 | |||
| 67 | protected: | ||
| 68 | 25688 | const PollingThread* getThread_() const | |
| 69 | { | ||
| 70 | 25688 | simdb_assert(thread_, "PollingThread never set"); | |
| 71 | 25688 | return thread_; | |
| 72 | } | ||
| 73 | |||
| 74 | private: | ||
| 75 | virtual std::string getDescription_() const = 0; | ||
| 76 | std::string description_; | ||
| 77 | bool enabled_ = true; | ||
| 78 | |||
| 79 | const PollingThread* thread_ = nullptr; | ||
| 80 | friend class PollingThread; | ||
| 81 | }; | ||
| 82 | |||
| 83 | /*! | ||
| 84 | * \class ScopedRunnableDisabler | ||
| 85 | * | ||
| 86 | * \brief RAII guard that disables the given runnables (and optionally pauses | ||
| 87 | * the given polling threads) on construction and re-enables/resumes | ||
| 88 | * them on destruction. Obtained from PipelineManager::scopedDisableAll(). | ||
| 89 | */ | ||
| 90 | class ScopedRunnableDisabler | ||
| 91 | { | ||
| 92 | public: | ||
| 93 | ~ScopedRunnableDisabler(); | ||
| 94 | |||
| 95 | private: | ||
| 96 | ScopedRunnableDisabler(PipelineManager* pipeline_mgr, const std::vector<Runnable*>& runnables, | ||
| 97 | const std::vector<PollingThread*>& polling_threads); | ||
| 98 | |||
| 99 | ScopedRunnableDisabler(PipelineManager* pipeline_mgr, const std::vector<Runnable*>& runnables); | ||
| 100 | |||
| 101 | void notifyPipelineMgrReenabled_(); | ||
| 102 | |||
| 103 | PipelineManager* pipeline_mgr_ = nullptr; | ||
| 104 | std::vector<Runnable*> disabled_runnables_; | ||
| 105 | std::vector<PollingThread*> paused_threads_; | ||
| 106 | friend class PipelineManager; | ||
| 107 | }; | ||
| 108 | |||
| 109 | } // namespace simdb::pipeline | ||
| 110 |