Import of qpid from etp:

  URL: https://etp.108.redhat.com/svn/etp/trunk/blaze
  Repository Root: https://etp.108.redhat.com/svn/etp
  Repository UUID: 06e15bec-b515-0410-bef0-cc27a458cf48
  Revision: 608



git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@447994 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 327 insertions, 0 deletions

diff --git a/cpp/common/concurrent/src/APRBase.cpp b/cpp/common/concurrent/src/APRBase.cpp
new file mode 100644
index 0000000000..f87ea9e25f
--- /dev/null
+++ b/cpp/common/concurrent/src/APRBase.cpp
@@ -0,0 +1,97 @@
+/*
+ *
+ * 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 "APRBase.h"
+#include "QpidError.h"
+
+using namespace qpid::concurrent;
+
+APRBase* APRBase::instance = 0;
+
+APRBase* APRBase::getInstance(){
+    if(instance == 0){
+	instance = new APRBase();
+    }
+    return instance;
+}
+
+
+APRBase::APRBase() : count(0){
+    apr_initialize();
+    CHECK_APR_SUCCESS(apr_pool_create(&pool, 0));
+    CHECK_APR_SUCCESS(apr_thread_mutex_create(&mutex, APR_THREAD_MUTEX_NESTED, pool));
+}
+
+APRBase::~APRBase(){
+    CHECK_APR_SUCCESS(apr_thread_mutex_destroy(mutex));
+    apr_pool_destroy(pool);
+    apr_terminate();  
+}
+
+bool APRBase::_increment(){
+    bool deleted(false);
+    CHECK_APR_SUCCESS(apr_thread_mutex_lock(mutex));
+    if(this == instance){
+	count++;
+    }else{
+	deleted = true;
+    }
+    CHECK_APR_SUCCESS(apr_thread_mutex_unlock(mutex));
+    return !deleted;
+}
+
+void APRBase::_decrement(){
+    APRBase* copy = 0;
+    CHECK_APR_SUCCESS(apr_thread_mutex_lock(mutex));
+    if(--count == 0){
+	copy = instance;
+	instance = 0;
+    }
+    CHECK_APR_SUCCESS(apr_thread_mutex_unlock(mutex));
+    if(copy != 0){
+	delete copy;
+    }
+}
+
+void APRBase::increment(){
+    int count = 0;
+    while(count++ < 2 && !getInstance()->_increment()){
+        std::cout << "WARNING: APR initialization triggered concurrently with termination." << std::endl;
+    }
+}
+
+void APRBase::decrement(){
+    getInstance()->_decrement();
+}
+
+void qpid::concurrent::check(apr_status_t status, const std::string& file, const int line){
+    if (status != APR_SUCCESS){
+        const int size = 50;
+        char tmp[size];
+        std::string msg(apr_strerror(status, tmp, size));
+        throw QpidError(APR_ERROR + ((int) status), msg, file, line);
+    }
+}
+
+std::string qpid::concurrent::get_desc(apr_status_t status){
+    const int size = 50;
+    char tmp[size];
+    std::string msg(apr_strerror(status, tmp, size));
+    return msg;
+}
+
diff --git a/cpp/common/concurrent/src/APRMonitor.cpp b/cpp/common/concurrent/src/APRMonitor.cpp
new file mode 100644
index 0000000000..428d76dff9
--- /dev/null
+++ b/cpp/common/concurrent/src/APRMonitor.cpp
@@ -0,0 +1,60 @@
+/*
+ *
+ * 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 "APRBase.h"
+#include "APRMonitor.h"
+#include <iostream>
+
+qpid::concurrent::APRMonitor::APRMonitor(){
+    APRBase::increment();
+    CHECK_APR_SUCCESS(apr_pool_create(&pool, NULL));
+    CHECK_APR_SUCCESS(apr_thread_mutex_create(&mutex, APR_THREAD_MUTEX_NESTED, pool));
+    CHECK_APR_SUCCESS(apr_thread_cond_create(&condition, pool));
+}
+
+qpid::concurrent::APRMonitor::~APRMonitor(){
+    CHECK_APR_SUCCESS(apr_thread_cond_destroy(condition));
+    CHECK_APR_SUCCESS(apr_thread_mutex_destroy(mutex));
+    apr_pool_destroy(pool);
+    APRBase::decrement();
+}
+
+void qpid::concurrent::APRMonitor::wait(){
+    CHECK_APR_SUCCESS(apr_thread_cond_wait(condition, mutex));
+}
+
+
+void qpid::concurrent::APRMonitor::wait(u_int64_t time){
+    apr_status_t status = apr_thread_cond_timedwait(condition, mutex, time * 1000);
+    if(!status == APR_TIMEUP) CHECK_APR_SUCCESS(status);
+}
+
+void qpid::concurrent::APRMonitor::notify(){
+    CHECK_APR_SUCCESS(apr_thread_cond_signal(condition));
+}
+
+void qpid::concurrent::APRMonitor::notifyAll(){
+    CHECK_APR_SUCCESS(apr_thread_cond_broadcast(condition));
+}
+
+void qpid::concurrent::APRMonitor::acquire(){
+    CHECK_APR_SUCCESS(apr_thread_mutex_lock(mutex));
+}
+
+void qpid::concurrent::APRMonitor::release(){
+    CHECK_APR_SUCCESS(apr_thread_mutex_unlock(mutex));
+}
diff --git a/cpp/common/concurrent/src/APRThread.cpp b/cpp/common/concurrent/src/APRThread.cpp
new file mode 100644
index 0000000000..4202fe81b6
--- /dev/null
+++ b/cpp/common/concurrent/src/APRThread.cpp
@@ -0,0 +1,50 @@
+/*
+ *
+ * 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 "APRBase.h"
+#include "APRThread.h"
+#include "apr_portable.h"
+
+using namespace qpid::concurrent;
+
+void* APR_THREAD_FUNC ExecRunnable(apr_thread_t* thread, void *data){
+    ((Runnable*) data)->run();
+    CHECK_APR_SUCCESS(apr_thread_exit(thread, APR_SUCCESS));
+    return NULL;
+} 
+
+APRThread::APRThread(apr_pool_t* _pool, Runnable* _runnable) : pool(_pool), runnable(_runnable){}
+
+APRThread::~APRThread(){
+}
+
+void APRThread::start(){
+    CHECK_APR_SUCCESS(apr_thread_create(&runner, NULL, ExecRunnable,(void*) runnable, pool));
+}
+
+void APRThread::join(){
+    apr_status_t status;
+    CHECK_APR_SUCCESS(apr_thread_join(&status, runner));
+}
+
+void APRThread::interrupt(){
+    CHECK_APR_SUCCESS(apr_thread_exit(runner, APR_SUCCESS));
+}
+
+unsigned int qpid::concurrent::APRThread::currentThread(){
+    return apr_os_thread_current();
+}
diff --git a/cpp/common/concurrent/src/APRThreadFactory.cpp b/cpp/common/concurrent/src/APRThreadFactory.cpp
new file mode 100644
index 0000000000..9ba68e9e56
--- /dev/null
+++ b/cpp/common/concurrent/src/APRThreadFactory.cpp
@@ -0,0 +1,35 @@
+/*
+ *
+ * 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 "APRBase.h"
+#include "APRThreadFactory.h"
+
+using namespace qpid::concurrent;
+
+APRThreadFactory::APRThreadFactory(){
+    APRBase::increment();
+    CHECK_APR_SUCCESS(apr_pool_create(&pool, NULL));
+}
+
+APRThreadFactory::~APRThreadFactory(){
+    apr_pool_destroy(pool);
+    APRBase::decrement();
+}
+
+Thread* APRThreadFactory::create(Runnable* runnable){
+    return new APRThread(pool, runnable);
+}
diff --git a/cpp/common/concurrent/src/APRThreadPool.cpp b/cpp/common/concurrent/src/APRThreadPool.cpp
new file mode 100644
index 0000000000..e0fcb804e6
--- /dev/null
+++ b/cpp/common/concurrent/src/APRThreadPool.cpp
@@ -0,0 +1,85 @@
+/*
+ *
+ * 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 "APRThreadFactory.h"
+#include "APRThreadPool.h"
+#include "QpidError.h"
+#include <iostream>
+
+using namespace qpid::concurrent;
+
+APRThreadPool::APRThreadPool(int _size) : size(_size), factory(new APRThreadFactory()), 
+                                          deleteFactory(true), running(false){
+    worker = new Worker(this);
+}
+
+APRThreadPool::APRThreadPool(int _size, ThreadFactory* _factory) : size(_size), factory(_factory), 
+                                                                   deleteFactory(false), running(false){
+    worker = new Worker(this);
+}
+
+APRThreadPool::~APRThreadPool(){
+    if(deleteFactory) delete factory;
+}
+
+void APRThreadPool::addTask(Runnable* task){
+    lock.acquire();
+    tasks.push(task);
+    lock.notifyAll();
+    lock.release();
+}
+
+void APRThreadPool::runTask(){
+    lock.acquire();
+    while(tasks.empty()){
+        lock.wait();
+    }
+    Runnable* task = tasks.front();
+    tasks.pop();
+    lock.release();
+    try{
+        task->run();
+    }catch(qpid::QpidError error){
+	std::cout << "Error [" << error.code << "] " << error.msg << " (" << error.file << ":" << error.line << ")" << std::endl;
+    }
+}
+
+void APRThreadPool::start(){
+    if(!running){
+        running = true;
+        for(int i = 0; i < size; i++){
+            Thread* t = factory->create(worker);
+            t->start();
+            threads.push_back(t);
+        }
+    }
+}
+
+void APRThreadPool::stop(){
+    if(!running){
+        running = false;
+        lock.acquire();
+        lock.notifyAll();
+        lock.release();
+        for(int i = 0; i < size; i++){
+            threads[i]->join();
+            delete threads[i];
+        }
+    }
+}
+
+