Add a schedule_in_range method

- Schedules timer in optimized slot, within the provided range
  - If timeout is already in right range, do nothing.
  - If timer can be scheduled at a slot boundary such that it does
    not need to be promoted to another wheel, do that.
  - In general prefer scheduling as late in the range as possible.

src/test/test_basic.cc

@@ -207,6 +207,54 @@ bool test_ticks_to_next_event() {
     return true;
 }
 
+bool test_schedule_in_range() {
+    typedef std::function<void()> Callback;
+    TimerWheel timers;
+    TimerEvent<Callback> timer([] () { });
+
+    // No useful rounding possible.
+    timers.schedule_in_range(&timer, 281, 290);
+    EXPECT_INTEQ(timers.ticks_to_next_event(), 290);
+
+    // Pick a time aligned at slot boundary if possible.
+    timers.schedule_in_range(&timer, 256*4 - 1, 256*5 - 1);
+    EXPECT_INTEQ(timers.ticks_to_next_event(), 256*4);
+
+    timers.schedule_in_range(&timer, 256*4 + 1, 256*5);
+    EXPECT_INTEQ(timers.ticks_to_next_event(), 256*5);
+
+    // Event already in right range.
+    timers.schedule_in_range(&timer, 256*1, 256*10);
+    EXPECT_INTEQ(timers.ticks_to_next_event(), 256*5);
+
+    // Event canceled, but was previously scheduled in
+    // the right range. Should be ignored, and scheduled
+    // as normal to the end of the range.
+    timer.cancel();
+    timers.schedule_in_range(&timer, 256*1, 256*10);
+    EXPECT_INTEQ(timers.ticks_to_next_event(), 256*10);
+
+    // Make sure the decision on whether timer is in range or
+    // not is done based on absolute ticks, not relative ticks.
+    timers.advance(256*9);
+    EXPECT_INTEQ(timers.ticks_to_next_event(), 256*1);
+    timers.schedule_in_range(&timer, 256*9, 256*10);
+    EXPECT_INTEQ(timers.ticks_to_next_event(), 256*10);
+
+    // Try scheduling timers in random ranges.
+    for (int i = 0; i < 10000; ++i) {
+        int len1 = rand() % 20;
+        int len2 = rand() % 20;
+        int r1 = rand() % (1 << len1);
+        int r2 = r1 + (1 + rand() % (1 << len2));
+        timers.schedule_in_range(&timer, r1, r2);
+        EXPECT(timers.ticks_to_next_event() >= r1);
+        EXPECT(timers.ticks_to_next_event() <= r2);
+    }
+
+    return true;
+}
+
 bool test_reschedule_from_timer() {
     typedef std::function<void()> Callback;
     TimerWheel timers;
@@ -296,6 +344,7 @@ int main(void) {
     TEST(test_single_timer_no_hierarchy);
     TEST(test_single_timer_hierarchy);
     TEST(test_ticks_to_next_event);
+    TEST(test_schedule_in_range);
     TEST(test_single_timer_random);
     TEST(test_reschedule_from_timer);
     TEST(test_timeout_method);

src/timer-wheel.h

@@ -160,6 +160,33 @@ public:
         event->relink(slot);
     }
 
+    void schedule_in_range(TimerEventInterface* event,
+                           Tick start, Tick end) {
+        assert(end > start);
+        if (event->active()) {
+            auto current = event->scheduled_at() - now_;
+            // Event is already scheduled to happen in this range. Instead
+            // of always using the old slot, we could check compute the
+            // new slot and switch iff it's aligned better than the old one.
+            // But it seems hard to believe that could be worthwhile.
+            if (current >= start && current <= end) {
+                return;
+            }
+        }
+
+        // Zero as many bits (in WIDTH_BITS chunks) as possible
+        // from "end" while still keeping the output in the
+        // right range.
+        Tick mask = ~0;
+        while ((start & mask) != (end & mask)) {
+            mask = (mask << WIDTH_BITS);
+        }
+
+        Tick delta = end & (mask >> WIDTH_BITS);
+
+        schedule(event, delta);
+    }
+
     // Return the current tick value. Note that if the timers advance by
     // multiple ticks during a single call to advance(), the value of now()
     // will be the tick on which the timer was first run. Not the tick that