dist_locked_worker.hpp

#pragma once
 
#include <chrono>
#include <functional>
#include <memory>
#include <optional>
#include <string>
 
#include <userver/dist_lock/dist_lock_settings.hpp>
#include <userver/dist_lock/dist_lock_strategy.hpp>
#include <userver/dist_lock/statistics.hpp>
#include <userver/engine/mutex.hpp>
#include <userver/engine/task/task_with_result.hpp>
#include <userver/logging/level.hpp>
 
USERVER_NAMESPACE_BEGIN
 
namespace dist_lock {
namespace impl {
 
class Locker;
 
}  // namespace impl
 
/// A high-level primitive that perpetually tries to acquire a distributed lock
/// and runs user callback in a separate task while the lock is held.
/// Cancels the task when the lock is lost.
class DistLockedWorker final {
public:
    using WorkerFunc = std::function<void()>;
 
    /// Creates a DistLockedWorker.
    /// @param name name of the worker
    /// @param worker_func a callback that's started each time we acquire the lock
    /// and is cancelled when the lock is lost.
    /// @param settings distributed lock settings
    /// @param strategy distributed locking strategy
    /// @param task_processor TaskProcessor for running `worker_func`,
    /// using current TaskProcessor if `nullptr`
    /// @param locker_log_level Log level for Locker (default is info)
    /// @note `worker_func` must honour task cancellation and stop ASAP when
    /// it is cancelled, otherwise brain split is possible (IOW, two different
    /// users do work assuming both of them hold the lock, which is not true).
    DistLockedWorker(
        std::string name,
        WorkerFunc worker_func,
        std::shared_ptr<DistLockStrategyBase> strategy,
        const DistLockSettings& settings = {},
        engine::TaskProcessor* task_processor = nullptr,
        logging::Level locker_log_level = logging::Level::kInfo
    );
 
    ~DistLockedWorker();
 
    /// Name of the worker.
    const std::string& Name() const;
 
    /// Retrieves settings in a thread-safe way.
    DistLockSettings GetSettings() const;
 
    /// Update settings in a thread-safe way.
    void UpdateSettings(const DistLockSettings&);
 
    /// Starts acquiring the lock. Please note that it's possible that the lock is
    /// acquired and the WorkerFunc is entered *before* Start() returns.
    /// @see DistLockedTask::DistLockedTask
    void Start();
 
    /// Stops acquiring the lock. It is guaranteed that the lock is not held after
    /// Stop() return and WorkerFunc is stopped (if was started).
    void Stop();
 
    /// Run task once acquiring the lock.
    /// @throws std::exception rethrows exception from `worker_func`.
    void RunOnce();
 
    /// @returns whether the DistLockedTask is started.
    bool IsRunning() const;
 
    /// @returns is current worker owns the lock
    bool OwnsLock() const noexcept;
 
    /// Returns for how long the lock is held (if held at all). Returned value
    /// may be less than the real duration.
    std::optional<std::chrono::steady_clock::duration> GetLockedDuration() const;
 
    /// Returns lock acquisition statistics.
    const Statistics& GetStatistics() const;
 
    /// Returns `true` when it is advised to cancel the task.
    bool IsCancelAdvised() const;
 
private:
    engine::TaskProcessor& GetTaskProcessor() const noexcept;
 
    std::shared_ptr<impl::Locker> locker_ptr_;
 
    mutable engine::Mutex locker_task_mutex_;
    engine::TaskWithResult<void> locker_task_;
 
    engine::TaskProcessor* const task_processor_;
};
 
void DumpMetric(utils::statistics::Writer& writer, const DistLockedWorker& worker);
 
}  // namespace dist_lock
 
USERVER_NAMESPACE_END