diff --git a/events/eventdispatcher.cpp b/events/eventdispatcher.cpp index 358bc25..459051b 100644 --- a/events/eventdispatcher.cpp +++ b/events/eventdispatcher.cpp @@ -1,369 +1,371 @@ /* Copyright (C) 2013 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #include "eventdispatcher.h" #include "eventdispatcher_p.h" #ifndef DFERRY_NO_NATIVE_POLL #ifdef __linux__ #include "epolleventpoller.h" #elif defined _WIN32 #include "selecteventpoller_win32.h" #else #include "selecteventpoller_unix.h" #endif #endif #include "event.h" #include "foreigneventloopintegrator.h" #include "ieventpoller.h" #include "iioeventlistener.h" #include "platformtime.h" #include "connection_p.h" #include "timer.h" #include #include #include //#define EVENTDISPATCHER_DEBUG #ifndef DFERRY_NO_NATIVE_POLL EventDispatcher::EventDispatcher() : d(new EventDispatcherPrivate) { #ifdef __linux__ d->m_poller = new EpollEventPoller(this); #else // TODO high performance IO multiplexers for non-Linux platforms d->m_poller = new SelectEventPoller(this); #endif } #endif EventDispatcher::EventDispatcher(ForeignEventLoopIntegrator *integrator) : d(new EventDispatcherPrivate) { d->m_integrator = integrator; d->m_poller = integrator->connectToDispatcher(this); } EventDispatcherPrivate::~EventDispatcherPrivate() { - for (const std::pair &fdListener : m_ioListeners) { - fdListener.second->setEventDispatcher(nullptr); + // Make a copy because setEventDispatcher() eventually mutates the container (removes the entry) + { + std::unordered_map ioListeners = m_ioListeners; + for (const std::pair &fdListener : ioListeners) { + fdListener.second->setEventDispatcher(nullptr); + } } for (const std::pair &dt : m_timers) { dt.second->m_eventDispatcher = nullptr; dt.second->m_isRunning = false; } - if (m_integrator) { - delete m_integrator; // owns the poller (its private class!) and deletes it - } else { + if (!m_integrator) { delete m_poller; } } EventDispatcher::~EventDispatcher() { delete d; d = nullptr; } bool EventDispatcher::poll(int timeout) { int nextDue = d->timeToFirstDueTimer(); if (timeout < 0) { timeout = nextDue; } else if (nextDue >= 0) { timeout = std::min(timeout, nextDue); } #ifdef EVENTDISPATCHER_DEBUG printf("EventDispatcher::poll(): timeout=%d, nextDue=%d.\n", timeout, nextDue); #endif IEventPoller::InterruptAction interrupAction = d->m_poller->poll(timeout); if (interrupAction == IEventPoller::Stop) { return false; } else if (interrupAction == IEventPoller::ProcessAuxEvents && d->m_connectionToNotify) { d->processAuxEvents(); } d->triggerDueTimers(); return true; } void EventDispatcher::interrupt() { d->m_poller->interrupt(IEventPoller::Stop); } void EventDispatcherPrivate::wakeForEvents() { m_poller->interrupt(IEventPoller::ProcessAuxEvents); } bool EventDispatcherPrivate::addIoEventListener(IioEventListener *iol) { std::pair::iterator, bool> insertResult; insertResult = m_ioListeners.insert(std::make_pair(iol->fileDescriptor(), iol)); const bool ret = insertResult.second; if (ret) { m_poller->addIoEventListener(iol); } return ret; } bool EventDispatcherPrivate::removeIoEventListener(IioEventListener *iol) { const bool ret = m_ioListeners.erase(iol->fileDescriptor()); if (ret) { m_poller->removeIoEventListener(iol); } return ret; } void EventDispatcherPrivate::setReadWriteInterest(IioEventListener *iol, bool read, bool write) { m_poller->setReadWriteInterest(iol, read, write); } void EventDispatcherPrivate::notifyListenerForReading(FileDescriptor fd) { std::unordered_map::iterator it = m_ioListeners.find(fd); if (it != m_ioListeners.end()) { it->second->handleCanRead(); } else { #ifdef IEVENTDISPATCHER_DEBUG // while interesting for debugging, this is not an error if a connection was in the epoll // set and disconnected in its handleCanRead() or handleCanWrite() implementation std::cerr << "EventDispatcherPrivate::notifyListenerForReading(): unhandled file descriptor " << fd << ".\n"; #endif } } void EventDispatcherPrivate::notifyListenerForWriting(FileDescriptor fd) { std::unordered_map::iterator it = m_ioListeners.find(fd); if (it != m_ioListeners.end()) { it->second->handleCanWrite(); } else { #ifdef IEVENTDISPATCHER_DEBUG // while interesting for debugging, this is not an error if a connection was in the epoll // set and disconnected in its handleCanRead() or handleCanWrite() implementation std::cerr << "EventDispatcherPrivate::notifyListenerForWriting(): unhandled file descriptor " << fd << ".\n"; #endif } } int EventDispatcherPrivate::timeToFirstDueTimer() const { std::multimap::const_iterator it = m_timers.cbegin(); if (it == m_timers.cend()) { return -1; } if (it->second == nullptr) { // this is the dead entry of the currently triggered, and meanwhile removed timer if (++it == m_timers.cend()) { return -1; } } uint64 nextTimeout = it->first >> 10; uint64 currentTime = PlatformTime::monotonicMsecs(); if (currentTime >= nextTimeout) { return 0; } return nextTimeout - currentTime; } uint EventDispatcherPrivate::nextTimerSerial() { if (++m_lastTimerSerial > s_maxTimerSerial) { m_lastTimerSerial = 0; } return m_lastTimerSerial; } void EventDispatcherPrivate::addTimer(Timer *timer) { if (timer->m_tag == 0) { timer->m_tag = nextTimerSerial(); } uint64 dueTime = PlatformTime::monotonicMsecs() + uint64(timer->m_interval); // ### When a timer is added from a timer callback, make sure it only runs in the *next* // iteration of the event loop. Otherwise, endless cascades of timers triggering, adding // more timers etc could occur without ever returning from triggerDueTimers(). // For the condition for this hazard, see "invariant:" in triggerDueTimers(): the only way // the new timer could trigger in this event loop iteration is when: // // m_triggerTime == currentTime(before call to trigger()) == timerAddedInTrigger().dueTime // // note: m_triggeredTimer.dueTime < m_triggerTime is well possible; if ==, the additional // condition applies that timerAddedInTrigger().serial >= m_triggeredTimer.serial; // we ignore this and do it conservatively and less complicated. // (the additional condition comes from serials as keys and that each "slot" in multimap with // the same keys is a list where new entries are back-inserted) // // As a countermeasure, tweak the new timer's timeout, putting it well before m_triggeredTimer's // iterator position in the multimap... because the new timer must have zero timeout in order for // its due time to occur within this triggerDueTimers() iteration, it is supposed to trigger ASAP // anyway. This disturbs the order of triggering a little compared to the usual, but all // timeouts are properly respected - the next event loop iteration is guaranteed to trigger // timers at times strictly greater-equal than this iteration (time goes only one way) ;) if (m_triggerTime && dueTime == m_triggerTime) { dueTime = m_triggerTime - 1; } timer->m_tag = (dueTime << 10) + (timer->m_tag & s_maxTimerSerial); m_timers.emplace(timer->m_tag, timer); maybeSetTimeoutForIntegrator(); } void EventDispatcherPrivate::removeTimer(Timer *timer) { assert(timer->m_tag != 0); // We cannot toggle m_isTriggeredTimerPendingRemoval back and forth, we can only set it once. // Because after the timer has been removed once, the next time we see the same pointer value, // it could be an entirely different timer. Consider this: // delete timer1; // calls removeTimer() // Timer *timer2 = new Timer(); // accidentally gets same memory address as timer1 // timer2->start(...); // timer2->stop(); // timer == m_triggeredTimer, uh oh // The last line does not necessarily cause a problem, but just don't be excessively clever. // On the other hand, not special-casing the currently triggered timer after it has been marked // for removal once is fine. In case it is re-added, it gets a new map entry in addTimer() // and from then on it can be handled like any other timer. bool removingTriggeredTimer = false; if (!m_isTriggeredTimerPendingRemoval && timer == m_triggeredTimer) { // using this variable, we can avoid dereferencing m_triggeredTimer should it have been // deleted while triggered m_isTriggeredTimerPendingRemoval = true; removingTriggeredTimer = true; } auto iterRange = m_timers.equal_range(timer->m_tag); for (; iterRange.first != iterRange.second; ++iterRange.first) { if (iterRange.first->second == timer) { if (!removingTriggeredTimer) { m_timers.erase(iterRange.first); } else { // mark it as dead for query methods such as timeToFirstDueTimer() iterRange.first->second = nullptr; } maybeSetTimeoutForIntegrator(); return; } } assert(false); // the timer should never request a remove when it has not been added } void EventDispatcherPrivate::maybeSetTimeoutForIntegrator() { if (m_integrator) { m_integrator->watchTimeout(timeToFirstDueTimer()); } } void EventDispatcherPrivate::triggerDueTimers() { m_triggerTime = PlatformTime::monotonicMsecs(); for (auto it = m_timers.begin(); it != m_timers.end();) { const uint64 timerTimeout = (it->first >> 10); if (timerTimeout > m_triggerTime) { break; } // careful here - protect against adding and removing any timer while inside its trigger()! // we do this by keeping the iterator at the current position (so changing any other timer // doesn't invalidate it) and blocking changes to the timer behind that iterator // (so we don't mess with its data should it have been deleted outright in the callback) m_triggeredTimer = it->second; Timer *const timer = m_triggeredTimer; m_isTriggeredTimerPendingRemoval = false; // invariant: // m_triggeredTimer.dueTime <= m_triggerTime <= currentTime(here) <= .dueTime timer->trigger(); m_triggeredTimer = nullptr; if (!m_isTriggeredTimerPendingRemoval && timer->m_isRunning) { // ### we are rescheduling timers based on triggerTime even though real time can be // much later - is this the desired behavior? I think so... if (timer->m_interval == 0) { // With the other branch we might iterate over this timer again in this invocation because // if there are several timers with the same tag, this entry will be back-inserted into the // list of values for the current tag / key slot. We only break out of the loop if // timerTimeout > m_triggerTime, so there would be an infinite loop. // Instead, we just leave the iterator alone, which does not put it in front of the current // iterator position. It's also good for performance. Win-win! ++it; } else { timer->m_tag = ((m_triggerTime + uint64(timer->m_interval)) << 10) + (timer->m_tag & s_maxTimerSerial); m_timers.erase(it++); m_timers.emplace(timer->m_tag, timer); } } else { m_timers.erase(it++); } } m_triggerTime = 0; maybeSetTimeoutForIntegrator(); } void EventDispatcherPrivate::queueEvent(std::unique_ptr evt) { // std::cerr << "EventDispatcherPrivate::queueEvent() " << evt->type << " " << this << std::endl; { SpinLocker locker(&m_queuedEventsLock); m_queuedEvents.emplace_back(std::move(evt)); } wakeForEvents(); } void EventDispatcherPrivate::processAuxEvents() { // std::cerr << "EventDispatcherPrivate::processAuxEvents() " << this << std::endl; // don't hog the lock while processing the events std::vector> events; { SpinLocker locker(&m_queuedEventsLock); std::swap(events, m_queuedEvents); } if (m_connectionToNotify) { for (const std::unique_ptr &evt : events) { m_connectionToNotify->processEvent(evt.get()); } } } diff --git a/events/eventdispatcher.h b/events/eventdispatcher.h index 35d4484..e4b3cc6 100644 --- a/events/eventdispatcher.h +++ b/events/eventdispatcher.h @@ -1,53 +1,54 @@ /* Copyright (C) 2013 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #ifndef EVENTDISPATCHER_H #define EVENTDISPATCHER_H #include "export.h" class EventDispatcherPrivate; class ForeignEventLoopIntegrator; class DFERRY_EXPORT EventDispatcher { public: #ifndef DFERRY_NO_NATIVE_POLL EventDispatcher(); #endif + // Does not take ownership of (i.e. does not delete in ~EventDispatcher()) integrator EventDispatcher(ForeignEventLoopIntegrator *integrator); ~EventDispatcher(); EventDispatcher(EventDispatcher &other) = delete; void operator=(EventDispatcher &other) = delete; bool poll(int timeout = -1); // returns false if interrupted by interrupt() // Asynchronously interrupt the waiting for events, i.e. at the current (if any) or next poll - this is // explicitly allowed to be called from any thread (including its own). void interrupt(); private: friend class EventDispatcherPrivate; EventDispatcherPrivate *d; }; #endif // EVENTDISPATCHER_H