worker.hpp

Go to the documentation of this file.

#pragma once
 
/// @file userver/ydb/dist_lock/worker.hpp
/// @brief @copybrief ydb::DistLockedWorker
 
#include <atomic>
#include <functional>
#include <memory>
#include <string>
 
#include <userver/engine/task/task_with_result.hpp>
#include <userver/utils/not_null.hpp>
#include <userver/utils/statistics/fwd.hpp>
#include <userver/ydb/coordination.hpp>
#include <userver/ydb/dist_lock/settings.hpp>
 
USERVER_NAMESPACE_BEGIN
 
namespace ydb {
 
namespace impl::dist_lock {
struct Statistics;
}  // namespace impl::dist_lock
 
/// 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 Callback = std::function<void()>;
 
  DistLockedWorker(engine::TaskProcessor& task_processor,
                   std::shared_ptr<CoordinationClient> coordination_client,
                   std::string_view coordination_node,
                   std::string_view semaphore_name, DistLockSettings settings,
                   Callback callback);
 
  ~DistLockedWorker();
 
  void Start();
 
  void Stop() noexcept;
 
  /// @brief Run the callback once under distlock, useful in tests.
  /// @warning Must not be run alongside a normal task launched via Start.
  void RunOnce();
 
  bool OwnsLock() const noexcept;
 
  friend void DumpMetric(utils::statistics::Writer& writer,
                         const DistLockedWorker& worker);
 
 private:
  void Run(bool run_once);
 
  engine::TaskProcessor& task_processor_;
  const std::shared_ptr<CoordinationClient> coordination_client_;
  const std::string coordination_node_;
  const std::string semaphore_name_;
  const DistLockSettings settings_;
  const Callback callback_;
 
  utils::UniqueRef<impl::dist_lock::Statistics> stats_;
  std::atomic<bool> owns_lock_{false};
 
  // Must be the last field, since it accesses other fields while alive.
  engine::TaskWithResult<void> worker_task_;
};
 
}  // namespace ydb
 
USERVER_NAMESPACE_END