path: root/src/common/ceph_timer.h

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software
 * Foundation.  See file COPYING.
 *
 */

#ifndef COMMON_CEPH_TIMER_H
#define COMMON_CEPH_TIMER_H

#include <condition_variable>
#include <thread>
#include <boost/intrusive/set.hpp>

namespace ceph {

  /// Newly constructed timer should be suspended at point of
  /// construction.

  struct construct_suspended_t { };
  constexpr construct_suspended_t construct_suspended { };

  namespace timer_detail {
    using boost::intrusive::member_hook;
    using boost::intrusive::set_member_hook;
    using boost::intrusive::link_mode;
    using boost::intrusive::normal_link;
    using boost::intrusive::set;
    using boost::intrusive::constant_time_size;
    using boost::intrusive::compare;

    // Compared to the SafeTimer this does fewer allocations (you
    // don't have to allocate a new Context every time you
    // want to cue the next tick.)
    //
    // It also does not share a lock with the caller. If you call
    // cancel event, it either cancels the event (and returns true) or
    // you missed it. If this does not work for you, you can set up a
    // flag and mutex of your own.
    //
    // You get to pick your clock. I like mono_clock, since I usually
    // want to wait FOR a given duration. real_clock is worthwhile if
    // you want to wait UNTIL a specific moment of wallclock time.  If
    // you want you can set up a timer that executes a function after
    // you use up ten seconds of CPU time.

    template <class TC>
    class timer {
      using sh = set_member_hook<link_mode<normal_link> >;

      struct event {
	typename TC::time_point t;
	uint64_t id;
	std::function<void()> f;

	sh schedule_link;
	sh event_link;

	event() : t(TC::time_point::min()), id(0) {}
	event(uint64_t _id) : t(TC::time_point::min()), id(_id) {}
	event(typename TC::time_point _t, uint64_t _id,
	      std::function<void()>&& _f) : t(_t), id(_id), f(_f) {}
	event(typename TC::time_point _t, uint64_t _id,
	      const std::function<void()>& _f) : t(_t), id(_id), f(_f) {}
	bool operator <(const event& e) {
	  return t == e.t ? id < e.id : t < e.t;
	}
      };
      struct SchedCompare {
	bool operator()(const event& e1, const event& e2) const {
	  return e1.t == e2.t ? e1.id < e2.id : e1.t < e2.t;
	}
      };
      struct EventCompare {
	bool operator()(const event& e1, const event& e2) const {
	  return e1.id < e2.id;
	}
      };

      using schedule_type = set<event,
				member_hook<event, sh, &event::schedule_link>,
				constant_time_size<false>,
				compare<SchedCompare> >;

      schedule_type schedule;

      using event_set_type = set<event,
				 member_hook<event, sh, &event::event_link>,
				 constant_time_size<false>,
				 compare<EventCompare> >;

      event_set_type events;

      std::mutex lock;
      using lock_guard = std::lock_guard<std::mutex>;
      using unique_lock = std::unique_lock<std::mutex>;
      std::condition_variable cond;

      event* running{ nullptr };
      uint64_t next_id{ 0 };

      bool suspended;
      std::thread thread;

      void timer_thread() {
	unique_lock l(lock);
	while (!suspended) {
	  typename TC::time_point now = TC::now();

	  while (!schedule.empty()) {
	    auto p = schedule.begin();
	    // Should we wait for the future?
	    if (p->t > now)
	      break;

	    event& e = *p;
	    schedule.erase(e);
	    events.erase(e);

	    // Since we have only one thread it is impossible to have more
	    // than one running event
	    running = &e;

	    l.unlock();
	    e.f();
	    l.lock();

	    if (running) {
	      running = nullptr;
	      delete &e;
	    } // Otherwise the event requeued itself
	  }

          if (suspended)
            break;
	  if (schedule.empty())
	    cond.wait(l);
	  else
	    cond.wait_until(l, schedule.begin()->t);
	}
      }

  public:
      timer() {
	lock_guard l(lock);
	suspended = false;
	thread = std::thread(&timer::timer_thread, this);
      }

      // Create a suspended timer, jobs will be executed in order when
      // it is resumed.
      timer(construct_suspended_t) {
	lock_guard l(lock);
	suspended = true;
      }

      timer(const timer &) = delete;
      timer& operator=(const timer &) = delete;

      ~timer() {
	suspend();
	cancel_all_events();
      }

