core: Basic implementation of Scheduler

refs #1119 (http://redmine.named-data.net/issues/1119)

Change-Id: I1783857452941794a2538534d3492d30ac58dd25

daemon/common.hpp

@@ -17,9 +17,12 @@
 #include <boost/function.hpp>
 #include <boost/bind.hpp>
 #include <boost/ref.hpp>
+#include <boost/asio.hpp>
+#include <boost/assert.hpp>
 
 #include <vector>
 #include <list>
+#include <set>
 
 namespace ndn {
 

daemon/core/monotonic_deadline_timer.hpp

@@ -0,0 +1,61 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (C) 2014 Named Data Networking Project
+ * See COPYING for copyright and distribution information.
+ */
+
+/**
+ * This code is based on https://svn.boost.org/trac/boost/attachment/ticket/3504/MonotonicDeadlineTimer.h
+ */
+
+#ifndef NFD_CORE_MONOTONIC_DEADLINE_TIMER_HPP
+#define NFD_CORE_MONOTONIC_DEADLINE_TIMER_HPP
+
+#include "time.hpp"
+
+namespace boost {
+namespace asio {
+
+template <> 
+struct time_traits<ndn::time::monotonic_clock>
+{
+  typedef ndn::time::Point time_type;
+  typedef ndn::time::Duration duration_type;
+
+  static time_type
+  now()
+  {
+    return ndn::time::now();
+  }
+
+  static time_type
+  add(const time_type& time, const duration_type& duration) 
+  {
+    return time + duration;
+  }
+
+  static duration_type
+  subtract(const time_type& timeLhs, const time_type& timeRhs)
+  {
+    return timeLhs - timeRhs;
+  }
+
+  static bool
+  less_than(const time_type& timeLhs, const time_type& timeRhs)
+  {
+    return timeLhs < timeRhs;
+  }
+
+  static boost::posix_time::time_duration
+  to_posix_duration(const duration_type& duration)
+  {
+    return boost::posix_time::microseconds(duration/1000);
+  }
+};
+
+typedef basic_deadline_timer<ndn::time::monotonic_clock> monotonic_deadline_timer; 
+
+} // namespace asio
+} // namespace boost
+
+#endif // NFD_CORE_MONOTONIC_DEADLINE_TIMER_HPP

daemon/core/scheduler.cpp

@@ -0,0 +1,176 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (C) 2014 Named Data Networking Project
+ * See COPYING for copyright and distribution information.
+ */
+
+#include "scheduler.hpp"
+
+namespace ndn {
+
+struct EventIdImpl
+{
+  EventIdImpl(const Scheduler::EventQueue::iterator& event)
+    : m_event(event)
+    , m_isValid(true)
+  {
+  }
+
+  void
+  invalidate()
+  {
+    m_isValid = false;
+  }
+
+  bool
+  isValid() const
+  {
+    return m_isValid;
+  }
+
+  operator const Scheduler::EventQueue::iterator&() const
+  {
+    return m_event;
+  }
+
+  void
+  reset(const Scheduler::EventQueue::iterator& newIterator)
+  {
+    m_event = newIterator;
+    m_isValid = true;
+  }
+
+private:
+  Scheduler::EventQueue::iterator m_event;
+  bool m_isValid;
+};
+
+Scheduler::EventInfo::EventInfo(const time::Duration& after,
+                        const time::Duration& period,
+                        const Event& event)
+  : m_scheduledTime(time::now() + after)
+  , m_period(period)
+  , m_event(event)
+{
+}
+
+Scheduler::EventInfo::EventInfo(const time::Point& when, const EventInfo& previousEvent)
+  : m_scheduledTime(when)
+  , m_period(previousEvent.m_period)
+  , m_event(previousEvent.m_event)
+  , m_eventId(previousEvent.m_eventId)
+{
+}
+
+time::Duration
+Scheduler::EventInfo::expiresFromNow() const
+{
+  time::Point now = time::now();
+  if (now > m_scheduledTime)
+    return time::seconds(0); // event should be scheduled ASAP
+  else
+    return m_scheduledTime - now;
+}
+
+
+Scheduler::Scheduler(boost::asio::io_service& ioService)
+  : m_ioService(ioService)
+  , m_scheduledEvent(m_events.end())
+  , m_deadlineTimer(ioService)
+{
+}
+
+EventId
+Scheduler::scheduleEvent(const time::Duration& after,
+                         const Event& event)
+{
+  return schedulePeriodicEvent(after, time::nanoseconds(-1), event);
+}
+
+EventId
+Scheduler::schedulePeriodicEvent(const time::Duration& after,
+                                 const time::Duration& period,
+                                 const Event& event)
+{
+  EventQueue::iterator i = m_events.insert(EventInfo(after, period, event));
+  i->m_eventId = make_shared<EventIdImpl>(boost::cref(i));
+  
+  if (m_scheduledEvent == m_events.end() ||
+      *i < *m_scheduledEvent)
+    {
+      m_deadlineTimer.expires_from_now(after);
+      m_deadlineTimer.async_wait(bind(&Scheduler::onEvent, this, _1));
+      m_scheduledEvent = i;
+    }
+
+  return i->m_eventId;
+}
+  
+void
+Scheduler::cancelEvent(const EventId& eventId)
+{
+  if (!eventId->isValid())
+    return; // event already fired or cancelled
+  
+  if (static_cast<EventQueue::iterator>(*eventId) != m_scheduledEvent) {
+    m_events.erase(*eventId);
+    eventId->invalidate();
+    return;
+  }
+  
+  m_deadlineTimer.cancel();
+  m_events.erase(static_cast<EventQueue::iterator>(*eventId));
+  eventId->invalidate();
+
+  if (!m_events.empty())
+    {
+      m_deadlineTimer.expires_from_now(m_events.begin()->expiresFromNow());
+      m_deadlineTimer.async_wait(bind(&Scheduler::onEvent, this, _1));
+      m_scheduledEvent = m_events.begin();
+    }
+  else
+    {
+      m_deadlineTimer.cancel();
+      m_scheduledEvent = m_events.end();
+    }
+}
+
+void
+Scheduler::onEvent(const boost::system::error_code& error)
+{
+  if (error) // e.g., cancelled
+    {
+      return;
+    }
+  
+  // process all expired events
+  time::Point now = time::now();
+  while(!m_events.empty() && m_events.begin()->m_scheduledTime <= now)
+    {
+      EventQueue::iterator head = m_events.begin();
+      
+      head->m_event();
+      if (head->m_period < 0)
+        {
+          head->m_eventId->invalidate();
+          m_events.erase(head);
+        }
+      else
+        {
+          // "reschedule" and update EventId data of the event
+          EventInfo event(now + head->m_period, *head);
+          EventQueue::iterator i = m_events.insert(event);
+          i->m_eventId->reset(i);
+          m_events.erase(head);
+        }
+    }
+
+  if (!m_events.empty())
+    {
+      m_deadlineTimer.expires_from_now(m_events.begin()->m_scheduledTime - now);
+      m_deadlineTimer.async_wait(bind(&Scheduler::onEvent, this, _1));
+    }
+}
+
+
+} // namespace ndn

daemon/core/scheduler.hpp

@@ -0,0 +1,99 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (C) 2014 Named Data Networking Project
+ * See COPYING for copyright and distribution information.
+ */
+
+#ifndef NFD_CORE_SCHEDULER_HPP
+#define NFD_CORE_SCHEDULER_HPP
+
+#include "common.hpp"
+#include "monotonic_deadline_timer.hpp"
+
+namespace ndn {
+
+struct EventIdImpl; ///< \brief Private storage of information about the event
+/**
+ * \brief Opaque type (shared_ptr) representing ID of the scheduled event
+ */
+typedef shared_ptr<EventIdImpl> EventId;
+
+/**
+ * \brief Generic scheduler
+ */
+class Scheduler
+{
+public:
+  typedef function<void()> Event;
+
+  Scheduler(boost::asio::io_service& ioService);
+
+  /**
+   * \brief Schedule one time event after the specified delay
+   * \returns EventId that can be used to cancel the scheduled event
+   */
+  EventId
+  scheduleEvent(const time::Duration& after, const Event& event);
+
+  /**
+   * \brief Schedule periodic event that should be fired every specified period.
+   *        First event will be fired after the specified delay.
+   * \returns EventId that can be used to cancel the scheduled event
+   */
+  EventId
+  schedulePeriodicEvent(const time::Duration& after,
+                        const time::Duration& period,
+                        const Event& event);
+  
+  /**
+   * \brief Cancel scheduled event
+   */
+  void
+  cancelEvent(const EventId& eventId);
+
+private:
+  void
+  onEvent(const boost::system::error_code& code);
+  
+private:
+  boost::asio::io_service& m_ioService;
+
+  struct EventInfo
+  {
+    EventInfo(const time::Duration& after,
+              const time::Duration& period,
+              const Event& event);
+    EventInfo(const time::Point& when, const EventInfo& previousEvent);
+
+    bool
+    operator <=(const EventInfo& other) const
+    {
+      return this->m_scheduledTime <= other.m_scheduledTime;
+    }
+
+    bool
+    operator <(const EventInfo& other) const
+    {
+      return this->m_scheduledTime < other.m_scheduledTime;
+    }
+
+    time::Duration
+    expiresFromNow() const;
+    
+    time::Point m_scheduledTime;
+    time::Duration m_period;
+    Event m_event;
+    mutable EventId m_eventId;
+  };
+
+  typedef std::multiset<EventInfo> EventQueue;
+  friend struct EventIdImpl;
+
+  EventQueue m_events;
+  EventQueue::iterator m_scheduledEvent;
+  boost::asio::monotonic_deadline_timer m_deadlineTimer;
+};
+
+} // namespace ndn
+
+#endif // NFD_CORE_SCHEDULER_HPP

daemon/core/time.cpp

@@ -24,7 +24,7 @@ now()
     throw std::runtime_error("clock_gettime");
   }
   
