diff options
Diffstat (limited to 'cpp/lib/common/sys/posix/EventChannelThreads.cpp')
-rw-r--r-- | cpp/lib/common/sys/posix/EventChannelThreads.cpp | 119 |
1 files changed, 119 insertions, 0 deletions
diff --git a/cpp/lib/common/sys/posix/EventChannelThreads.cpp b/cpp/lib/common/sys/posix/EventChannelThreads.cpp new file mode 100644 index 0000000000..95e699e0b0 --- /dev/null +++ b/cpp/lib/common/sys/posix/EventChannelThreads.cpp @@ -0,0 +1,119 @@ +/* + * + * 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 "EventChannelThreads.h" +#include <sys/Runnable.h> +#include <iostream> +using namespace std; +#include <boost/bind.hpp> + +namespace qpid { +namespace sys { + +EventChannelThreads::shared_ptr EventChannelThreads::create( + EventChannel::shared_ptr ec) +{ + return EventChannelThreads::shared_ptr(new EventChannelThreads(ec)); +} + +EventChannelThreads::EventChannelThreads(EventChannel::shared_ptr ec) : + channel(ec), nWaiting(0), state(RUNNING) +{ + // TODO aconway 2006-11-15: Estimate initial threads based on CPUs. + addThread(); +} + +EventChannelThreads::~EventChannelThreads() { + shutdown(); + join(); +} + +void EventChannelThreads::shutdown() +{ + ScopedLock lock(*this); + if (state != RUNNING) // Already shutting down. + return; + for (size_t i = 0; i < workers.size(); ++i) { + channel->postEvent(terminate); + } + state = TERMINATE_SENT; + notify(); // Wake up one join() thread. +} + +void EventChannelThreads::join() +{ + { + ScopedLock lock(*this); + while (state == RUNNING) // Wait for shutdown to start. + wait(); + if (state == SHUTDOWN) // Shutdown is complete + return; + if (state == JOINING) { + // Someone else is doing the join. + while (state != SHUTDOWN) + wait(); + return; + } + // I'm the joining thread + assert(state == TERMINATE_SENT); + 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; + notifyAll(); // Notify other join() threaeds. +} + +void EventChannelThreads::addThread() { + ScopedLock l(*this); + workers.push_back(Thread(*this)); +} + +void EventChannelThreads::run() +{ + // Start life waiting. Decrement on exit. + AtomicCount::ScopedIncrement inc(nWaiting); + try { + while (true) { + Event* e = channel->getEvent(); + assert(e != 0); + if (e == &terminate) { + return; + } + AtomicCount::ScopedDecrement dec(nWaiting); + // I'm no longer waiting, make sure someone is. + if (dec == 0) + addThread(); + e->dispatch(); + } + } + catch (const std::exception& e) { + // TODO aconway 2006-11-15: need better logging across the board. + std::cerr << "EventChannelThreads::run() caught: " << e.what() + << std::endl; + } + catch (...) { + std::cerr << "EventChannelThreads::run() caught unknown exception." + << std::endl; + } +} + +}} |