--- /dev/null
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 <thrift/concurrency/TimerManager.h>
+#include <thrift/concurrency/Exception.h>
+
+#include <assert.h>
+#include <iostream>
+#include <memory>
+#include <set>
+
+namespace apache {
+namespace thrift {
+namespace concurrency {
+
+using std::shared_ptr;
+using std::weak_ptr;
+
+/**
+ * TimerManager class
+ *
+ * @version $Id:$
+ */
+class TimerManager::Task : public Runnable {
+
+public:
+ enum STATE { WAITING, EXECUTING, CANCELLED, COMPLETE };
+
+ Task(shared_ptr<Runnable> runnable) : runnable_(runnable), state_(WAITING) {}
+
+ ~Task() override = default;
+
+ void run() override {
+ if (state_ == EXECUTING) {
+ runnable_->run();
+ state_ = COMPLETE;
+ }
+ }
+
+ bool operator==(const shared_ptr<Runnable> & runnable) const { return runnable_ == runnable; }
+
+ task_iterator it_;
+
+private:
+ shared_ptr<Runnable> runnable_;
+ friend class TimerManager::Dispatcher;
+ STATE state_;
+};
+
+class TimerManager::Dispatcher : public Runnable {
+
+public:
+ Dispatcher(TimerManager* manager) : manager_(manager) {}
+
+ ~Dispatcher() override = default;
+
+ /**
+ * Dispatcher entry point
+ *
+ * As long as dispatcher thread is running, pull tasks off the task taskMap_
+ * and execute.
+ */
+ void run() override {
+ {
+ Synchronized s(manager_->monitor_);
+ if (manager_->state_ == TimerManager::STARTING) {
+ manager_->state_ = TimerManager::STARTED;
+ manager_->monitor_.notifyAll();
+ }
+ }
+
+ do {
+ std::set<shared_ptr<TimerManager::Task> > expiredTasks;
+ {
+ Synchronized s(manager_->monitor_);
+ task_iterator expiredTaskEnd;
+ auto now = std::chrono::steady_clock::now();
+ while (manager_->state_ == TimerManager::STARTED
+ && (expiredTaskEnd = manager_->taskMap_.upper_bound(now))
+ == manager_->taskMap_.begin()) {
+ std::chrono::milliseconds timeout(0);
+ if (!manager_->taskMap_.empty()) {
+ timeout = std::chrono::duration_cast<std::chrono::milliseconds>(manager_->taskMap_.begin()->first - now);
+ //because the unit of steady_clock is smaller than millisecond,timeout may be 0.
+ if (timeout.count() == 0) {
+ timeout = std::chrono::milliseconds(1);
+ }
+ manager_->monitor_.waitForTimeRelative(timeout);
+ } else {
+ manager_->monitor_.waitForTimeRelative(0);
+ }
+ now = std::chrono::steady_clock::now();
+ }
+
+ if (manager_->state_ == TimerManager::STARTED) {
+ for (auto ix = manager_->taskMap_.begin(); ix != expiredTaskEnd; ix++) {
+ shared_ptr<TimerManager::Task> task = ix->second;
+ expiredTasks.insert(task);
+ task->it_ = manager_->taskMap_.end();
+ if (task->state_ == TimerManager::Task::WAITING) {
+ task->state_ = TimerManager::Task::EXECUTING;
+ }
+ manager_->taskCount_--;
+ }
+ manager_->taskMap_.erase(manager_->taskMap_.begin(), expiredTaskEnd);
+ }
+ }
+
+ for (const auto & expiredTask : expiredTasks) {
+ expiredTask->run();
+ }
+
+ } while (manager_->state_ == TimerManager::STARTED);
+
+ {
+ Synchronized s(manager_->monitor_);
+ if (manager_->state_ == TimerManager::STOPPING) {
+ manager_->state_ = TimerManager::STOPPED;
+ manager_->monitor_.notifyAll();
+ }
+ }
+ return;
+ }
+
+private:
+ TimerManager* manager_;
+ friend class TimerManager;
+};
+
+#if defined(_MSC_VER)
+#pragma warning(push)
+#pragma warning(disable : 4355) // 'this' used in base member initializer list
+#endif
+
+TimerManager::TimerManager()
+ : taskCount_(0),
+ state_(TimerManager::UNINITIALIZED),
+ dispatcher_(std::make_shared<Dispatcher>(this)) {
+}
+
+#if defined(_MSC_VER)
+#pragma warning(pop)
+#endif
+
+TimerManager::~TimerManager() {
+
+ // If we haven't been explicitly stopped, do so now. We don't need to grab
+ // the monitor here, since stop already takes care of reentrancy.
+
+ if (state_ != STOPPED) {
+ try {
+ stop();
+ } catch (...) {
+ // We're really hosed.
+ }
+ }
+}
+
+void TimerManager::start() {
+ bool doStart = false;
+ {
+ Synchronized s(monitor_);
+ if (!threadFactory_) {
+ throw InvalidArgumentException();
+ }
+ if (state_ == TimerManager::UNINITIALIZED) {
+ state_ = TimerManager::STARTING;
+ doStart = true;
+ }
+ }
+
+ if (doStart) {
+ dispatcherThread_ = threadFactory_->newThread(dispatcher_);
+ dispatcherThread_->start();
+ }
+
+ {
+ Synchronized s(monitor_);
+ while (state_ == TimerManager::STARTING) {
+ monitor_.wait();
+ }
+ assert(state_ != TimerManager::STARTING);
+ }
+}
+
+void TimerManager::stop() {
+ bool doStop = false;
+ {
+ Synchronized s(monitor_);
+ if (state_ == TimerManager::UNINITIALIZED) {
+ state_ = TimerManager::STOPPED;
+ } else if (state_ != STOPPING && state_ != STOPPED) {
+ doStop = true;
+ state_ = STOPPING;
+ monitor_.notifyAll();
+ }
+ while (state_ != STOPPED) {
+ monitor_.wait();
+ }
+ }
+
+ if (doStop) {
+ // Clean up any outstanding tasks
+ taskMap_.clear();
+
+ // Remove dispatcher's reference to us.
+ dispatcher_->manager_ = nullptr;
+ }
+}
+
+shared_ptr<const ThreadFactory> TimerManager::threadFactory() const {
+ Synchronized s(monitor_);
+ return threadFactory_;
+}
+
+void TimerManager::threadFactory(shared_ptr<const ThreadFactory> value) {
+ Synchronized s(monitor_);
+ threadFactory_ = value;
+}
+
+size_t TimerManager::taskCount() const {
+ return taskCount_;
+}
+
+TimerManager::Timer TimerManager::add(shared_ptr<Runnable> task, const std::chrono::milliseconds &timeout) {
+ return add(task, std::chrono::steady_clock::now() + timeout);
+}
+
+TimerManager::Timer TimerManager::add(shared_ptr<Runnable> task,
+ const std::chrono::time_point<std::chrono::steady_clock>& abstime) {
+ auto now = std::chrono::steady_clock::now();
+
+ if (abstime < now) {
+ throw InvalidArgumentException();
+ }
+ Synchronized s(monitor_);
+ if (state_ != TimerManager::STARTED) {
+ throw IllegalStateException();
+ }
+
+ // If the task map is empty, we will kick the dispatcher for sure. Otherwise, we kick him
+ // if the expiration time is shorter than the current value. Need to test before we insert,
+ // because the new task might insert at the front.
+ bool notifyRequired = (taskCount_ == 0) ? true : abstime < taskMap_.begin()->first;
+
+ shared_ptr<Task> timer(new Task(task));
+ taskCount_++;
+ timer->it_ = taskMap_.emplace(abstime, timer);
+
+ // If the task map was empty, or if we have an expiration that is earlier
+ // than any previously seen, kick the dispatcher so it can update its
+ // timeout
+ if (notifyRequired) {
+ monitor_.notify();
+ }
+
+ return timer;
+}
+
+void TimerManager::remove(shared_ptr<Runnable> task) {
+ Synchronized s(monitor_);
+ if (state_ != TimerManager::STARTED) {
+ throw IllegalStateException();
+ }
+ bool found = false;
+ for (auto ix = taskMap_.begin(); ix != taskMap_.end();) {
+ if (*ix->second == task) {
+ found = true;
+ taskCount_--;
+ taskMap_.erase(ix++);
+ } else {
+ ++ix;
+ }
+ }
+ if (!found) {
+ throw NoSuchTaskException();
+ }
+}
+
+void TimerManager::remove(Timer handle) {
+ Synchronized s(monitor_);
+ if (state_ != TimerManager::STARTED) {
+ throw IllegalStateException();
+ }
+
+ shared_ptr<Task> task = handle.lock();
+ if (!task) {
+ throw NoSuchTaskException();
+ }
+
+ if (task->it_ == taskMap_.end()) {
+ // Task is being executed
+ throw UncancellableTaskException();
+ }
+
+ taskMap_.erase(task->it_);
+ taskCount_--;
+}
+
+TimerManager::STATE TimerManager::state() const {
+ return state_;
+}
+}
+}
+} // apache::thrift::concurrency
projects / ceph.git / blobdiff