-  return t.tv_sec * 1000000000 + t.tv_nsec;
+  return Point(time::seconds(t.tv_sec) + time::nanoseconds(t.tv_nsec));
 
 #else
   // fallback to wall clock time
@@ -36,7 +36,7 @@ now()
     throw std::runtime_error("gettimeofday");
   }
   
-  return tv.tv_sec * 1000000000 + tv.tv_usec * 1000;
+  return Point(time::seconds(tv.tv_sec) + time::microseconds(tv.tv_usec));
   
 #endif
 }

daemon/core/time.hpp

@@ -12,22 +12,149 @@
 namespace ndn {
 namespace time {
 
+class monotonic_clock;
+
 /** \class Duration
  *  \brief represents a time interval
  *  Time unit is nanosecond.
  */
-typedef int64_t Duration;
+class Duration
+{
+public:
+  Duration()
+    : m_value(0)
+  {
+  }
+
+  explicit
+  Duration(int64_t value)
+    : m_value(value)
+  {
+  }
+  
+  operator int64_t&()
+  {
+    return m_value;
+  }
+
+  operator const int64_t&() const
+  {
+    return m_value;
+  }
+
+  Duration
+  operator+(const Duration& other) const
+  {
+    return Duration(this->m_value + other.m_value);
+  }
+  
+  Duration
+  operator-(const Duration& other) const
+  {
+    return Duration(this->m_value - other.m_value);
+  }
+
+private:
+  int64_t m_value;
+};
 
 /** \class Point
  *  \brief represents a point in time
  *  This uses monotonic clock.
  */
-typedef Duration Point;
+class Point
+{
+public:
+  Point()
+    : m_value(0)
+  {
+  }
 
-/// \return{ the current time in monotonic clock }
+  explicit
+  Point(int64_t value)
+    : m_value(value)
+  {
+  }
+  
+  operator int64_t&()
+  {
+    return m_value;
+  }
+
+  operator const int64_t&() const
+  {
+    return m_value;
+  }
+
+  Point
+  operator+(const Duration& other) const
+  {
+    return Point(this->m_value + static_cast<int64_t>(other));
+  }
+  
+  Duration
+  operator-(const Point& other) const
+  {
+    return Duration(this->m_value - other.m_value);
+  }
+
+  Point
+  operator-(const Duration& other) const
+  {
+    return Point(this->m_value  - static_cast<int64_t>(other));
+  }
+  
+private:
+  int64_t m_value;
+};
+
+/**
+ * \brief Get current time
+ * \return{ the current time in monotonic clock }
+ */
 Point
 now();
 
+/**
+ * \brief Get time::Duration for the specified number of seconds
+ */
+template<class T>
+inline Duration
+seconds(T value)
+{
+  return Duration(value * static_cast<int64_t>(1000000000));
+}
+
+/**
+ * \brief Get time::Duration for the specified number of milliseconds
+ */
+template<class T>
+inline Duration
+milliseconds(T value)
+{
+  return Duration(value * static_cast<int64_t>(1000000));
+}
+
+/**
+ * \brief Get time::Duration for the specified number of microseconds
+ */
+template<class T>
+inline Duration
+microseconds(T value)
+{
+  return Duration(value * static_cast<int64_t>(1000));
+}
+
+/**
+ * \brief Get time::Duration for the specified number of nanoseconds
+ */
+inline Duration
+nanoseconds(int64_t value)
+{
+  return Duration(value);
+}
+
+
 } // namespace time
 } // namespace ndn
 

