diff options
Diffstat (limited to 'cpp/src/qpid/sys/posix')
| -rw-r--r-- | cpp/src/qpid/sys/posix/AsynchIO.cpp | 122 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/posix/EventChannel.cpp | 596 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/posix/EventChannel.h | 214 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/posix/EventChannelAcceptor.cpp | 165 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/posix/EventChannelConnection.cpp | 237 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/posix/EventChannelConnection.h | 96 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/posix/EventChannelThreads.cpp | 126 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/posix/EventChannelThreads.h | 105 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/posix/PosixAcceptor.cpp | 49 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/posix/PrivatePosix.h | 5 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/posix/Socket.cpp | 106 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/posix/Socket.h | 76 |
12 files changed, 185 insertions, 1712 deletions
diff --git a/cpp/src/qpid/sys/posix/AsynchIO.cpp b/cpp/src/qpid/sys/posix/AsynchIO.cpp index 473ef7936f..2b462cbd7a 100644 --- a/cpp/src/qpid/sys/posix/AsynchIO.cpp +++ b/cpp/src/qpid/sys/posix/AsynchIO.cpp @@ -24,8 +24,6 @@ #include "check.h" #include <unistd.h> -#include <fcntl.h> -#include <sys/types.h> #include <sys/socket.h> #include <signal.h> #include <errno.h> @@ -37,13 +35,6 @@ using namespace qpid::sys; namespace { /* - * Make file descriptor non-blocking - */ -void nonblocking(int fd) { - QPID_POSIX_CHECK(::fcntl(fd, F_SETFL, O_NONBLOCK)); -} - -/* * Make *process* not generate SIGPIPE when writing to closed * pipe/socket (necessary as default action is to terminate process) */ @@ -57,11 +48,11 @@ void ignoreSigpipe() { * Asynch Acceptor */ -AsynchAcceptor::AsynchAcceptor(int fd, Callback callback) : +AsynchAcceptor::AsynchAcceptor(const Socket& s, Callback callback) : acceptedCallback(callback), - handle(fd, boost::bind(&AsynchAcceptor::readable, this, _1), 0, 0) { + handle(s, boost::bind(&AsynchAcceptor::readable, this, _1), 0, 0) { - nonblocking(fd); + s.setNonblocking(); ignoreSigpipe(); } @@ -73,18 +64,16 @@ void AsynchAcceptor::start(Poller::shared_ptr poller) { * We keep on accepting as long as there is something to accept */ void AsynchAcceptor::readable(DispatchHandle& h) { - int afd; + Socket* s; do { errno = 0; // TODO: Currently we ignore the peers address, perhaps we should // log it or use it for connection acceptance. - afd = ::accept(h.getFD(), 0, 0); - if (afd >= 0) { - acceptedCallback(afd); - } else if (errno == EAGAIN) { - break; + s = h.getSocket().accept(0, 0); + if (s) { + acceptedCallback(*s); } else { - QPID_POSIX_CHECK(afd); + break; } } while (true); @@ -94,21 +83,23 @@ void AsynchAcceptor::readable(DispatchHandle& h) { /* * Asynch reader/writer */ -AsynchIO::AsynchIO(int fd, +AsynchIO::AsynchIO(const Socket& s, ReadCallback rCb, EofCallback eofCb, DisconnectCallback disCb, - BuffersEmptyCallback eCb, IdleCallback iCb) : + ClosedCallback cCb, BuffersEmptyCallback eCb, IdleCallback iCb) : - DispatchHandle(fd, + DispatchHandle(s, boost::bind(&AsynchIO::readable, this, _1), boost::bind(&AsynchIO::writeable, this, _1), boost::bind(&AsynchIO::disconnected, this, _1)), readCallback(rCb), eofCallback(eofCb), disCallback(disCb), + closedCallback(cCb), emptyCallback(eCb), - idleCallback(iCb) { + idleCallback(iCb), + queuedClose(false) { - nonblocking(fd); + s.setNonblocking(); } struct deleter @@ -131,15 +122,58 @@ void AsynchIO::start(Poller::shared_ptr poller) { } void AsynchIO::queueReadBuffer(Buffer* buff) { + assert(buff); + buff->dataStart = 0; + buff->dataCount = 0; + bufferQueue.push_back(buff); + DispatchHandle::rewatchRead(); +} + +void AsynchIO::unread(Buffer* buff) { + assert(buff); + if (buff->dataStart != 0) { + memmove(buff->bytes, buff->bytes+buff->dataStart, buff->dataCount); + buff->dataStart = 0; + } bufferQueue.push_front(buff); DispatchHandle::rewatchRead(); } +// Either queue for writing or announce that there is something to write +// and we should ask for it void AsynchIO::queueWrite(Buffer* buff) { - writeQueue.push_front(buff); + // If no buffer then don't queue anything + // (but still wake up for writing) + if (buff) { + // If we've already closed the socket then throw the write away + if (queuedClose) { + bufferQueue.push_front(buff); + return; + } else { + writeQueue.push_front(buff); + } + } DispatchHandle::rewatchWrite(); } +void AsynchIO::queueWriteClose() { + queuedClose = true; +} + +/** Return a queued buffer if there are enough + * to spare + */ +AsynchIO::Buffer* AsynchIO::getQueuedBuffer() { + // Always keep at least one buffer (it might have data that was "unread" in it) + if (bufferQueue.size()<=1) + return 0; + Buffer* buff = bufferQueue.back(); + buff->dataStart = 0; + buff->dataCount = 0; + bufferQueue.pop_back(); + return buff; +} + /* * We keep on reading as long as we have something to read and a buffer to put * it in @@ -149,19 +183,19 @@ void AsynchIO::readable(DispatchHandle& h) { // (Try to) get a buffer if (!bufferQueue.empty()) { // Read into buffer - Buffer* buff = bufferQueue.back(); - bufferQueue.pop_back(); + Buffer* buff = bufferQueue.front(); + bufferQueue.pop_front(); errno = 0; - int rc = ::read(h.getFD(), buff->bytes, buff->byteCount); + int readCount = buff->byteCount-buff->dataCount; + int rc = h.getSocket().read(buff->bytes + buff->dataCount, readCount); if (rc == 0) { eofCallback(*this); h.unwatchRead(); return; } else if (rc > 0) { - buff->dataStart = 0; - buff->dataCount = rc; + buff->dataCount += rc; readCallback(*this, buff); - if (rc != buff->byteCount) { + if (rc != readCount) { // If we didn't fill the read buffer then time to stop reading return; } @@ -209,7 +243,7 @@ void AsynchIO::writeable(DispatchHandle& h) { writeQueue.pop_back(); errno = 0; assert(buff->dataStart+buff->dataCount <= buff->byteCount); - int rc = ::write(h.getFD(), buff->bytes+buff->dataStart, buff->dataCount); + int rc = h.getSocket().write(buff->bytes+buff->dataStart, buff->dataCount); if (rc >= 0) { // If we didn't write full buffer put rest back if (rc != buff->dataCount) { @@ -238,12 +272,17 @@ void AsynchIO::writeable(DispatchHandle& h) { } } } else { + // If we're waiting to close the socket then can do it now as there is nothing to write + if (queuedClose) { + close(h); + return; + } // Fd is writable, but nothing to write if (idleCallback) { idleCallback(*this); } // If we still have no buffers to write we can't do anything more - if (writeQueue.empty()) { + if (writeQueue.empty() && !queuedClose) { h.unwatchWrite(); return; } @@ -252,8 +291,25 @@ void AsynchIO::writeable(DispatchHandle& h) { } void AsynchIO::disconnected(DispatchHandle& h) { + // If we've already queued close do it before callback + if (queuedClose) { + close(h); + } + if (disCallback) { disCallback(*this); h.unwatch(); } } + +/* + * Close the socket and callback to say we've done it + */ +void AsynchIO::close(DispatchHandle& h) { + h.stopWatch(); + h.getSocket().close(); + if (closedCallback) { + closedCallback(*this, getSocket()); + } +} + diff --git a/cpp/src/qpid/sys/posix/EventChannel.cpp b/cpp/src/qpid/sys/posix/EventChannel.cpp deleted file mode 100644 index d35eedf5a5..0000000000 --- a/cpp/src/qpid/sys/posix/EventChannel.cpp +++ /dev/null @@ -1,596 +0,0 @@ -/* - * - * Copyright (c) 2006 The Apache Software Foundation - * - * Licensed 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. - * - */ - -// TODO aconway 2006-12-15: Locking review. - -// TODO aconway 2006-12-15: use Descriptor pointers everywhere, -// get them from channel, pass them to Event constructors. -// Eliminate lookup. - - -#include "EventChannel.h" -#include "check.h" - -#include "qpid/QpidError.h" -#include "qpid/sys/AtomicCount.h" - -#include <mqueue.h> -#include <string.h> -#include <iostream> - -#include <sys/errno.h> -#include <sys/socket.h> -#include <sys/epoll.h> - -#include <typeinfo> -#include <iostream> -#include <queue> - -#include <boost/ptr_container/ptr_map.hpp> -#include <boost/noncopyable.hpp> -#include <boost/bind.hpp> - -using namespace std; - - -namespace qpid { -namespace sys { - - -// ================================================================ -// Private class declarations - -namespace { - -typedef enum { IN, OUT } Direction; - -typedef std::pair<Event*, Event*> EventPair; - -/** - * Template to zero out a C-struct on construction. Avoids uninitialized memory - * warnings from valgrind or other mem checking tool. - */ -template <class T> struct CleanStruct : public T { - CleanStruct() { memset(this, 0, sizeof(*this)); } -}; - -} // namespace - -/** - * Queue of events corresponding to one IO direction (IN or OUT). - * Each Descriptor contains two Queues. - */ -class EventChannel::Queue : private boost::noncopyable -{ - public: - Queue(Descriptor& container, Direction dir); - - /** Called by Event classes in prepare() */ - void push(Event* e); - - /** Called when epoll wakes. - *@return The next completed event or 0. - */ - Event* wake(uint32_t epollFlags); - - Event* pop() { Event* e = queue.front(); queue.pop_front(); return e; } - - bool empty() { return queue.empty(); } - - void setBit(uint32_t &epollFlags); - - void shutdown(); - - private: - typedef std::deque<Event*> EventQ; - - inline bool isMyEvent(uint32_t flags) { return flags | myEvent; } - - Mutex& lock; // Shared with Descriptor. - Descriptor& descriptor; - uint32_t myEvent; // Epoll event flag. - EventQ queue; -}; - - -/** - * Manages a file descriptor in an epoll set. - * - * Can be shutdown and re-activated for the same file descriptor. - */ -class EventChannel::Descriptor : private boost::noncopyable { - public: - explicit Descriptor(int fd) : epollFd(-1), myFd(fd), - inQueue(*this, IN), outQueue(*this, OUT) {} - - void activate(int epollFd_); - - /** Epoll woke up for this descriptor. */ - Event* wake(uint32_t epollEvents); - - /** Shut down: close and remove file descriptor. - * May be re-activated if fd is reused. - */ - void shutdown(); - - // TODO aconway 2006-12-18: Nasty. Need to clean up interaction. - void shutdownUnsafe(); - - bool isShutdown() { return epollFd == -1; } - - Queue& getQueue(Direction d) { return d==IN ? inQueue : outQueue; } - int getFD() const { return myFd; } - - private: - void update(); - void epollCtl(int op, uint32_t events); - Queue* pick(); - - Mutex lock; - int epollFd; - int myFd; - Queue inQueue, outQueue; - bool preferIn; - - friend class Queue; -}; - - -/** - * Holds a map of Descriptors, which do most of the work. - */ -class EventChannel::Impl { - public: - Impl(int size = 256); - - ~Impl(); - - /** - * Activate descriptor - */ - void activate(Descriptor& d) { - d.activate(epollFd); - } - - /** Wait for an event, return 0 on timeout */ - Event* wait(Duration timeout); - - void shutdown(); - - private: - - Monitor monitor; - int epollFd; - int shutdownPipe[2]; - AtomicCount nWaiters; - bool isShutdown; -}; - - -// ================================================================ -// EventChannel::Queue::implementation. - -static const char* shutdownMsg = "Event queue shut down."; - -EventChannel::Queue::Queue(Descriptor& d, Direction dir) : - lock(d.lock), descriptor(d), - myEvent(dir==IN ? EPOLLIN : EPOLLOUT) -{} - -void EventChannel::Queue::push(Event* e) { - Mutex::ScopedLock l(lock); - if (descriptor.isShutdown()) - THROW_QPID_ERROR(INTERNAL_ERROR, shutdownMsg); - queue.push_back(e); - descriptor.update(); -} - -void EventChannel::Queue::setBit(uint32_t &epollFlags) { - if (queue.empty()) - epollFlags &= ~myEvent; - else - epollFlags |= myEvent; -} - -// TODO aconway 2006-12-20: REMOVE -Event* EventChannel::Queue::wake(uint32_t epollFlags) { - // Called with lock held. - if (!queue.empty() && (isMyEvent(epollFlags))) { - assert(!queue.empty()); - Event* e = queue.front(); - assert(e); - if (!e->getException()) { - // TODO aconway 2006-12-20: Can/should we move event completion - // out into dispatch() so it doesn't happen in Descriptor locks? - e->complete(descriptor); - } - queue.pop_front(); - return e; - } - return 0; -} - -void EventChannel::Queue::shutdown() { - // Mark all pending events with a shutdown exception. - // The server threads will remove and dispatch the events. - // - qpid::QpidError ex(INTERNAL_ERROR, shutdownMsg, SRCLINE); - for_each(queue.begin(), queue.end(), - boost::bind(&Event::setException, _1, ex)); -} - - -// ================================================================ -// Descriptor - - -void EventChannel::Descriptor::activate(int epollFd_) { - Mutex::ScopedLock l(lock); - if (isShutdown()) { - epollFd = epollFd_; // We're back in business. - epollCtl(EPOLL_CTL_ADD, 0); - } -} - -void EventChannel::Descriptor::shutdown() { - Mutex::ScopedLock l(lock); - shutdownUnsafe(); -} - -void EventChannel::Descriptor::shutdownUnsafe() { - // Caller holds lock. - ::close(myFd); - epollFd = -1; // Mark myself as shutdown. - inQueue.shutdown(); - outQueue.shutdown(); -} - -// TODO aconway 2006-12-20: Inline into wake(). -void EventChannel::Descriptor::update() { - // Caller holds lock. - if (isShutdown()) // Nothing to do - return; - uint32_t events = EPOLLONESHOT | EPOLLERR | EPOLLHUP; - inQueue.setBit(events); - outQueue.setBit(events); - epollCtl(EPOLL_CTL_MOD, events); -} - -void EventChannel::Descriptor::epollCtl(int op, uint32_t events) { - // Caller holds lock - assert(!isShutdown()); - CleanStruct<epoll_event> ee; - ee.data.ptr = this; - ee.events = events; - int status = ::epoll_ctl(epollFd, op, myFd, &ee); - if (status < 0) { - if (errno == EEXIST) // It's okay to add an existing fd - return; - else if (errno == EBADF) // FD was closed externally. - shutdownUnsafe(); - else - throw QPID_POSIX_ERROR(errno); - } -} - - -EventChannel::Queue* EventChannel::Descriptor::pick() { - if (inQueue.empty() && outQueue.empty()) - return 0; - if (inQueue.empty() || outQueue.empty()) - return !inQueue.empty() ? &inQueue : &outQueue; - // Neither is empty, pick fairly. - preferIn = !preferIn; - return preferIn ? &inQueue : &outQueue; -} - -Event* EventChannel::Descriptor::wake(uint32_t epollEvents) { - Mutex::ScopedLock l(lock); - // On error, shut down the Descriptor and both queues. - if (epollEvents & (EPOLLERR | EPOLLHUP)) { - shutdownUnsafe(); - // TODO aconway 2006-12-20: This error handling models means - // that any error reported by epoll will result in a shutdown - // exception on the events. Can we get more accurate error - // reporting somehow? - } - Queue*q = 0; - bool in = (epollEvents & EPOLLIN); - bool out = (epollEvents & EPOLLOUT); - if ((in && out) || isShutdown()) - q = pick(); // Choose fairly, either non-empty queue. - else if (in) - q = &inQueue; - else if (out) - q = &outQueue; - Event* e = (q && !q->empty()) ? q->pop() : 0; - update(); - if (e) - e->complete(*this); - return e; -} - - - -// ================================================================ -// EventChannel::Impl - - -EventChannel::Impl::Impl(int epollSize): - epollFd(-1), isShutdown(false) -{ - // Create the epoll file descriptor. - epollFd = epoll_create(epollSize); - QPID_POSIX_CHECK(epollFd); - - // Create a pipe and write a single byte. The byte is never - // read so the pipes read fd is always ready for read. - // We activate the FD when there are messages in the queue. - QPID_POSIX_CHECK(::pipe(shutdownPipe)); - static char zero = '\0'; - QPID_POSIX_CHECK(::write(shutdownPipe[1], &zero, 1)); -} - -EventChannel::Impl::~Impl() { - shutdown(); - ::close(epollFd); - ::close(shutdownPipe[0]); - ::close(shutdownPipe[1]); -} - - -void EventChannel::Impl::shutdown() { - Monitor::ScopedLock l(monitor); - if (!isShutdown) { // I'm starting shutdown. - isShutdown = true; - if (nWaiters == 0) - return; - - // TODO aconway 2006-12-20: If I just close the epollFd will - // that wake all threads? If so with what? Would be simpler than: - - CleanStruct<epoll_event> ee; - ee.data.ptr = 0; - ee.events = EPOLLIN; - QPID_POSIX_CHECK( - epoll_ctl(epollFd, EPOLL_CTL_ADD, shutdownPipe[0], &ee)); - } - // Wait for nWaiters to get out. - while (nWaiters > 0) { - monitor.wait(); - } -} - -// TODO aconway 2006-12-20: DEBUG remove -struct epoll { - epoll(uint32_t e) : events(e) { } - uint32_t events; -}; - -#define BIT(X) out << ((e.events & X) ? __STRING(X) "." : "") -ostream& operator << (ostream& out, epoll e) { - out << "epoll_event.events: "; - BIT(EPOLLIN); - BIT(EPOLLPRI); - BIT(EPOLLOUT); - BIT(EPOLLRDNORM); - BIT(EPOLLRDBAND); - BIT(EPOLLWRNORM); - BIT(EPOLLWRBAND); - BIT(EPOLLMSG); - BIT(EPOLLERR); - BIT(EPOLLHUP); - BIT(EPOLLONESHOT); - BIT(EPOLLET); - return out; -} - - - -/** - * Wait for epoll to wake up, return the descriptor or 0 on timeout. - */ -Event* EventChannel::Impl::wait(Duration timeoutNs) -{ - { - Monitor::ScopedLock l(monitor); - if (isShutdown) - throw ShutdownException(); - } - - // Increase nWaiters for the duration, notify the monitor if I'm - // the last one out. - // - AtomicCount::ScopedIncrement si( - nWaiters, boost::bind(&Monitor::notifyAll, &monitor)); - - // No lock, all thread safe calls or local variables: - // - const long timeoutMs = - (timeoutNs == TIME_INFINITE) ? -1 : timeoutNs/TIME_MSEC; - CleanStruct<epoll_event> ee; - Event* event = 0; - - // Loop till we get a completed event. Some events may repost - // themselves and return 0, e.g. incomplete read or write events. - //TODO aconway 2006-12-20: FIX THIS! - while (!event) { - int n = ::epoll_wait(epollFd, &ee, 1, timeoutMs); // Thread safe. - if (n == 0) // Timeout - return 0; - if (n < 0 && errno == EINTR) // Interrupt, ignore it. - continue; - if (n < 0) - throw QPID_POSIX_ERROR(errno); - assert(n == 1); - Descriptor* ed = - reinterpret_cast<Descriptor*>(ee.data.ptr); - if (ed == 0) // We're being shut-down. - throw ShutdownException(); - assert(ed != 0); - event = ed->wake(ee.events); - } - return event; -} - -//EventChannel::Descriptor& EventChannel::Impl::getDescriptor(int fd) { -// Mutex::ScopedLock l(monitor); -// Descriptor& ed = descriptors[fd]; -// ed.activate(epollFd, fd); -// return ed; -//} - - -// ================================================================ -// EventChannel - -EventChannel::shared_ptr EventChannel::create() { - return shared_ptr(new EventChannel()); -} - -EventChannel::EventChannel() : impl(new EventChannel::Impl()) {} - -EventChannel::~EventChannel() {} - -void EventChannel::post(Event& e) -{ - e.prepare(*impl); -} - -Event* EventChannel::wait(Duration timeoutNs) -{ - return impl->wait(timeoutNs); -} - -void EventChannel::shutdown() { - impl->shutdown(); -} - - -// ================================================================ -// Event and subclasses. - -Event::~Event() {} - -Exception::shared_ptr_const Event::getException() const { - return exception; -} - -void Event::throwIfException() { - if (getException()) - exception->throwSelf(); -} - -void Event::dispatch() -{ - if (!callback.empty()) - callback(); -} - -void Event::setException(const std::exception& e) { - const Exception* ex = dynamic_cast<const Exception*>(&e); - if (ex) - exception.reset(ex->clone().release()); - else - exception.reset(new Exception(e)); -#ifndef NDEBUG - // Throw and re-catch the exception. Has no effect on the - // program but it triggers debuggers watching for throw. The - // context that sets the exception is more informative for - // debugging purposes than the one that ultimately throws it. - // - try { - throwIfException(); - } - catch (...) { } // Ignored. -#endif -} - -int FDEvent::getFDescriptor() const { - return descriptor.getFD(); -} - -// TODO: AMS 21/12/06 Don't like the inline new, probably cause a memory leak -ReadEvent::ReadEvent(int fd, void* buf, size_t sz,Callback cb, bool noWait) : - IOEvent(cb, *(new EventChannel::Descriptor(fd)), sz, noWait), buffer(buf), bytesRead(0) { -} - -void ReadEvent::prepare(EventChannel::Impl& impl) { - EventChannel::Descriptor& d = getDescriptor(); - impl.activate(d); - d.getQueue(IN).push(this); -} - -void ReadEvent::complete(EventChannel::Descriptor& ed) -{ - ssize_t n = ::read(getFDescriptor(), - static_cast<char*>(buffer) + bytesRead, - size - bytesRead); - if (n > 0) - bytesRead += n; - if (n == 0 || (n < 0 && errno != EAGAIN)) { - // Use ENODATA for file closed. - setException(QPID_POSIX_ERROR(n == 0 ? ENODATA : errno)); - ed.shutdownUnsafe(); - } -} - -WriteEvent::WriteEvent(int fd, const void* buf, size_t sz, Callback cb) : - IOEvent(cb, *(new EventChannel::Descriptor(fd)), sz, noWait), buffer(buf), bytesWritten(0) { -} - -void WriteEvent::prepare(EventChannel::Impl& impl) { - EventChannel::Descriptor& d = getDescriptor(); - impl.activate(d); - d.getQueue(OUT).push(this); -} - - -void WriteEvent::complete(EventChannel::Descriptor& ed) -{ - ssize_t n = ::write(getFDescriptor(), - static_cast<const char*>(buffer) + bytesWritten, - size - bytesWritten); - if (n > 0) - bytesWritten += n; - if(n < 0 && errno != EAGAIN) { - setException(QPID_POSIX_ERROR(errno)); - ed.shutdownUnsafe(); // Called with lock held. - } -} - -AcceptEvent::AcceptEvent(int fd, Callback cb) : - FDEvent(cb, *(new EventChannel::Descriptor(fd))), accepted(0) { -} - -void AcceptEvent::prepare(EventChannel::Impl& impl) { - EventChannel::Descriptor& d = getDescriptor(); - impl.activate(d); - d.getQueue(IN).push(this); -} - -void AcceptEvent::complete(EventChannel::Descriptor& ed) -{ - accepted = ::accept(getFDescriptor(), 0, 0); - if (accepted < 0) { - setException(QPID_POSIX_ERROR(errno)); - ed.shutdownUnsafe(); // Called with lock held. - } -} - -}} diff --git a/cpp/src/qpid/sys/posix/EventChannel.h b/cpp/src/qpid/sys/posix/EventChannel.h deleted file mode 100644 index 85e121379a..0000000000 --- a/cpp/src/qpid/sys/posix/EventChannel.h +++ /dev/null @@ -1,214 +0,0 @@ -#ifndef _sys_EventChannel_h -#define _sys_EventChannel_h - -/* - * - * Copyright (c) 2006 The Apache Software Foundation - * - * Licensed 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/SharedObject.h" -#include "qpid/Exception.h" -#include "qpid/sys/Monitor.h" -#include "qpid/sys/Time.h" - -#include <boost/function.hpp> -#include <memory> - -namespace qpid { -namespace sys { - -class Event; - -/** - * Channel to post and wait for events. - */ -class EventChannel : public qpid::SharedObject<EventChannel> -{ - public: - static shared_ptr create(); - - /** Exception throw from wait() if channel is shut down. */ - class ShutdownException : public qpid::Exception {}; - - ~EventChannel(); - - /** Post an event to the channel. */ - void post(Event& event); - - /** - * Wait for the next complete event, up to timeout. - *@return Pointer to event or 0 if timeout elapses. - *@exception ShutdownException if the channel is shut down. - */ - Event* wait(Duration timeout = TIME_INFINITE); - - /** - * Shut down the event channel. - * Blocks till all threads have exited wait() - */ - void shutdown(); - - - // Internal classes. - class Impl; - class Queue; - class Descriptor; - - private: - - EventChannel(); - - Mutex lock; - boost::shared_ptr<Impl> impl; -}; - -/** - * Base class for all Events. - * - * Derived classes define events representing various async IO operations. - * When an event is complete, it is returned by the EventChannel to - * a thread calling wait. The thread will call Event::dispatch() to - * execute code associated with event completion. - */ -class Event -{ - public: - /** Type for callback when event is dispatched */ - typedef boost::function0<void> Callback; - - virtual ~Event(); - - /** Call the callback provided to the constructor, if any. */ - void dispatch(); - - /** - *If there was an exception processing this Event, return it. - *@return 0 if there was no exception. - */ - qpid::Exception::shared_ptr_const getException() const; - - /** If getException() throw the corresponding exception. */ - void throwIfException(); - - /** Set the dispatch callback. */ - void setCallback(Callback cb) { callback = cb; } - - /** Set the exception. */ - void setException(const std::exception& e); - - protected: - Event(Callback cb=0) : callback(cb) {} - - virtual void prepare(EventChannel::Impl&) = 0; - virtual void complete(EventChannel::Descriptor&) = 0; - - Callback callback; - Exception::shared_ptr_const exception; - - friend class EventChannel; - friend class EventChannel::Queue; -}; - -/** Base class for events related to a file descriptor */ -class FDEvent : public Event { - public: - EventChannel::Descriptor& getDescriptor() const { return descriptor; } - int getFDescriptor() const; - - protected: - FDEvent(Callback cb, EventChannel::Descriptor& fd) - : Event(cb), descriptor(fd) {} - // TODO AMS: 1/6/07 I really don't think this is correct, but - // the descriptor is immutable - FDEvent& operator=(const FDEvent& rhs) { Event::operator=(rhs); return *this; } - - private: - EventChannel::Descriptor& descriptor; -}; - -/** Base class for read or write events. */ -class IOEvent : public FDEvent { - public: - size_t getSize() const { return size; } - - protected: - IOEvent(Callback cb, EventChannel::Descriptor& fd, size_t sz, bool noWait_) : - FDEvent(cb, fd), size(sz), noWait(noWait_) {} - - size_t size; - bool noWait; -}; - -/** Asynchronous read event */ -class ReadEvent : public IOEvent -{ - public: - explicit ReadEvent(int fd, void* buf=0, size_t sz=0,Callback cb=0, bool noWait=false); - - void* getBuffer() const { return buffer; } - size_t getBytesRead() const { return bytesRead; } - - private: - void prepare(EventChannel::Impl&); - void complete(EventChannel::Descriptor&); - ssize_t doRead(); - - void* buffer; - size_t bytesRead; -}; - -/** Asynchronous write event */ -class WriteEvent : public IOEvent -{ - public: - explicit WriteEvent(int fd, const void* buf=0, size_t sz=0, Callback cb=0); - - const void* getBuffer() const { return buffer; } - size_t getBytesWritten() const { return bytesWritten; } - - private: - void prepare(EventChannel::Impl&); - void complete(EventChannel::Descriptor&); - ssize_t doWrite(); - - const void* buffer; - size_t bytesWritten; -}; - - -/** Asynchronous socket accept event */ -class AcceptEvent : public FDEvent -{ - public: - /** Accept a connection on fd. */ - explicit AcceptEvent(int fd, Callback cb=0); - - /** Get descriptor for accepted server socket */ - int getAcceptedDesscriptor() const { return accepted; } - - private: - void prepare(EventChannel::Impl&); - void complete(EventChannel::Descriptor&); - - int accepted; -}; - - -}} - - - -#endif /*!_sys_EventChannel_h*/ diff --git a/cpp/src/qpid/sys/posix/EventChannelAcceptor.cpp b/cpp/src/qpid/sys/posix/EventChannelAcceptor.cpp deleted file mode 100644 index a61a66c577..0000000000 --- a/cpp/src/qpid/sys/posix/EventChannelAcceptor.cpp +++ /dev/null @@ -1,165 +0,0 @@ -/* - * - * 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 "EventChannelConnection.h" - -#include "qpid/sys/ConnectionOutputHandler.h" -#include "qpid/sys/ConnectionInputHandler.h" -#include "qpid/sys/ConnectionInputHandlerFactory.h" -#include "qpid/sys/Acceptor.h" -#include "qpid/sys/Socket.h" -#include "qpid/framing/Buffer.h" -#include "qpid/framing/AMQFrame.h" -#include "qpid/Exception.h" - -#include <sys/socket.h> -#include <netdb.h> - -#include <boost/assert.hpp> -#include <boost/ptr_container/ptr_vector.hpp> -#include <boost/ptr_container/ptr_deque.hpp> -#include <boost/bind.hpp> -#include <boost/scoped_ptr.hpp> - -#include <iostream> - -namespace qpid { -namespace sys { - -using namespace qpid::framing; -using namespace std; - -class EventChannelAcceptor : public Acceptor { - public: - - - EventChannelAcceptor( - int16_t port_, int backlog, int nThreads, bool trace_ - ); - - uint16_t getPort() const; - std::string getHost() const; - - void run(ConnectionInputHandlerFactory* factory); - - void shutdown(); - - private: - - void accept(); - - Mutex lock; - Socket listener; - const int port; - const bool isTrace; - bool isRunning; - boost::ptr_vector<EventChannelConnection> connections; - AcceptEvent acceptEvent; - ConnectionInputHandlerFactory* factory; - bool isShutdown; - EventChannelThreads::shared_ptr threads; -}; - -Acceptor::shared_ptr Acceptor::create( - int16_t port, int backlog, int threads, bool trace) -{ - return Acceptor::shared_ptr( - new EventChannelAcceptor(port, backlog, threads, trace)); -} - -// Must define Acceptor virtual dtor. -Acceptor::~Acceptor() {} - -EventChannelAcceptor::EventChannelAcceptor( - int16_t port_, int backlog, int nThreads, bool trace_ -) : listener(Socket::createTcp()), - port(listener.listen(int(port_), backlog)), - isTrace(trace_), - isRunning(false), - acceptEvent(listener.fd(), - boost::bind(&EventChannelAcceptor::accept, this)), - factory(0), - isShutdown(false), - threads(EventChannelThreads::create(EventChannel::create(), nThreads)) -{ } - -uint16_t EventChannelAcceptor::getPort() const { - return port; // Immutable no need for lock. -} - -std::string EventChannelAcceptor::getHost() const { - ::sockaddr_storage name; // big enough for any socket address - ::socklen_t namelen = sizeof(name); - if (::getsockname(listener.fd(), (::sockaddr*)&name, &namelen) < 0) - throw QPID_POSIX_ERROR(errno); - - char dispName[NI_MAXHOST]; - if (int rc=::getnameinfo((::sockaddr*)&name, namelen, dispName, sizeof(dispName), 0, 0, NI_NUMERICHOST) != 0) - throw QPID_POSIX_ERROR(rc); - return dispName; -} - -void EventChannelAcceptor::run(ConnectionInputHandlerFactory* f) { - { - Mutex::ScopedLock l(lock); - if (!isRunning && !isShutdown) { - isRunning = true; - factory = f; - threads->post(acceptEvent); - } - } - threads->join(); // Wait for shutdown. -} - -void EventChannelAcceptor::shutdown() { - bool doShutdown = false; - { - Mutex::ScopedLock l(lock); - doShutdown = !isShutdown; // I'm the shutdown thread. - isShutdown = true; - } - if (doShutdown) { - ::close(acceptEvent.getFDescriptor()); - threads->shutdown(); - for_each(connections.begin(), connections.end(), - boost::bind(&EventChannelConnection::close, _1)); - } - threads->join(); -} - -void EventChannelAcceptor::accept() -{ - // No lock, we only post one accept event at a time. - if (isShutdown) - return; - if (acceptEvent.getException()) { - Exception::log(*acceptEvent.getException(), - "EventChannelAcceptor::accept"); - shutdown(); - return; - } - int fd = acceptEvent.getAcceptedDesscriptor(); - threads->post(acceptEvent); // Keep accepting. - // TODO aconway 2006-11-29: Need to reap closed connections also. - connections.push_back( - new EventChannelConnection(threads, *factory, fd, fd, isTrace)); -} - -}} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/posix/EventChannelConnection.cpp b/cpp/src/qpid/sys/posix/EventChannelConnection.cpp deleted file mode 100644 index f4b6396dd1..0000000000 --- a/cpp/src/qpid/sys/posix/EventChannelConnection.cpp +++ /dev/null @@ -1,237 +0,0 @@ -/* - * - * Copyright (c) 2006 The Apache Software Foundation - * - * Licensed 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 <iostream> - -#include <boost/bind.hpp> -#include <boost/assert.hpp> - -#include "EventChannelConnection.h" -#include "qpid/sys/ConnectionInputHandlerFactory.h" -#include "qpid/QpidError.h" - -using namespace std; -using namespace qpid; -using namespace qpid::framing; - -namespace qpid { -namespace sys { - -const size_t EventChannelConnection::bufferSize = 65536; - -EventChannelConnection::EventChannelConnection( - EventChannelThreads::shared_ptr threads_, - ConnectionInputHandlerFactory& factory_, - int rfd, - int wfd, - bool isTrace_ -) : - readFd(rfd), - writeFd(wfd ? wfd : rfd), - readEvent(readFd), - writeEvent(writeFd), - readCallback(boost::bind(&EventChannelConnection::closeOnException, - this, &EventChannelConnection::endInitRead)), - - isWriting(false), - isClosed(false), - threads(threads_), - handler(factory_.create(this)), - in(bufferSize), - out(bufferSize), - isTrace(isTrace_) -{ - assert(readFd > 0); - assert(writeFd > 0); - closeOnException(&EventChannelConnection::startRead); -} - - -void EventChannelConnection::send(AMQFrame& frame) { - { - Monitor::ScopedLock lock(monitor); - writeFrames.push_back(frame); - } - closeOnException(&EventChannelConnection::startWrite); -} - -void EventChannelConnection::close() { - { - Monitor::ScopedLock lock(monitor); - if (isClosed) - return; - isClosed = true; - } - ::close(readFd); - ::close(writeFd); - { - Monitor::ScopedLock lock(monitor); - while (busyThreads > 0) - monitor.wait(); - } - handler->closed(); -} - -void EventChannelConnection::closeNoThrow() { - Exception::tryCatchLog<void>( - boost::bind(&EventChannelConnection::close, this), - false, - "Exception closing channel" - ); -} - -/** - * Call f in a try/catch block and close the connection if - * an exception is thrown. - */ -void EventChannelConnection::closeOnException(MemberFnPtr f) -{ - try { - Exception::tryCatchLog<void>( - boost::bind(f, this), - "Closing connection due to exception" - ); - return; - } catch (...) { - // Exception was already logged by tryCatchLog - closeNoThrow(); - } -} - -// Post the write event. -// Always called inside closeOnException. -// Called by endWrite and send, but only one thread writes at a time. -// -void EventChannelConnection::startWrite() { - { - Monitor::ScopedLock lock(monitor); - // Stop if closed or a write event is already in progress. - if (isClosed || isWriting) - return; - if (writeFrames.empty()) { - isWriting = false; - return; - } - isWriting = true; - AMQFrame& frame = writeFrames.front(); - writeFrames.pop_front(); - // No need to lock here - only one thread can be writing at a time. - out.clear(); - if (isTrace) - cout << "Send on socket " << writeFd << ": " << frame << endl; - frame.encode(out); - out.flip(); - } - // TODO: AMS 1/6/07 This only works because we already have the correct fd - // in the descriptor - change not to use assigment - writeEvent = WriteEvent( - writeFd, out.start(), out.available(), - boost::bind(&EventChannelConnection::closeOnException, - this, &EventChannelConnection::endWrite)); - threads->post(writeEvent); -} - -// ScopedBusy ctor increments busyThreads. -// dtor decrements and calls monitor.notifyAll if it reaches 0. -// -struct EventChannelConnection::ScopedBusy : public AtomicCount::ScopedIncrement -{ - ScopedBusy(EventChannelConnection& ecc) - : AtomicCount::ScopedIncrement( - ecc.busyThreads, boost::bind(&Monitor::notifyAll, &ecc.monitor)) - {} -}; - -// Write event completed. -// Always called by a channel thread inside closeOnException. -// -void EventChannelConnection::endWrite() { - ScopedBusy(*this); - { - Monitor::ScopedLock lock(monitor); - assert(isWriting); - isWriting = false; - if (isClosed) - return; - writeEvent.throwIfException(); - if (writeEvent.getBytesWritten() < writeEvent.getSize()) { - // Keep writing the current event till done. - isWriting = true; - threads->post(writeEvent); - } - } - // Continue writing from writeFrames queue. - startWrite(); -} - - -// Post the read event. -// Always called inside closeOnException. -// Called from ctor and end[Init]Read, so only one call at a time -// is possible since we only post one read event at a time. -// -void EventChannelConnection::startRead() { - // Non blocking read, as much as we can swallow. - readEvent = ReadEvent( - readFd, in.start(), in.available(), readCallback); - threads->post(readEvent); -} - -// Completion of initial read, expect protocolInit. -// Always called inside closeOnException in channel thread. -// Only called by one thread at a time. -void EventChannelConnection::endInitRead() { - ScopedBusy(*this); - if (!isClosed) { - readEvent.throwIfException(); - in.move(readEvent.getBytesRead()); - in.flip(); - ProtocolInitiation protocolInit; - if(protocolInit.decode(in)){ - handler->initiated(protocolInit); - readCallback = boost::bind( - &EventChannelConnection::closeOnException, - this, &EventChannelConnection::endRead); - } - in.compact(); - // Continue reading. - startRead(); - } -} - -// Normal reads, expect a frame. -// Always called inside closeOnException in channel thread. -void EventChannelConnection::endRead() { - ScopedBusy(*this); - if (!isClosed) { - readEvent.throwIfException(); - in.move(readEvent.getBytesRead()); - in.flip(); - AMQFrame frame; - while (frame.decode(in)) { - if (isTrace) - cout << "Received on socket " << readFd - << ": " << frame << endl; - handler->received(frame); - } - in.compact(); - startRead(); - } -} - -}} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/posix/EventChannelConnection.h b/cpp/src/qpid/sys/posix/EventChannelConnection.h deleted file mode 100644 index a4ca5de517..0000000000 --- a/cpp/src/qpid/sys/posix/EventChannelConnection.h +++ /dev/null @@ -1,96 +0,0 @@ -#ifndef _posix_EventChannelConnection_h -#define _posix_EventChannelConnection_h - -/* - * - * Copyright (c) 2006 The Apache Software Foundation - * - * Licensed 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 <boost/ptr_container/ptr_deque.hpp> - -#include "EventChannelThreads.h" -#include "qpid/sys/Monitor.h" -#include "qpid/sys/ConnectionOutputHandler.h" -#include "qpid/sys/ConnectionInputHandler.h" -#include "qpid/sys/AtomicCount.h" -#include "qpid/framing/AMQFrame.h" - -namespace qpid { -namespace sys { - -class ConnectionInputHandlerFactory; - -/** - * Implements SessionContext and delegates to a SessionHandler - * for a connection via the EventChannel. - *@param readDescriptor file descriptor for reading. - *@param writeDescriptor file descriptor for writing, - * by default same as readDescriptor - */ -class EventChannelConnection : public ConnectionOutputHandler { - public: - EventChannelConnection( - EventChannelThreads::shared_ptr threads, - ConnectionInputHandlerFactory& factory, - int readDescriptor, - int writeDescriptor = 0, - bool isTrace = false - ); - - virtual void send(qpid::framing::AMQFrame& frame); - virtual void close(); - - private: - typedef std::deque<qpid::framing::AMQFrame> FrameQueue; - typedef void (EventChannelConnection::*MemberFnPtr)(); - struct ScopedBusy; - - void startWrite(); - void endWrite(); - void startRead(); - void endInitRead(); - void endRead(); - void closeNoThrow(); - void closeOnException(MemberFnPtr); - bool shouldContinue(bool& flag); - - static const size_t bufferSize; - - Monitor monitor; - - int readFd, writeFd; - ReadEvent readEvent; - WriteEvent writeEvent; - Event::Callback readCallback; - bool isWriting; - bool isClosed; - AtomicCount busyThreads; - - EventChannelThreads::shared_ptr threads; - std::auto_ptr<ConnectionInputHandler> handler; - qpid::framing::Buffer in, out; - FrameQueue writeFrames; - bool isTrace; - - friend struct ScopedBusy; -}; - - -}} // namespace qpid::sys - - - -#endif /*!_posix_EventChannelConnection_h*/ diff --git a/cpp/src/qpid/sys/posix/EventChannelThreads.cpp b/cpp/src/qpid/sys/posix/EventChannelThreads.cpp deleted file mode 100644 index 70954d0c16..0000000000 --- a/cpp/src/qpid/sys/posix/EventChannelThreads.cpp +++ /dev/null @@ -1,126 +0,0 @@ -/* - * - * Copyright (c) 2006 The Apache Software Foundation - * - * Licensed 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 <iostream> -#include <limits> - -#include <boost/bind.hpp> - -#include "qpid/sys/Runnable.h" - -#include "EventChannelThreads.h" - -namespace qpid { -namespace sys { - -const size_t EventChannelThreads::unlimited = - std::numeric_limits<size_t>::max(); - -EventChannelThreads::shared_ptr EventChannelThreads::create( - EventChannel::shared_ptr ec, size_t min, size_t max -) -{ - return EventChannelThreads::shared_ptr( - new EventChannelThreads(ec, min, max)); -} - -EventChannelThreads::EventChannelThreads( - EventChannel::shared_ptr ec, size_t min, size_t max) : - minThreads(std::max(size_t(1), min)), - maxThreads(std::min(min, max)), - channel(ec), - nWaiting(0), - state(RUNNING) -{ - Monitor::ScopedLock l(monitor); - while (workers.size() < minThreads) - workers.push_back(Thread(*this)); -} - -EventChannelThreads::~EventChannelThreads() { - shutdown(); - join(); -} - -void EventChannelThreads::shutdown() -{ - Monitor::ScopedLock lock(monitor); - if (state != RUNNING) // Already shutting down. - return; - state = TERMINATING; - channel->shutdown(); - monitor.notify(); // Wake up one join() thread. -} - -void EventChannelThreads::join() -{ - { - Monitor::ScopedLock lock(monitor); - while (state == RUNNING) // Wait for shutdown to start. - monitor.wait(); - if (state == SHUTDOWN) // Shutdown is complete - return; - if (state == JOINING) { - // Someone else is doing the join. - while (state != SHUTDOWN) - monitor.wait(); - return; - } - // I'm the joining thread - assert(state == TERMINATING); - state = JOINING; - } // Drop the lock. - - for (size_t i = 0; i < workers.size(); ++i) { - assert(state == JOINING); // Only this thread can change JOINING. - workers[i].join(); - } - state = SHUTDOWN; - monitor.notifyAll(); // Notify any other join() threads. -} - -void EventChannelThreads::addThread() { - Monitor::ScopedLock l(monitor); - if (workers.size() < maxThreads) - workers.push_back(Thread(*this)); -} - -void EventChannelThreads::run() -{ - // Start life waiting. Decrement on exit. - AtomicCount::ScopedIncrement inc(nWaiting); - try { - while (true) { - Event* e = channel->wait(); - assert(e != 0); - AtomicCount::ScopedDecrement dec(nWaiting); - // Make sure there's at least one waiting thread. - if (dec == 0 && state == RUNNING) - addThread(); - e->dispatch(); - } - } - catch (const EventChannel::ShutdownException& e) { - return; - } - catch (const std::exception& e) { - Exception::log(e, "Exception in EventChannelThreads::run()"); - } -} - -}} diff --git a/cpp/src/qpid/sys/posix/EventChannelThreads.h b/cpp/src/qpid/sys/posix/EventChannelThreads.h deleted file mode 100644 index 19112cf4db..0000000000 --- a/cpp/src/qpid/sys/posix/EventChannelThreads.h +++ /dev/null @@ -1,105 +0,0 @@ -#ifndef _posix_EventChannelThreads_h -#define _sys_EventChannelThreads_h - -/* - * - * Copyright (c) 2006 The Apache Software Foundation - * - * Licensed 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 "EventChannel.h" - -#include "qpid/Exception.h" -#include "qpid/sys/AtomicCount.h" -#include "qpid/sys/Monitor.h" -#include "qpid/sys/Thread.h" -#include "qpid/sys/Time.h" -#include "qpid/sys/Runnable.h" - -#include <vector> - -namespace qpid { -namespace sys { - -/** - Dynamic thread pool serving an EventChannel. - - Threads run a loop { e = wait(); e->dispatch(); } - The size of the thread pool is automatically adjusted to optimal size. -*/ -class EventChannelThreads : - public qpid::SharedObject<EventChannelThreads>, - private sys::Runnable -{ - public: - /** Constant to represent an unlimited number of threads */ - static const size_t unlimited; - - /** - * Create the thread pool and start initial threads. - * @param minThreads Pool will initialy contain minThreads threads and - * will never shrink to less until shutdown. - * @param maxThreads Pool will never grow to more than maxThreads. - */ - static EventChannelThreads::shared_ptr create( - EventChannel::shared_ptr channel = EventChannel::create(), - size_t minThreads = 1, - size_t maxThreads = unlimited - ); - - ~EventChannelThreads(); - - /** Post event to the underlying channel */ - void post(Event& event) { channel->post(event); } - - /** - * Terminate all threads. - * - * Returns immediately, use join() to wait till all threads are - * shut down. - */ - void shutdown(); - - /** Wait for all threads to terminate. */ - void join(); - - private: - typedef std::vector<sys::Thread> Threads; - typedef enum { - RUNNING, TERMINATING, JOINING, SHUTDOWN - } State; - - EventChannelThreads( - EventChannel::shared_ptr channel, size_t min, size_t max); - - void addThread(); - - void run(); - bool keepRunning(); - void adjustThreads(); - - Monitor monitor; - size_t minThreads; - size_t maxThreads; - EventChannel::shared_ptr channel; - Threads workers; - sys::AtomicCount nWaiting; - State state; -}; - - -}} - - -#endif /*!_sys_EventChannelThreads_h*/ diff --git a/cpp/src/qpid/sys/posix/PosixAcceptor.cpp b/cpp/src/qpid/sys/posix/PosixAcceptor.cpp deleted file mode 100644 index 0575380a14..0000000000 --- a/cpp/src/qpid/sys/posix/PosixAcceptor.cpp +++ /dev/null @@ -1,49 +0,0 @@ -/* - * - * 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/sys/Acceptor.h" -#include "qpid/Exception.h" - -namespace qpid { -namespace sys { - -namespace { -void fail() { throw qpid::Exception("PosixAcceptor not implemented"); } -} - -class PosixAcceptor : public Acceptor { - public: - virtual uint16_t getPort() const { fail(); return 0; } - virtual std::string getPort() const { fail(); return std::string(); } - virtual void run(qpid::sys::ConnectionInputHandlerFactory* ) { fail(); } - virtual void shutdown() { fail(); } -}; - -// Define generic Acceptor::create() to return APRAcceptor. - Acceptor::shared_ptr Acceptor::create(int16_t , int, int, bool) -{ - return Acceptor::shared_ptr(new PosixAcceptor()); -} - -// Must define Acceptor virtual dtor. -Acceptor::~Acceptor() {} - -}} diff --git a/cpp/src/qpid/sys/posix/PrivatePosix.h b/cpp/src/qpid/sys/posix/PrivatePosix.h index 2707057ef0..9ec9770cab 100644 --- a/cpp/src/qpid/sys/posix/PrivatePosix.h +++ b/cpp/src/qpid/sys/posix/PrivatePosix.h @@ -30,10 +30,15 @@ struct timeval; namespace qpid { namespace sys { +// Private Time related implementation details struct timespec& toTimespec(struct timespec& ts, const Duration& t); struct timeval& toTimeval(struct timeval& tv, const Duration& t); Duration toTime(const struct timespec& ts); +// Private socket related implementation details +class SocketPrivate; +int toFd(const SocketPrivate* s); + }} #endif /*!_sys_posix_PrivatePosix_h*/ diff --git a/cpp/src/qpid/sys/posix/Socket.cpp b/cpp/src/qpid/sys/posix/Socket.cpp index d46e7943d9..6ee7a84beb 100644 --- a/cpp/src/qpid/sys/posix/Socket.cpp +++ b/cpp/src/qpid/sys/posix/Socket.cpp @@ -25,6 +25,8 @@ #include "check.h" #include "PrivatePosix.h" +#include <fcntl.h> +#include <sys/types.h> #include <sys/socket.h> #include <sys/errno.h> #include <netinet/in.h> @@ -32,30 +34,63 @@ #include <boost/format.hpp> -using namespace qpid::sys; +namespace qpid { +namespace sys { -Socket Socket::createTcp() +class SocketPrivate { +public: + SocketPrivate(int f = -1) : + fd(f) + {} + + int fd; +}; + +Socket::Socket() : + impl(new SocketPrivate) +{ + createTcp(); +} + +Socket::Socket(SocketPrivate* sp) : + impl(sp) +{} + +Socket::~Socket() { + delete impl; +} + +void Socket::createTcp() const { + int& socket = impl->fd; + if (socket != -1) Socket::close(); int s = ::socket (PF_INET, SOCK_STREAM, 0); if (s < 0) throw QPID_POSIX_ERROR(errno); - return s; + socket = s; } -Socket::Socket(int descriptor) : socket(descriptor) {} - -void Socket::setTimeout(const Duration& interval) +void Socket::setTimeout(const Duration& interval) const { + const int& socket = impl->fd; struct timeval tv; toTimeval(tv, interval); setsockopt(socket, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); setsockopt(socket, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); } -void Socket::connect(const std::string& host, int port) +void Socket::setNonblocking() const { + QPID_POSIX_CHECK(::fcntl(impl->fd, F_SETFL, O_NONBLOCK)); +} + + +void Socket::connect(const std::string& host, int port) const { + const int& socket = impl->fd; struct sockaddr_in name; name.sin_family = AF_INET; name.sin_port = htons(port); + // TODO: Be good to make this work for IPv6 as well as IPv4 + // Use more modern lookup functions struct hostent* hp = gethostbyname ( host.c_str() ); if (hp == 0) throw QPID_POSIX_ERROR(errno); memcpy(&name.sin_addr.s_addr, hp->h_addr_list[0], hp->h_length); @@ -64,16 +99,18 @@ void Socket::connect(const std::string& host, int port) } void -Socket::close() +Socket::close() const { - if (socket == 0) return; + int& socket = impl->fd; + if (socket == -1) return; if (::close(socket) < 0) throw QPID_POSIX_ERROR(errno); - socket = 0; + socket = -1; } ssize_t -Socket::send(const void* data, size_t size) +Socket::send(const void* data, size_t size) const { + const int& socket = impl->fd; ssize_t sent = ::send(socket, data, size, 0); if (sent < 0) { if (errno == ECONNRESET) return SOCKET_EOF; @@ -84,8 +121,9 @@ Socket::send(const void* data, size_t size) } ssize_t -Socket::recv(void* data, size_t size) +Socket::recv(void* data, size_t size) const { + const int& socket = impl->fd; ssize_t received = ::recv(socket, data, size, 0); if (received < 0) { if (errno == ETIMEDOUT) return SOCKET_TIMEOUT; @@ -94,8 +132,9 @@ Socket::recv(void* data, size_t size) return received; } -int Socket::listen(int port, int backlog) +int Socket::listen(int port, int backlog) const { + const int& socket = impl->fd; struct sockaddr_in name; name.sin_family = AF_INET; name.sin_port = htons(port); @@ -111,8 +150,45 @@ int Socket::listen(int port, int backlog) return ntohs(name.sin_port); } + +Socket* Socket::accept(struct sockaddr *addr, socklen_t *addrlen) const +{ + int afd = ::accept(impl->fd, addr, addrlen); + if ( afd >= 0) + return new Socket(new SocketPrivate(afd)); + else if (errno == EAGAIN) + return 0; + else throw QPID_POSIX_ERROR(errno); +} + +int Socket::read(void *buf, size_t count) const +{ + return ::read(impl->fd, buf, count); +} + +int Socket::write(const void *buf, size_t count) const +{ + return ::write(impl->fd, buf, count); +} + +std::string Socket::getSockname() const +{ + ::sockaddr_storage name; // big enough for any socket address + ::socklen_t namelen = sizeof(name); + + const int& socket = impl->fd; + if (::getsockname(socket, (::sockaddr*)&name, &namelen) < 0) + throw QPID_POSIX_ERROR(errno); -int Socket::fd() const + char dispName[NI_MAXHOST]; + if (int rc=::getnameinfo((::sockaddr*)&name, namelen, dispName, sizeof(dispName), 0, 0, NI_NUMERICHOST) != 0) + throw QPID_POSIX_ERROR(rc); + return dispName; +} + +int toFd(const SocketPrivate* s) { - return socket; + return s->fd; } + +}} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/posix/Socket.h b/cpp/src/qpid/sys/posix/Socket.h deleted file mode 100644 index ca87104471..0000000000 --- a/cpp/src/qpid/sys/posix/Socket.h +++ /dev/null @@ -1,76 +0,0 @@ -#ifndef _sys_posix_Socket_h -#define _sys_posix_Socket_h - -/* - * - * 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 <string> -#include "qpid/sys/Time.h" - -namespace qpid { -namespace sys { - -class Socket -{ - public: - /** Create an initialized TCP socket */ - static Socket createTcp(); - - /** Create a socket wrapper for descriptor. */ - Socket(int descriptor = 0); - - /** Set timeout for read and write */ - void setTimeout(const Duration& interval); - - void connect(const std::string& host, int port); - - void close(); - - enum { SOCKET_TIMEOUT=-2, SOCKET_EOF=-3 } ErrorCode; - - /** Returns bytes sent or an ErrorCode value < 0. */ - ssize_t send(const void* data, size_t size); - - /** - * Returns bytes received, an ErrorCode value < 0 or 0 - * if the connection closed in an orderly manner. - */ - ssize_t recv(void* data, size_t size); - - /** Bind to a port and start listening. - *@param port 0 means choose an available port. - *@param backlog maximum number of pending connections. - *@return The bound port. - */ - int listen(int port = 0, int backlog = 10); - - /** Get file descriptor */ - int fd() const; - - private: - void init() const; - mutable int socket; // Initialized on demand. -}; - -}} - - -#endif /*!_sys_Socket_h*/ |