X-Git-Url: http://g0dil.de/git?a=blobdiff_plain;f=Scheduler%2FFIFORunner.cc;h=a6fb824aa8be3d061285f19df8ad9b2a091236b0;hb=f2f5d59e83863f3b513950173baee1b6da2aee3c;hp=8a1293b1a082500b20ca395f6c2a0033311a04ba;hpb=919e588a2c387c9a910aa8761e65155a0d205bba;p=senf.git diff --git a/Scheduler/FIFORunner.cc b/Scheduler/FIFORunner.cc index 8a1293b..a6fb824 100644 --- a/Scheduler/FIFORunner.cc +++ b/Scheduler/FIFORunner.cc @@ -27,12 +27,104 @@ //#include "FIFORunner.ih" // Custom includes +#include +#include +#include +#include "../Utils/Exception.hh" +#include "../Utils/senfassert.hh" +#include "../Utils/ScopeExit.hh" //#include "FIFORunner.mpp" #define prefix_ ///////////////////////////////cc.p//////////////////////////////////////// -prefix_ void senf::scheduler::FIFORunner::dequeue(TaskInfo * task) +prefix_ senf::scheduler::detail::FIFORunner::FIFORunner() + : tasks_ (), next_ (tasks_.end()), watchdogRunning_ (false), watchdogMs_ (1000), + watchdogAbort_ (false), watchdogCount_(0), hangCount_ (0) +{ + 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; + } +} + +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). +// +// 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 +// some threshold (e.g. 2^31 -1 or some such), the task list is traversed from beginning to end to +// assign new consecutive order values. This O(N) operation is so seldom, that it is amortized over +// a very long time. +// +// With this value at hand, we can do several optimizations: One idea would be the following: The +// runnable set always has two types of tasks: There are tasks, which are heavily active and are +// signaled constantly and other tasks which lie dormant most of the time. Those dormant tasks will +// end up at the beginning of the task queue. +// +// With the above defined 'ordering' field available, we can manage an iterator pointing to the +// first and the last runnable task. This will often help a lot since the group of runnable tasks +// 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::detail::FIFORunner::dequeue(TaskInfo * task) { TaskList::iterator i (TaskList::current(*task)); if (next_ == i) @@ -40,43 +132,100 @@ prefix_ void senf::scheduler::FIFORunner::dequeue(TaskInfo * task) tasks_.erase(i); } -namespace { - struct NullTask - : public senf::scheduler::FIFORunner::TaskInfo - { - void run() {}; - }; +prefix_ void senf::scheduler::detail::FIFORunner::run() +{ + TaskList::iterator f (tasks_.begin()); + TaskList::iterator l (TaskList::current(highPriorityEnd_)); + run(f, l); + + f = l; ++f; + l = TaskList::current(normalPriorityEnd_); + run(f, l); + + f = l; ++f; + l = tasks_.end(); + run(f, l); } -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; task.run(); } 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(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////////////////////////////////////////