faux-folly/folly/futures/detail/ThreadWheelTimekeeper.cpp

/*
 * Copyright 2015 Facebook, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#include "ThreadWheelTimekeeper.h"

#include <folly/futures/Future.h>
#include <future>
#include <memory>
#include <vector>

namespace folly { namespace detail {

using std::vector;

namespace {
  using lock_guard = std::lock_guard<std::mutex>;
  using unique_lock = std::unique_lock<std::mutex>;
  using std::make_shared;

  std::mutex g_timekeeperSingletonCreateionMutex_;
  ThreadWheelTimekeeper *g_timekeeperSingleton_{};
} // namespace


ThreadWheelTimekeeper::ThreadWheelTimekeeper() :
    timerMap_(std::chrono::milliseconds(1)),
    thread_([this]{ thread_run(); })
{
}

ThreadWheelTimekeeper::~ThreadWheelTimekeeper() {
    unique_lock lock(mutex_);
    running_ = false;
    cond_.notify_one();
    lock.unlock();

    thread_.join();
}

void ThreadWheelTimekeeper::thread_run()
{
    unique_lock lock(mutex_);

    while (running_) {
        if (timerMap_.empty()) {
            cond_.wait(lock);
            continue;
        }

        auto tp = timerMap_.nextTimeout();
        cond_.wait_until(lock, tp);

        vector<TimerMap::data_type> vec;
        timerMap_.popExpiredTimers(std::inserter(vec, vec.begin()));
        lock.unlock();
        TimerMap::invokeTimers(vec.begin(), vec.end());
        lock.lock();
    }
}

Future<Unit> ThreadWheelTimekeeper::after(Duration dur) {
    // first: the promise, second: a cancellation
    using two_type = std::pair< Promise<Unit>, unsigned >;
    auto twoptr = make_shared<two_type>(Promise<Unit>(), 0);
    auto& prom = twoptr->first;
    auto& cx = twoptr->second;

    prom.setInterruptHandler([twoptr, &prom, &cx] (exception_wrapper const& w) {
            cx = 1;
            prom.setException(w);
        });

    auto f = prom.getFuture();

    lock_guard lock(mutex_);
    timerMap_.scheduleTimeout(dur, [twoptr, &prom, &cx] () {
            if (cx) {
                return;
            }
            prom.setValue();
        });
    cond_.notify_one();

    return f;
}

Timekeeper* getTimekeeperSingleton() {
    if (!g_timekeeperSingleton_)
    {
        lock_guard lock{g_timekeeperSingletonCreateionMutex_};
        if (!g_timekeeperSingleton_)
        {
            g_timekeeperSingleton_ = new ThreadWheelTimekeeper;
        }
    }

    return g_timekeeperSingleton_;
}

}} // folly::detail