pipeline/AsyncDatabaseAccessor.hpp
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | // <AsyncDatabaseAccessor.hpp> -*- C++ -*- | ||
| 2 | |||
| 3 | #pragma once | ||
| 4 | |||
| 5 | #include <functional> | ||
| 6 | #include <future> | ||
| 7 | #include <memory> | ||
| 8 | #include <string> | ||
| 9 | |||
| 10 | namespace simdb { | ||
| 11 | class DatabaseManager; | ||
| 12 | } | ||
| 13 | |||
| 14 | namespace simdb::pipeline { | ||
| 15 | |||
| 16 | /// \brief Callable queued by any thread and invoked on the dedicated database thread. | ||
| 17 | using AsyncDbAccessFunc = std::function<void(DatabaseManager*)>; | ||
| 18 | |||
| 19 | /*! | ||
| 20 | * \struct AsyncDatabaseTask | ||
| 21 | * | ||
| 22 | * \brief Wraps an AsyncDbAccessFunc and a std::promise for the exception message; | ||
| 23 | * used to run the function on the DB thread and propagate exceptions to the caller. | ||
| 24 | */ | ||
| 25 | struct AsyncDatabaseTask | ||
| 26 | { | ||
| 27 | AsyncDbAccessFunc func; | ||
| 28 | std::promise<std::string> exception_reason; | ||
| 29 | |||
| 30 | 4 | explicit AsyncDatabaseTask(AsyncDbAccessFunc func) : | |
| 31 | 4 | func(func) | |
| 32 | { | ||
| 33 | 4 | } | |
| 34 | AsyncDatabaseTask() = default; | ||
| 35 | }; | ||
| 36 | |||
| 37 | using AsyncDatabaseTaskPtr = std::shared_ptr<AsyncDatabaseTask>; | ||
| 38 | |||
| 39 | /*! | ||
| 40 | * \class AsyncDatabaseAccessHandler | ||
| 41 | * | ||
| 42 | * \brief Interface implemented by DatabaseThread. Receives tasks and runs them | ||
| 43 | * on the DB thread inside safeTransaction(); callers block until the | ||
| 44 | * task completes (or timeout). | ||
| 45 | */ | ||
| 46 | class AsyncDatabaseAccessHandler | ||
| 47 | { | ||
| 48 | public: | ||
| 49 | 9 | virtual ~AsyncDatabaseAccessHandler() = default; | |
| 50 | |||
| 51 | /// \brief Run the task on the DB thread; block until done (or timeout). | ||
| 52 | /// \param task Task to run inside a safeTransaction(). | ||
| 53 | /// \param timeout_seconds If > 0, throw if not completed within this many seconds. | ||
| 54 | /// \throws DBException on timeout or if the task throws. | ||
| 55 | virtual void eval(AsyncDatabaseTaskPtr&& task, double timeout_seconds = 0) = 0; | ||
| 56 | }; | ||
| 57 | |||
| 58 | /*! | ||
| 59 | * \class AsyncDatabaseAccessor | ||
| 60 | * | ||
| 61 | * \brief Handle for pipeline stages and apps to run code on the dedicated | ||
| 62 | * database thread. eval(func) blocks the caller until func(DatabaseManager*) | ||
| 63 | * has run. Obtained from PipelineManager::getAsyncDatabaseAccessor() or | ||
| 64 | * DatabaseThread::getAsyncDatabaseAccessor(). | ||
| 65 | */ | ||
| 66 | class AsyncDatabaseAccessor | ||
| 67 | { | ||
| 68 | public: | ||
| 69 | /// \brief Run \p func on the DB thread; block until it completes (or timeout). | ||
| 70 | /// \param func Callable that receives the DatabaseManager*. | ||
| 71 | /// \param timeout_seconds If > 0, throw if not completed within this many seconds. | ||
| 72 | /// \throws DBException on timeout or if \p func throws. | ||
| 73 | 4 | void eval(const AsyncDbAccessFunc& func, double timeout_seconds = 0) | |
| 74 | { | ||
| 75 | 4 | auto task = std::make_shared<AsyncDatabaseTask>(func); | |
| 76 | 4 | db_access_handler_->eval(std::move(task), timeout_seconds); | |
| 77 | 4 | } | |
| 78 | |||
| 79 | private: | ||
| 80 | 9 | AsyncDatabaseAccessor(AsyncDatabaseAccessHandler* db_access_handler) : | |
| 81 | 9 | db_access_handler_(db_access_handler) | |
| 82 | { | ||
| 83 | 9 | } | |
| 84 | |||
| 85 | AsyncDatabaseAccessHandler* db_access_handler_ = nullptr; | ||
| 86 | friend class DatabaseThread; | ||
| 87 | }; | ||
| 88 | |||
| 89 | } // namespace simdb::pipeline | ||
| 90 |