tests/core/scheduler.cpp

@@ -0,0 +1,85 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (C) 2014 Named Data Networking Project
+ * See COPYING for copyright and distribution information.
+ */
+
+#include "core/scheduler.hpp"
+
+#include <boost/test/unit_test.hpp>
+
+namespace ndn {
+
+BOOST_AUTO_TEST_SUITE(CoreScheduler)
+
+struct SchedulerFixture
+{
+  SchedulerFixture()
+    : count1(0)
+    , count2(0)
+    , count3(0)
+    , count4(0)
+  {
+  }
+    
+  void
+  event1()
+  {
+    BOOST_CHECK_EQUAL(count3, 1);
+    ++count1;
+  }
+
+  void
+  event2()
+  {
+    ++count2;
+  }
+
+  void
+  event3()
+  {
+    BOOST_CHECK_EQUAL(count1, 0);
+    ++count3;
+  }
+
+  void
+  event4()
+  {
+    ++count4;
+  }
+  
+  int count1;
+  int count2;
+  int count3;
+  int count4;
+};
+
+BOOST_FIXTURE_TEST_CASE(Events, SchedulerFixture)
+{
+  boost::asio::io_service io; 
+
+  Scheduler scheduler(io);
+  scheduler.scheduleEvent(time::seconds(0.1), bind(&SchedulerFixture::event1, this));
+  
+  EventId i = scheduler.scheduleEvent(time::seconds(0.2), bind(&SchedulerFixture::event2, this));
+  scheduler.cancelEvent(i);
+
+  scheduler.scheduleEvent(time::seconds(0.05), bind(&SchedulerFixture::event3, this));
+
+  i = scheduler.scheduleEvent(time::seconds(0.01), bind(&SchedulerFixture::event2, this));
+  scheduler.cancelEvent(i);
+
+  i = scheduler.schedulePeriodicEvent(time::seconds(0.3), time::seconds(0.1), bind(&SchedulerFixture::event4, this));
+  scheduler.scheduleEvent(time::seconds(0.69), bind(&Scheduler::cancelEvent, &scheduler, i));
+  
+  io.run();
+
+  BOOST_CHECK_EQUAL(count1, 1);
+  BOOST_CHECK_EQUAL(count2, 0);
+  BOOST_CHECK_EQUAL(count3, 1);
+  BOOST_CHECK_EQUAL(count4, 4);
+}
+
+BOOST_AUTO_TEST_SUITE_END()
+
+} // namespace ndn

