//#include "FIFORunner.ih"
// Custom includes
+#include <signal.h>
+#include <time.h>
+#include <boost/lambda/lambda.hpp>
+#include "../Utils/Exception.hh"
+#include "../Utils/senfassert.hh"
+#include "../Utils/ScopeExit.hh"
//#include "FIFORunner.mpp"
#define prefix_
///////////////////////////////cc.p////////////////////////////////////////
+prefix_ senf::scheduler::detail::FIFORunner::FIFORunner()
+ : tasks_ (), next_ (tasks_.end()), watchdogRunning_ (false), watchdogMs_ (1000),
+ watchdogAbort_ (false), watchdogCount_(0), hangCount_ (0), yield_ (false)
+{
+ struct sigevent ev;
+ ::memset(&ev, 0, sizeof(ev));
+ ev.sigev_notify = SIGEV_SIGNAL;
+ ev.sigev_signo = SIGURG;
+ ev.sigev_value.sival_ptr = this;
+ if (timer_create(CLOCK_MONOTONIC, &ev, &watchdogId_) < 0)
+ SENF_THROW_SYSTEM_EXCEPTION("timer_create()");
+
+ struct sigaction sa;
+ ::memset(&sa, 0, sizeof(sa));
+ sa.sa_sigaction = &watchdog;
+ sa.sa_flags = SA_SIGINFO;
+ if (sigaction(SIGURG, &sa, 0) < 0)
+ SENF_THROW_SYSTEM_EXCEPTION("sigaction()");
+
+ sigset_t mask;
+ sigemptyset(&mask);
+ sigaddset(&mask, SIGURG);
+ if (sigprocmask(SIG_UNBLOCK, &mask, 0) < 0)
+ SENF_THROW_SYSTEM_EXCEPTION("sigprocmask()");
+
+ tasks_.push_back(highPriorityEnd_);
+ tasks_.push_back(normalPriorityEnd_);
+}
+
+prefix_ senf::scheduler::detail::FIFORunner::~FIFORunner()
+{
+ timer_delete(watchdogId_);
+ signal(SIGURG, SIG_DFL);
+}
+
+prefix_ void senf::scheduler::detail::FIFORunner::startWatchdog()
+{
+ if (watchdogMs_ > 0) {
+ struct itimerspec timer;
+ ::memset(&timer, 0, sizeof(timer));
+
+ timer.it_interval.tv_sec = watchdogMs_ / 1000;
+ timer.it_interval.tv_nsec = (watchdogMs_ % 1000) * 1000000ul;
+ timer.it_value.tv_sec = timer.it_interval.tv_sec;
+ timer.it_value.tv_nsec = timer.it_interval.tv_nsec;
+
+ if (timer_settime(watchdogId_, 0, &timer, 0) < 0)
+ SENF_THROW_SYSTEM_EXCEPTION("timer_settime()");
+
+ watchdogRunning_ = true;
+ }
+ else
+ stopWatchdog();
+}
+
+prefix_ void senf::scheduler::detail::FIFORunner::stopWatchdog()
+{
+ struct itimerspec timer;
+ ::memset(&timer, 0, sizeof(timer));
+
+ if (timer_settime(watchdogId_, 0, &timer, 0) < 0)
+ SENF_THROW_SYSTEM_EXCEPTION("timer_settime()");
+
+ watchdogRunning_ = false;
+}
+
// At the moment, the FIFORunner is not very efficient with many non-runnable tasks since the
// complete list of tasks is traversed on each run().
//
// To optimize this, we woould need a way to find the relative ordering of two tasks in O(1) (at the
-// moment, this is an O)(N) operation by traversing the list).
+// moment, this is an O(N) operation by traversing the list).
//
// One idea is, to give each task an 'order' value. Whenever a task is added at the end, it's order
// value is set to the order value of the last task + 1. Whenever the order value such added exceeds
// will mostly be localized to the end of the queue. only occasionally one of the dormant tasks will
// be runnable. This additional traversal time will be amortized over a larger time.
-prefix_ void senf::scheduler::FIFORunner::dequeue(TaskInfo * task)
+prefix_ void senf::scheduler::detail::FIFORunner::dequeue(TaskInfo * task)
{
TaskList::iterator i (TaskList::current(*task));
if (next_ == i)
tasks_.erase(i);
}
-namespace {
- struct NullTask
- : public senf::scheduler::FIFORunner::TaskInfo
- {
- void run() {};
- };
+prefix_ void senf::scheduler::detail::FIFORunner::run()
+{
+ for(;;) {
+ TaskList::iterator f (tasks_.begin());
+ TaskList::iterator l (TaskList::current(highPriorityEnd_));
+ run(f, l);
+ if (yield_) {
+ yield_ = false;
+ continue;
+ }
+
+ f = l; ++f;
+ l = TaskList::current(normalPriorityEnd_);
+ run(f, l);
+ if (yield_) {
+ yield_ = false;
+ continue;
+ }
+
+ f = l; ++f;
+ l = tasks_.end();
+ run(f, l);
+ if (yield_) {
+ yield_ = false;
+ continue;
+ }
+ break;
+ }
}
-prefix_ void senf::scheduler::FIFORunner::run()
+prefix_ void senf::scheduler::detail::FIFORunner::run(TaskList::iterator f, TaskList::iterator l)
{
+ if (f == l)
+ // We'll have problems inserting NullTask between f and l below, so just explicitly bail out
+ return;
+
// This algorithm is carefully adjusted to make it work even when arbitrary tasks are removed
// from the queue
// - Before we begin, we add a NullTask to the queue. The only purpose of this node is, to mark
// the current end of the queue. The iterator to this node becomes the end iterator of the
// range to process
- // - We update the TaskInfo and move it to the end of the queue before calling the callback so
+ // - We update the TaskInfo and move it to the next queue Element before calling the callback so
// we don't access the TaskInfo if it is removed while the callback is running
// - We keep the next to-be-processed node in a class variable which is checked and updated
// whenever a node is removed.
+
NullTask null;
- tasks_.push_back(null);
+ tasks_.insert(l, null);
TaskList::iterator end (TaskList::current(null));
- next_ = tasks_.begin();
+ next_ = f;
+
+ using namespace boost::lambda;
+ ScopeExit atExit ((
+ var(watchdogCount_) = 0,
+ var(next_) = l
+ ));
+
while (next_ != end) {
TaskInfo & task (*next_);
- if (task.runnable) {
- task.runnable = false;
+ if (task.runnable_) {
+ task.runnable_ = false;
+ runningName_ = task.name();
+# ifdef SENF_DEBUG
+ runningBacktrace_ = task.backtrace_;
+# endif
TaskList::iterator i (next_);
++ next_;
- tasks_.splice(tasks_.end(), tasks_, i);
+ tasks_.splice(l, tasks_, i);
+ watchdogCount_ = 1;
+ yield_ = false;
task.run();
+ if (yield_)
+ return;
}
else
++ next_;
}
- tasks_.erase(end);
- next_ = tasks_.end();
+}
+
+prefix_ senf::scheduler::detail::FIFORunner::TaskList::iterator
+senf::scheduler::detail::FIFORunner::priorityEnd(TaskInfo::Priority p)
+{
+ switch (p) {
+ case senf::scheduler::detail::FIFORunner::TaskInfo::PRIORITY_LOW :
+ return tasks_.end();
+ case senf::scheduler::detail::FIFORunner::TaskInfo::PRIORITY_NORMAL :
+ return TaskList::current(normalPriorityEnd_);
+ case senf::scheduler::detail::FIFORunner::TaskInfo::PRIORITY_HIGH :
+ return TaskList::current(highPriorityEnd_);
+ }
+ return tasks_.begin();
+}
+
+prefix_ void senf::scheduler::detail::FIFORunner::watchdog(int, siginfo_t * si, void *)
+{
+ FIFORunner & runner (*static_cast<FIFORunner *>(si->si_value.sival_ptr));
+ if (runner.watchdogCount_ > 0) {
+ ++ runner.watchdogCount_;
+ if (runner.watchdogCount_ > 2) {
+ ++ runner.hangCount_;
+ write(1, "\n\n*** Scheduler task hanging: ", 30);
+ write(1, runner.runningName_.c_str(), runner.runningName_.size());
+ write(1, "\n", 1);
+#ifdef SENF_DEBUG
+ write(1, "Task was initialized at\n", 24);
+ write(1, runner.runningBacktrace_.c_str(), runner.runningBacktrace_.size());
+#endif
+ write(1, "\n", 1);
+ if (runner.watchdogAbort_)
+ assert(false);
+ }
+ }
}
///////////////////////////////cc.e////////////////////////////////////////
projects / senf.git / blobdiff