diff options
author | Andrew Stitcher <astitcher@apache.org> | 2007-04-02 11:40:48 +0000 |
---|---|---|
committer | Andrew Stitcher <astitcher@apache.org> | 2007-04-02 11:40:48 +0000 |
commit | 9ecd69ebc88fb5d82a693e51eef0475c1a6b282e (patch) | |
tree | 841ab9ff2ebf92ad57bc9189eefc7448260577d1 /qpid/cpp/src/sys/ProducerConsumer.cpp | |
parent | 4ee7e8cbd677bd2ddf3f49d535a547e99c0aa150 (diff) | |
download | qpid-python-9ecd69ebc88fb5d82a693e51eef0475c1a6b282e.tar.gz |
Fix for the most disruptive items in QPID-243.
* All #include lines now use '""' rather than '<>' where appropriate.
* #include lines within the qpid project use relative includes so that
the same path will work in /usr/include when installed as part of the
client libraries.
* All the source code has now been rearranged to be under src in a directory
analogous to the namespace of the classes in it.
git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@524769 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'qpid/cpp/src/sys/ProducerConsumer.cpp')
-rw-r--r-- | qpid/cpp/src/sys/ProducerConsumer.cpp | 141 |
1 files changed, 141 insertions, 0 deletions
diff --git a/qpid/cpp/src/sys/ProducerConsumer.cpp b/qpid/cpp/src/sys/ProducerConsumer.cpp new file mode 100644 index 0000000000..1cf2a79784 --- /dev/null +++ b/qpid/cpp/src/sys/ProducerConsumer.cpp @@ -0,0 +1,141 @@ +/* + * + * 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 "../QpidError.h" +#include "ScopedIncrement.h" +#include "ProducerConsumer.h" + +namespace qpid { +namespace sys { + +// // ================ ProducerConsumer + +ProducerConsumer::ProducerConsumer(size_t init_items) + : items(init_items), waiters(0), shutdownFlag(false) +{} + +void ProducerConsumer::shutdown() { + Mutex::ScopedLock l(monitor); + shutdownFlag = true; + monitor.notifyAll(); + // Wait for waiting consumers to wake up. + while (waiters > 0) + monitor.wait(); +} + +size_t ProducerConsumer::available() const { + Mutex::ScopedLock l(monitor); + return items; +} + +size_t ProducerConsumer::consumers() const { + Mutex::ScopedLock l(monitor); + return waiters; +} + +// ================ Lock + +ProducerConsumer::Lock::Lock(ProducerConsumer& p) + : pc(p), lock(p.monitor), status(INCOMPLETE) {} + +bool ProducerConsumer::Lock::isOk() const { + return !pc.isShutdown() && status==INCOMPLETE; +} + +void ProducerConsumer::Lock::checkOk() const { + assert(!pc.isShutdown()); + assert(status == INCOMPLETE); +} + +ProducerConsumer::Lock::~Lock() { + assert(status != INCOMPLETE || pc.isShutdown()); +} + +void ProducerConsumer::Lock::confirm() { + checkOk(); + status = CONFIRMED; +} + +void ProducerConsumer::Lock::cancel() { + checkOk(); + status = CANCELLED; +} + +// ================ ProducerLock + +ProducerConsumer::ProducerLock::ProducerLock(ProducerConsumer& p) : Lock(p) +{} + + +ProducerConsumer::ProducerLock::~ProducerLock() { + if (status == CONFIRMED) { + pc.items++; + pc.monitor.notify(); // Notify a consumer. + } +} + +// ================ ConsumerLock + +ProducerConsumer::ConsumerLock::ConsumerLock(ProducerConsumer& p) : Lock(p) +{ + if (isOk()) { + ScopedIncrement<size_t> inc(pc.waiters); + while (pc.items == 0 && !pc.shutdownFlag) { + pc.monitor.wait(); + } + } +} + +ProducerConsumer::ConsumerLock::ConsumerLock( + ProducerConsumer& p, const Time& timeout) : Lock(p) +{ + if (isOk()) { + // Don't wait if timeout==0 + if (timeout == 0) { + if (pc.items == 0) + status = TIMEOUT; + return; + } + else { + Time deadline = now() + timeout; + ScopedIncrement<size_t> inc(pc.waiters); + while (pc.items == 0 && !pc.shutdownFlag) { + if (!pc.monitor.wait(deadline)) { + status = TIMEOUT; + return; + } + } + } + } +} + +ProducerConsumer::ConsumerLock::~ConsumerLock() { + if (pc.isShutdown()) { + if (pc.waiters == 0) + pc.monitor.notifyAll(); // Notify shutdown thread(s) + } + else if (status==CONFIRMED) { + pc.items--; + if (pc.items > 0) + pc.monitor.notify(); // Notify another consumer. + } +} + + +}} // namespace qpid::sys |