/* * * 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 "qpid/log/Logger.h" #include "qpid/sys/Poller.h" #include "qpid/sys/IOHandle.h" #include "qpid/sys/Mutex.h" #include "qpid/sys/DeletionManager.h" #include "qpid/sys/posix/check.h" #include "qpid/sys/posix/PrivatePosix.h" #include #include #include #include #include #include //TODO: Remove this #include "qpid/sys/Dispatcher.h" namespace qpid { namespace sys { // Deletion manager to handle deferring deletion of PollerHandles to when they definitely aren't being used DeletionManager PollerHandleDeletionManager; // Instantiate (and define) class static for DeletionManager template <> DeletionManager::AllThreadsStatuses DeletionManager::allThreadsStatuses(0); class PollerHandlePrivate { friend class Poller; friend class PollerHandle; enum FDStat { ABSENT, MONITORED, INACTIVE, HUNGUP, MONITORED_HUNGUP }; int fd; uint32_t events; FDStat stat; Mutex lock; PollerHandlePrivate(int f) : fd(f), events(0), stat(ABSENT) { } bool isActive() const { return stat == MONITORED || stat == MONITORED_HUNGUP; } void setActive() { stat = (stat == HUNGUP) ? MONITORED_HUNGUP : MONITORED; } bool isInactive() const { return stat == INACTIVE || stat == HUNGUP; } void setInactive() { stat = INACTIVE; } bool isIdle() const { return stat == ABSENT; } void setIdle() { stat = ABSENT; } bool isHungup() const { return stat == MONITORED_HUNGUP || stat == HUNGUP; } void setHungup() { assert(stat == MONITORED); stat = HUNGUP; } }; PollerHandle::PollerHandle(const IOHandle& h) : impl(new PollerHandlePrivate(toFd(h.impl))) {} PollerHandle::~PollerHandle() { delete impl; } void PollerHandle::deferDelete() { PollerHandleDeletionManager.markForDeletion(this); } /** * Concrete implementation of Poller to use the Solaris Event Completion * Framework interface */ class PollerPrivate { friend class Poller; const int portId; static uint32_t directionToPollEvent(Poller::Direction dir) { switch (dir) { case Poller::INPUT: return POLLIN; case Poller::OUTPUT: return POLLOUT; case Poller::INOUT: return POLLIN | POLLOUT; default: return 0; } } static Poller::EventType pollToDirection(uint32_t events) { uint32_t e = events & (POLLIN | POLLOUT); switch (e) { case POLLIN: return Poller::READABLE; case POLLOUT: return Poller::WRITABLE; case POLLIN | POLLOUT: return Poller::READ_WRITABLE; default: return (events & (POLLHUP | POLLERR)) ? Poller::DISCONNECTED : Poller::INVALID; } } PollerPrivate() : portId(::port_create()) { } ~PollerPrivate() { } }; void Poller::addFd(PollerHandle& handle, Direction dir) { PollerHandlePrivate& eh = *handle.impl; ScopedLock l(eh.lock); uint32_t events = 0; if (eh.isIdle()) { events = PollerPrivate::directionToPollEvent(dir); } else { assert(eh.isActive()); events = eh.events | PollerPrivate::directionToPollEvent(dir); } //port_associate can be used to add an association or modify an //existing one QPID_POSIX_CHECK(::port_associate(impl->portId, PORT_SOURCE_FD, (uintptr_t) eh.fd, events, &handle)); eh.events = events; eh.setActive(); QPID_LOG(trace, "Poller::addFd(handle=" << &handle << "[" << typeid(&handle).name() << "], fd=" << eh.fd << ")"); //assert(dynamic_cast(&handle)); } void Poller::delFd(PollerHandle& handle) { PollerHandlePrivate& eh = *handle.impl; ScopedLock l(eh.lock); assert(!eh.isIdle()); int rc = ::port_dissociate(impl->portId, PORT_SOURCE_FD, (uintptr_t) eh.fd); //Allow closing an invalid fd, allowing users to close fd before //doing delFd() if (rc == -1 && errno != EBADFD) { QPID_POSIX_CHECK(rc); } eh.setIdle(); QPID_LOG(trace, "Poller::delFd(handle=" << &handle << ", fd=" << eh.fd << ")"); } // modFd is equivalent to delFd followed by addFd void Poller::modFd(PollerHandle& handle, Direction dir) { PollerHandlePrivate& eh = *handle.impl; ScopedLock l(eh.lock); assert(!eh.isIdle()); eh.events = PollerPrivate::directionToPollEvent(dir); //If fd is already associated, events and user arguments are updated //So, no need to check if fd is already associated QPID_POSIX_CHECK(::port_associate(impl->portId, PORT_SOURCE_FD, (uintptr_t) eh.fd, eh.events, &handle)); eh.setActive(); QPID_LOG(trace, "Poller::modFd(handle=" << &handle << ", fd=" << eh.fd << ")"); } void Poller::rearmFd(PollerHandle& handle) { PollerHandlePrivate& eh = *handle.impl; ScopedLock l(eh.lock); assert(eh.isInactive()); QPID_POSIX_CHECK(::port_associate(impl->portId, PORT_SOURCE_FD, (uintptr_t) eh.fd, eh.events, &handle)); eh.setActive(); QPID_LOG(trace, "Poller::rearmdFd(handle=" << &handle << ", fd=" << eh.fd << ")"); } void Poller::shutdown() { //Send an Alarm to the port //We need to send a nonzero event mask, using POLLHUP, but //The wait method will only look for a PORT_ALERT_SET QPID_POSIX_CHECK(::port_alert(impl->portId, PORT_ALERT_SET, POLLHUP, NULL)); QPID_LOG(trace, "Poller::shutdown"); } Poller::Event Poller::wait(Duration timeout) { timespec_t tout; timespec_t* ptout = NULL; port_event_t pe; if (timeout != TIME_INFINITE) { tout.tv_sec = 0; tout.tv_nsec = timeout; ptout = &tout; } do { PollerHandleDeletionManager.markAllUnusedInThisThread(); QPID_LOG(trace, "About to enter port_get. Thread " << pthread_self() << ", timeout=" << timeout); int rc = ::port_get(impl->portId, &pe, ptout); if (rc < 0) { switch (errno) { case EINTR: continue; case ETIME: return Event(0, TIMEOUT); default: QPID_POSIX_CHECK(rc); } } else { //We use alert mode to notify the shutdown of the Poller if (pe.portev_source == PORT_SOURCE_ALERT) { return Event(0, SHUTDOWN); } if (pe.portev_source == PORT_SOURCE_FD) { PollerHandle *handle = static_cast(pe.portev_user); PollerHandlePrivate& eh = *handle->impl; ScopedLock l(eh.lock); QPID_LOG(trace, "About to send handle: " << handle); if (eh.isActive()) { if (pe.portev_events & POLLHUP) { if (eh.isHungup()) { return Event(handle, DISCONNECTED); } eh.setHungup(); } else { eh.setInactive(); } QPID_LOG(trace, "Sending event (thread: " << pthread_self() << ") for handle " << handle << ", direction= " << PollerPrivate::pollToDirection(pe.portev_events)); return Event(handle, PollerPrivate::pollToDirection(pe.portev_events)); } } } } while (true); } // Concrete constructors Poller::Poller() : impl(new PollerPrivate()) {} Poller::~Poller() { delete impl; } }}