      // Suspend operation of the timer (and let its thread die).
      void suspend() {
	unique_lock l(lock);
	if (suspended)
	  return;

	suspended = true;
	cond.notify_one();
	l.unlock();
	thread.join();
      }


      // Resume operation of the timer. (Must have been previously
      // suspended.)
      void resume() {
	unique_lock l(lock);
	  if (!suspended)
	  return;

	suspended = false;
	ceph_assert(!thread.joinable());
	thread = std::thread(&timer::timer_thread, this);
      }

      // Schedule an event in the relative future
      template<typename Callable, typename... Args>
      uint64_t add_event(typename TC::duration duration,
			 Callable&& f, Args&&... args) {
	typename TC::time_point when = TC::now();
	when += duration;
	return add_event(when,
			 std::forward<Callable>(f),
			 std::forward<Args>(args)...);
      }

      // Schedule an event in the absolute future
      template<typename Callable, typename... Args>
      uint64_t add_event(typename TC::time_point when,
			 Callable&& f, Args&&... args) {
	std::lock_guard l(lock);
	event& e = *(new event(
		       when, ++next_id,
		       std::forward<std::function<void()> >(
			 std::bind(std::forward<Callable>(f),
				   std::forward<Args>(args)...))));
	auto i = schedule.insert(e);
	events.insert(e);

	/* If the event we have just inserted comes before everything
	 * else, we need to adjust our timeout. */
	if (i.first == schedule.begin())
	  cond.notify_one();

	// Previously each event was a context, identified by a
	// pointer, and each context to be called only once. Since you
	// can queue the same function pointer, member function,
	// lambda, or functor up multiple times, identifying things by
	// function for the purposes of cancellation is no longer
	// suitable. Thus:
	return e.id;
      }

      // Adjust the timeout of a currently-scheduled event (relative)
      bool adjust_event(uint64_t id, typename TC::duration duration) {
	return adjust_event(id, TC::now() + duration);
      }

      // Adjust the timeout of a currently-scheduled event (absolute)
      bool adjust_event(uint64_t id, typename TC::time_point when) {
	std::lock_guard l(lock);

	event key(id);
	typename event_set_type::iterator it = events.find(key);

	if (it == events.end())
	  return false;

	event& e = *it;

	schedule.erase(e);
	e.t = when;
	schedule.insert(e);

	return true;
      }

      // Cancel an event. If the event has already come and gone (or you
      // never submitted it) you will receive false. Otherwise you will
      // receive true and it is guaranteed the event will not execute.
      bool cancel_event(const uint64_t id) {
	std::lock_guard l(lock);
	event dummy(id);
	auto p = events.find(dummy);
	if (p == events.end()) {
	  return false;
	}

	event& e = *p;
	events.erase(e);
	schedule.erase(e);
	delete &e;

	return true;
      }

      // Reschedules a currently running event in the relative
      // future. Must be called only from an event executed by this
      // timer. If you have a function that can be called either from
      // this timer or some other way, it is your responsibility to make
      // sure it can tell the difference only does not call
      // reschedule_me in the non-timer case.
      //
      // Returns an event id. If you had an event_id from the first
      // scheduling, replace it with this return value.
      uint64_t reschedule_me(typename TC::duration duration) {
	return reschedule_me(TC::now() + duration);
      }

      // Reschedules a currently running event in the absolute
      // future. Must be called only from an event executed by this
      // timer. if you have a function that can be called either from
      // this timer or some other way, it is your responsibility to make
      // sure it can tell the difference only does not call
      // reschedule_me in the non-timer case.
      //
      // Returns an event id. If you had an event_id from the first
      // scheduling, replace it with this return value.
      uint64_t reschedule_me(typename TC::time_point when) {
	if (std::this_thread::get_id() != thread.get_id())
	  throw std::make_error_condition(std::errc::operation_not_permitted);
	std::lock_guard l(lock);
	running->t = when;
	uint64_t id = ++next_id;
	running->id = id;
	schedule.insert(*running);
	events.insert(*running);

	// Hacky, but keeps us from being deleted
	running = nullptr;

	// Same function, but you get a new ID.
	return id;
      }

      // Remove all events from the queue.
      void cancel_all_events() {
	std::lock_guard l(lock);
	while (!events.empty()) {
	  auto p = events.begin();
	  event& e = *p;
	  schedule.erase(e);
	  events.erase(e);
	  delete &e;
	}
      }
    }; // timer
  }; // timer_detail

  using timer_detail::timer;
}; // ceph

#endif