tests/core/time.cpp

@@ -19,6 +19,29 @@ BOOST_AUTO_TEST_CASE(Now)
   BOOST_CHECK_LE(p1, p2);
 }
 
+BOOST_AUTO_TEST_CASE(Operations)
+{
+  // helpers
+  BOOST_CHECK_GT(time::seconds(1), time::milliseconds(1));
+  BOOST_CHECK_GT(time::milliseconds(1), time::microseconds(1));
+  BOOST_CHECK_GT(time::microseconds(1), time::nanoseconds(1));
+  
+  // duration operations + helpers
+  BOOST_CHECK_EQUAL(time::seconds(8) + time::microseconds(101), time::nanoseconds(8000101000));
+  BOOST_CHECK_EQUAL(time::seconds(7) - time::milliseconds(234), time::microseconds(6766000));
+
+  // point operations
+  time::Point p1 = time::now();
+  time::Point p2 = p1 + time::milliseconds(10000);
+  time::Point p3 = p2 - time::microseconds(5000000);
+  BOOST_CHECK_LE(p1, p2);
+  BOOST_CHECK_LE(p1, p3);
+  BOOST_CHECK_LE(p3, p2);
+  
+  BOOST_CHECK_EQUAL(p2 - p1, time::seconds(10));
+  BOOST_CHECK_EQUAL(p3 - p1, time::seconds(5));
+}
+
 BOOST_AUTO_TEST_SUITE_END()
 
 } // namespace ndn