You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hawq.apache.org by es...@apache.org on 2016/09/08 05:20:42 UTC
[03/54] [partial] incubator-hawq git commit: HAWQ-959. revert thrift
build commands.
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/928eade3/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/Mutex.cpp
----------------------------------------------------------------------
diff --git a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/Mutex.cpp b/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/Mutex.cpp
deleted file mode 100644
index d9921aa..0000000
--- a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/Mutex.cpp
+++ /dev/null
@@ -1,374 +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 <thrift/thrift-config.h>
-
-#include <thrift/Thrift.h>
-#include <thrift/concurrency/Mutex.h>
-#include <thrift/concurrency/Util.h>
-
-#include <assert.h>
-#ifdef HAVE_PTHREAD_H
-#include <pthread.h>
-#endif
-#include <signal.h>
-
-using boost::shared_ptr;
-
-namespace apache {
-namespace thrift {
-namespace concurrency {
-
-#ifndef THRIFT_NO_CONTENTION_PROFILING
-
-static sig_atomic_t mutexProfilingSampleRate = 0;
-static MutexWaitCallback mutexProfilingCallback = 0;
-
-volatile static sig_atomic_t mutexProfilingCounter = 0;
-
-void enableMutexProfiling(int32_t profilingSampleRate, MutexWaitCallback callback) {
- mutexProfilingSampleRate = profilingSampleRate;
- mutexProfilingCallback = callback;
-}
-
-#define PROFILE_MUTEX_START_LOCK() int64_t _lock_startTime = maybeGetProfilingStartTime();
-
-#define PROFILE_MUTEX_NOT_LOCKED() \
- do { \
- if (_lock_startTime > 0) { \
- int64_t endTime = Util::currentTimeUsec(); \
- (*mutexProfilingCallback)(this, endTime - _lock_startTime); \
- } \
- } while (0)
-
-#define PROFILE_MUTEX_LOCKED() \
- do { \
- profileTime_ = _lock_startTime; \
- if (profileTime_ > 0) { \
- profileTime_ = Util::currentTimeUsec() - profileTime_; \
- } \
- } while (0)
-
-#define PROFILE_MUTEX_START_UNLOCK() \
- int64_t _temp_profileTime = profileTime_; \
- profileTime_ = 0;
-
-#define PROFILE_MUTEX_UNLOCKED() \
- do { \
- if (_temp_profileTime > 0) { \
- (*mutexProfilingCallback)(this, _temp_profileTime); \
- } \
- } while (0)
-
-static inline int64_t maybeGetProfilingStartTime() {
- if (mutexProfilingSampleRate && mutexProfilingCallback) {
- // This block is unsynchronized, but should produce a reasonable sampling
- // rate on most architectures. The main race conditions are the gap
- // between the decrement and the test, the non-atomicity of decrement, and
- // potential caching of different values at different CPUs.
- //
- // - if two decrements race, the likeliest result is that the counter
- // decrements slowly (perhaps much more slowly) than intended.
- //
- // - many threads could potentially decrement before resetting the counter
- // to its large value, causing each additional incoming thread to
- // profile every call. This situation is unlikely to persist for long
- // as the critical gap is quite short, but profiling could be bursty.
- sig_atomic_t localValue = --mutexProfilingCounter;
- if (localValue <= 0) {
- mutexProfilingCounter = mutexProfilingSampleRate;
- return Util::currentTimeUsec();
- }
- }
-
- return 0;
-}
-
-#else
-#define PROFILE_MUTEX_START_LOCK()
-#define PROFILE_MUTEX_NOT_LOCKED()
-#define PROFILE_MUTEX_LOCKED()
-#define PROFILE_MUTEX_START_UNLOCK()
-#define PROFILE_MUTEX_UNLOCKED()
-#endif // THRIFT_NO_CONTENTION_PROFILING
-
-/**
- * Implementation of Mutex class using POSIX mutex
- *
- * @version $Id:$
- */
-class Mutex::impl {
-public:
- impl(Initializer init) : initialized_(false) {
-#ifndef THRIFT_NO_CONTENTION_PROFILING
- profileTime_ = 0;
-#endif
- init(&pthread_mutex_);
- initialized_ = true;
- }
-
- ~impl() {
- if (initialized_) {
- initialized_ = false;
- int ret = pthread_mutex_destroy(&pthread_mutex_);
- THRIFT_UNUSED_VARIABLE(ret);
- assert(ret == 0);
- }
- }
-
- void lock() const {
- PROFILE_MUTEX_START_LOCK();
- pthread_mutex_lock(&pthread_mutex_);
- PROFILE_MUTEX_LOCKED();
- }
-
- bool trylock() const { return (0 == pthread_mutex_trylock(&pthread_mutex_)); }
-
- bool timedlock(int64_t milliseconds) const {
-#if defined(_POSIX_TIMEOUTS) && _POSIX_TIMEOUTS >= 200112L
- PROFILE_MUTEX_START_LOCK();
-
- struct THRIFT_TIMESPEC ts;
- Util::toTimespec(ts, milliseconds + Util::currentTime());
- int ret = pthread_mutex_timedlock(&pthread_mutex_, &ts);
- if (ret == 0) {
- PROFILE_MUTEX_LOCKED();
- return true;
- }
-
- PROFILE_MUTEX_NOT_LOCKED();
- return false;
-#else
- /* Otherwise follow solution used by Mono for Android */
- struct THRIFT_TIMESPEC sleepytime, now, to;
-
- /* This is just to avoid a completely busy wait */
- sleepytime.tv_sec = 0;
- sleepytime.tv_nsec = 10000000L; /* 10ms */
-
- Util::toTimespec(to, milliseconds + Util::currentTime());
-
- while ((trylock()) == false) {
- Util::toTimespec(now, Util::currentTime());
- if (now.tv_sec >= to.tv_sec && now.tv_nsec >= to.tv_nsec) {
- return false;
- }
- nanosleep(&sleepytime, NULL);
- }
-
- return true;
-#endif
- }
-
- void unlock() const {
- PROFILE_MUTEX_START_UNLOCK();
- pthread_mutex_unlock(&pthread_mutex_);
- PROFILE_MUTEX_UNLOCKED();
- }
-
- void* getUnderlyingImpl() const { return (void*)&pthread_mutex_; }
-
-private:
- mutable pthread_mutex_t pthread_mutex_;
- mutable bool initialized_;
-#ifndef THRIFT_NO_CONTENTION_PROFILING
- mutable int64_t profileTime_;
-#endif
-};
-
-Mutex::Mutex(Initializer init) : impl_(new Mutex::impl(init)) {
-}
-
-void* Mutex::getUnderlyingImpl() const {
- return impl_->getUnderlyingImpl();
-}
-
-void Mutex::lock() const {
- impl_->lock();
-}
-
-bool Mutex::trylock() const {
- return impl_->trylock();
-}
-
-bool Mutex::timedlock(int64_t ms) const {
- return impl_->timedlock(ms);
-}
-
-void Mutex::unlock() const {
- impl_->unlock();
-}
-
-void Mutex::DEFAULT_INITIALIZER(void* arg) {
- pthread_mutex_t* pthread_mutex = (pthread_mutex_t*)arg;
- int ret = pthread_mutex_init(pthread_mutex, NULL);
- THRIFT_UNUSED_VARIABLE(ret);
- assert(ret == 0);
-}
-
-#if defined(PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP) \
- || defined(PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP)
-static void init_with_kind(pthread_mutex_t* mutex, int kind) {
- pthread_mutexattr_t mutexattr;
- int ret = pthread_mutexattr_init(&mutexattr);
- assert(ret == 0);
-
- // Apparently, this can fail. Should we really be aborting?
- ret = pthread_mutexattr_settype(&mutexattr, kind);
- assert(ret == 0);
-
- ret = pthread_mutex_init(mutex, &mutexattr);
- assert(ret == 0);
-
- ret = pthread_mutexattr_destroy(&mutexattr);
- assert(ret == 0);
- THRIFT_UNUSED_VARIABLE(ret);
-}
-#endif
-
-#ifdef PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
-void Mutex::ADAPTIVE_INITIALIZER(void* arg) {
- // From mysql source: mysys/my_thr_init.c
- // Set mutex type to "fast" a.k.a "adaptive"
- //
- // In this case the thread may steal the mutex from some other thread
- // that is waiting for the same mutex. This will save us some
- // context switches but may cause a thread to 'starve forever' while
- // waiting for the mutex (not likely if the code within the mutex is
- // short).
- init_with_kind((pthread_mutex_t*)arg, PTHREAD_MUTEX_ADAPTIVE_NP);
-}
-#endif
-
-#ifdef PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP
-void Mutex::RECURSIVE_INITIALIZER(void* arg) {
- init_with_kind((pthread_mutex_t*)arg, PTHREAD_MUTEX_RECURSIVE_NP);
-}
-#endif
-
-/**
- * Implementation of ReadWriteMutex class using POSIX rw lock
- *
- * @version $Id:$
- */
-class ReadWriteMutex::impl {
-public:
- impl() : initialized_(false) {
-#ifndef THRIFT_NO_CONTENTION_PROFILING
- profileTime_ = 0;
-#endif
- int ret = pthread_rwlock_init(&rw_lock_, NULL);
- THRIFT_UNUSED_VARIABLE(ret);
- assert(ret == 0);
- initialized_ = true;
- }
-
- ~impl() {
- if (initialized_) {
- initialized_ = false;
- int ret = pthread_rwlock_destroy(&rw_lock_);
- THRIFT_UNUSED_VARIABLE(ret);
- assert(ret == 0);
- }
- }
-
- void acquireRead() const {
- PROFILE_MUTEX_START_LOCK();
- pthread_rwlock_rdlock(&rw_lock_);
- PROFILE_MUTEX_NOT_LOCKED(); // not exclusive, so use not-locked path
- }
-
- void acquireWrite() const {
- PROFILE_MUTEX_START_LOCK();
- pthread_rwlock_wrlock(&rw_lock_);
- PROFILE_MUTEX_LOCKED();
- }
-
- bool attemptRead() const { return !pthread_rwlock_tryrdlock(&rw_lock_); }
-
- bool attemptWrite() const { return !pthread_rwlock_trywrlock(&rw_lock_); }
-
- void release() const {
- PROFILE_MUTEX_START_UNLOCK();
- pthread_rwlock_unlock(&rw_lock_);
- PROFILE_MUTEX_UNLOCKED();
- }
-
-private:
- mutable pthread_rwlock_t rw_lock_;
- mutable bool initialized_;
-#ifndef THRIFT_NO_CONTENTION_PROFILING
- mutable int64_t profileTime_;
-#endif
-};
-
-ReadWriteMutex::ReadWriteMutex() : impl_(new ReadWriteMutex::impl()) {
-}
-
-void ReadWriteMutex::acquireRead() const {
- impl_->acquireRead();
-}
-
-void ReadWriteMutex::acquireWrite() const {
- impl_->acquireWrite();
-}
-
-bool ReadWriteMutex::attemptRead() const {
- return impl_->attemptRead();
-}
-
-bool ReadWriteMutex::attemptWrite() const {
- return impl_->attemptWrite();
-}
-
-void ReadWriteMutex::release() const {
- impl_->release();
-}
-
-NoStarveReadWriteMutex::NoStarveReadWriteMutex() : writerWaiting_(false) {
-}
-
-void NoStarveReadWriteMutex::acquireRead() const {
- if (writerWaiting_) {
- // writer is waiting, block on the writer's mutex until he's done with it
- mutex_.lock();
- mutex_.unlock();
- }
-
- ReadWriteMutex::acquireRead();
-}
-
-void NoStarveReadWriteMutex::acquireWrite() const {
- // if we can acquire the rwlock the easy way, we're done
- if (attemptWrite()) {
- return;
- }
-
- // failed to get the rwlock, do it the hard way:
- // locking the mutex and setting writerWaiting will cause all new readers to
- // block on the mutex rather than on the rwlock.
- mutex_.lock();
- writerWaiting_ = true;
- ReadWriteMutex::acquireWrite();
- writerWaiting_ = false;
- mutex_.unlock();
-}
-}
-}
-} // apache::thrift::concurrency
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/928eade3/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/Mutex.h
----------------------------------------------------------------------
diff --git a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/Mutex.h b/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/Mutex.h
deleted file mode 100644
index 6f892dc..0000000
--- a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/Mutex.h
+++ /dev/null
@@ -1,180 +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.
- */
-
-#ifndef _THRIFT_CONCURRENCY_MUTEX_H_
-#define _THRIFT_CONCURRENCY_MUTEX_H_ 1
-
-#include <boost/shared_ptr.hpp>
-#include <boost/noncopyable.hpp>
-#include <stdint.h>
-
-namespace apache {
-namespace thrift {
-namespace concurrency {
-
-#ifndef THRIFT_NO_CONTENTION_PROFILING
-
-/**
- * Determines if the Thrift Mutex and ReadWriteMutex classes will attempt to
- * profile their blocking acquire methods. If this value is set to non-zero,
- * Thrift will attempt to invoke the callback once every profilingSampleRate
- * times. However, as the sampling is not synchronized the rate is not
- * guranateed, and could be subject to big bursts and swings. Please ensure
- * your sampling callback is as performant as your application requires.
- *
- * The callback will get called with the wait time taken to lock the mutex in
- * usec and a (void*) that uniquely identifies the Mutex (or ReadWriteMutex)
- * being locked.
- *
- * The enableMutexProfiling() function is unsynchronized; calling this function
- * while profiling is already enabled may result in race conditions. On
- * architectures where a pointer assignment is atomic, this is safe but there
- * is no guarantee threads will agree on a single callback within any
- * particular time period.
- */
-typedef void (*MutexWaitCallback)(const void* id, int64_t waitTimeMicros);
-void enableMutexProfiling(int32_t profilingSampleRate, MutexWaitCallback callback);
-
-#endif
-
-/**
- * A simple mutex class
- *
- * @version $Id:$
- */
-class Mutex {
-public:
- typedef void (*Initializer)(void*);
-
- Mutex(Initializer init = DEFAULT_INITIALIZER);
- virtual ~Mutex() {}
- virtual void lock() const;
- virtual bool trylock() const;
- virtual bool timedlock(int64_t milliseconds) const;
- virtual void unlock() const;
-
- void* getUnderlyingImpl() const;
-
- static void DEFAULT_INITIALIZER(void*);
- static void ADAPTIVE_INITIALIZER(void*);
- static void RECURSIVE_INITIALIZER(void*);
-
-private:
- class impl;
- boost::shared_ptr<impl> impl_;
-};
-
-class ReadWriteMutex {
-public:
- ReadWriteMutex();
- virtual ~ReadWriteMutex() {}
-
- // these get the lock and block until it is done successfully
- virtual void acquireRead() const;
- virtual void acquireWrite() const;
-
- // these attempt to get the lock, returning false immediately if they fail
- virtual bool attemptRead() const;
- virtual bool attemptWrite() const;
-
- // this releases both read and write locks
- virtual void release() const;
-
-private:
- class impl;
- boost::shared_ptr<impl> impl_;
-};
-
-/**
- * A ReadWriteMutex that guarantees writers will not be starved by readers:
- * When a writer attempts to acquire the mutex, all new readers will be
- * blocked from acquiring the mutex until the writer has acquired and
- * released it. In some operating systems, this may already be guaranteed
- * by a regular ReadWriteMutex.
- */
-class NoStarveReadWriteMutex : public ReadWriteMutex {
-public:
- NoStarveReadWriteMutex();
-
- virtual void acquireRead() const;
- virtual void acquireWrite() const;
-
-private:
- Mutex mutex_;
- mutable volatile bool writerWaiting_;
-};
-
-class Guard : boost::noncopyable {
-public:
- Guard(const Mutex& value, int64_t timeout = 0) : mutex_(&value) {
- if (timeout == 0) {
- value.lock();
- } else if (timeout < 0) {
- if (!value.trylock()) {
- mutex_ = NULL;
- }
- } else {
- if (!value.timedlock(timeout)) {
- mutex_ = NULL;
- }
- }
- }
- ~Guard() {
- if (mutex_) {
- mutex_->unlock();
- }
- }
-
- operator bool() const { return (mutex_ != NULL); }
-
-private:
- const Mutex* mutex_;
-};
-
-// Can be used as second argument to RWGuard to make code more readable
-// as to whether we're doing acquireRead() or acquireWrite().
-enum RWGuardType { RW_READ = 0, RW_WRITE = 1 };
-
-class RWGuard : boost::noncopyable {
-public:
- RWGuard(const ReadWriteMutex& value, bool write = false) : rw_mutex_(value) {
- if (write) {
- rw_mutex_.acquireWrite();
- } else {
- rw_mutex_.acquireRead();
- }
- }
-
- RWGuard(const ReadWriteMutex& value, RWGuardType type) : rw_mutex_(value) {
- if (type == RW_WRITE) {
- rw_mutex_.acquireWrite();
- } else {
- rw_mutex_.acquireRead();
- }
- }
- ~RWGuard() { rw_mutex_.release(); }
-
-private:
- const ReadWriteMutex& rw_mutex_;
-};
-}
-}
-} // apache::thrift::concurrency
-
-#endif // #ifndef _THRIFT_CONCURRENCY_MUTEX_H_
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/928eade3/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/PlatformThreadFactory.h
----------------------------------------------------------------------
diff --git a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/PlatformThreadFactory.h b/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/PlatformThreadFactory.h
deleted file mode 100644
index 545b572..0000000
--- a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/PlatformThreadFactory.h
+++ /dev/null
@@ -1,52 +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.
- */
-
-#ifndef _THRIFT_CONCURRENCY_PLATFORMTHREADFACTORY_H_
-#define _THRIFT_CONCURRENCY_PLATFORMTHREADFACTORY_H_ 1
-
-// clang-format off
-#include <thrift/thrift-config.h>
-#if USE_BOOST_THREAD
-# include <thrift/concurrency/BoostThreadFactory.h>
-#elif USE_STD_THREAD
-# include <thrift/concurrency/StdThreadFactory.h>
-#else
-# include <thrift/concurrency/PosixThreadFactory.h>
-#endif
-// clang-format on
-
-namespace apache {
-namespace thrift {
-namespace concurrency {
-
-// clang-format off
-#if USE_BOOST_THREAD
- typedef BoostThreadFactory PlatformThreadFactory;
-#elif USE_STD_THREAD
- typedef StdThreadFactory PlatformThreadFactory;
-#else
- typedef PosixThreadFactory PlatformThreadFactory;
-#endif
-// clang-format on
-
-}
-}
-} // apache::thrift::concurrency
-
-#endif // #ifndef _THRIFT_CONCURRENCY_PLATFORMTHREADFACTORY_H_
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/928eade3/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/PosixThreadFactory.cpp
----------------------------------------------------------------------
diff --git a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/PosixThreadFactory.cpp b/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/PosixThreadFactory.cpp
deleted file mode 100644
index 47c5034..0000000
--- a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/PosixThreadFactory.cpp
+++ /dev/null
@@ -1,360 +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 <thrift/thrift-config.h>
-
-#include <thrift/concurrency/PosixThreadFactory.h>
-#include <thrift/concurrency/Exception.h>
-
-#if GOOGLE_PERFTOOLS_REGISTER_THREAD
-#include <google/profiler.h>
-#endif
-
-#include <assert.h>
-#include <pthread.h>
-
-#include <iostream>
-
-#include <boost/weak_ptr.hpp>
-
-namespace apache {
-namespace thrift {
-namespace concurrency {
-
-using boost::shared_ptr;
-using boost::weak_ptr;
-
-/**
- * The POSIX thread class.
- *
- * @version $Id:$
- */
-class PthreadThread : public Thread {
-public:
- enum STATE { uninitialized, starting, started, stopping, stopped };
-
- static const int MB = 1024 * 1024;
-
- static void* threadMain(void* arg);
-
-private:
- pthread_t pthread_;
- STATE state_;
- int policy_;
- int priority_;
- int stackSize_;
- weak_ptr<PthreadThread> self_;
- bool detached_;
-
-public:
- PthreadThread(int policy,
- int priority,
- int stackSize,
- bool detached,
- shared_ptr<Runnable> runnable)
- :
-
-#ifndef _WIN32
- pthread_(0),
-#endif // _WIN32
-
- state_(uninitialized),
- policy_(policy),
- priority_(priority),
- stackSize_(stackSize),
- detached_(detached) {
-
- this->Thread::runnable(runnable);
- }
-
- ~PthreadThread() {
- /* Nothing references this thread, if is is not detached, do a join
- now, otherwise the thread-id and, possibly, other resources will
- be leaked. */
- if (!detached_) {
- try {
- join();
- } catch (...) {
- // We're really hosed.
- }
- }
- }
-
- void start() {
- if (state_ != uninitialized) {
- return;
- }
-
- pthread_attr_t thread_attr;
- if (pthread_attr_init(&thread_attr) != 0) {
- throw SystemResourceException("pthread_attr_init failed");
- }
-
- if (pthread_attr_setdetachstate(&thread_attr,
- detached_ ? PTHREAD_CREATE_DETACHED : PTHREAD_CREATE_JOINABLE)
- != 0) {
- throw SystemResourceException("pthread_attr_setdetachstate failed");
- }
-
- // Set thread stack size
- if (pthread_attr_setstacksize(&thread_attr, MB * stackSize_) != 0) {
- throw SystemResourceException("pthread_attr_setstacksize failed");
- }
-
-// Set thread policy
-#ifdef _WIN32
- // WIN32 Pthread implementation doesn't seem to support sheduling policies other then
- // PosixThreadFactory::OTHER - runtime error
- policy_ = PosixThreadFactory::OTHER;
-#endif
-
- if (pthread_attr_setschedpolicy(&thread_attr, policy_) != 0) {
- throw SystemResourceException("pthread_attr_setschedpolicy failed");
- }
-
- struct sched_param sched_param;
- sched_param.sched_priority = priority_;
-
- // Set thread priority
- if (pthread_attr_setschedparam(&thread_attr, &sched_param) != 0) {
- throw SystemResourceException("pthread_attr_setschedparam failed");
- }
-
- // Create reference
- shared_ptr<PthreadThread>* selfRef = new shared_ptr<PthreadThread>();
- *selfRef = self_.lock();
-
- state_ = starting;
-
- if (pthread_create(&pthread_, &thread_attr, threadMain, (void*)selfRef) != 0) {
- throw SystemResourceException("pthread_create failed");
- }
- }
-
- void join() {
- if (!detached_ && state_ != uninitialized) {
- void* ignore;
- /* XXX
- If join fails it is most likely due to the fact
- that the last reference was the thread itself and cannot
- join. This results in leaked threads and will eventually
- cause the process to run out of thread resources.
- We're beyond the point of throwing an exception. Not clear how
- best to handle this. */
- int res = pthread_join(pthread_, &ignore);
- detached_ = (res == 0);
- if (res != 0) {
- GlobalOutput.printf("PthreadThread::join(): fail with code %d", res);
- }
- } else {
- GlobalOutput.printf("PthreadThread::join(): detached thread");
- }
- }
-
- Thread::id_t getId() {
-
-#ifndef _WIN32
- return (Thread::id_t)pthread_;
-#else
- return (Thread::id_t)pthread_.p;
-#endif // _WIN32
- }
-
- shared_ptr<Runnable> runnable() const { return Thread::runnable(); }
-
- void runnable(shared_ptr<Runnable> value) { Thread::runnable(value); }
-
- void weakRef(shared_ptr<PthreadThread> self) {
- assert(self.get() == this);
- self_ = weak_ptr<PthreadThread>(self);
- }
-};
-
-void* PthreadThread::threadMain(void* arg) {
- shared_ptr<PthreadThread> thread = *(shared_ptr<PthreadThread>*)arg;
- delete reinterpret_cast<shared_ptr<PthreadThread>*>(arg);
-
- if (thread == NULL) {
- return (void*)0;
- }
-
- if (thread->state_ != starting) {
- return (void*)0;
- }
-
-#if GOOGLE_PERFTOOLS_REGISTER_THREAD
- ProfilerRegisterThread();
-#endif
-
- thread->state_ = started;
- thread->runnable()->run();
- if (thread->state_ != stopping && thread->state_ != stopped) {
- thread->state_ = stopping;
- }
-
- return (void*)0;
-}
-
-/**
- * POSIX Thread factory implementation
- */
-class PosixThreadFactory::Impl {
-
-private:
- POLICY policy_;
- PRIORITY priority_;
- int stackSize_;
- bool detached_;
-
- /**
- * Converts generic posix thread schedule policy enums into pthread
- * API values.
- */
- static int toPthreadPolicy(POLICY policy) {
- switch (policy) {
- case OTHER:
- return SCHED_OTHER;
- case FIFO:
- return SCHED_FIFO;
- case ROUND_ROBIN:
- return SCHED_RR;
- }
- return SCHED_OTHER;
- }
-
- /**
- * Converts relative thread priorities to absolute value based on posix
- * thread scheduler policy
- *
- * The idea is simply to divide up the priority range for the given policy
- * into the correpsonding relative priority level (lowest..highest) and
- * then pro-rate accordingly.
- */
- static int toPthreadPriority(POLICY policy, PRIORITY priority) {
- int pthread_policy = toPthreadPolicy(policy);
- int min_priority = 0;
- int max_priority = 0;
-#ifdef HAVE_SCHED_GET_PRIORITY_MIN
- min_priority = sched_get_priority_min(pthread_policy);
-#endif
-#ifdef HAVE_SCHED_GET_PRIORITY_MAX
- max_priority = sched_get_priority_max(pthread_policy);
-#endif
- int quanta = (HIGHEST - LOWEST) + 1;
- float stepsperquanta = (float)(max_priority - min_priority) / quanta;
-
- if (priority <= HIGHEST) {
- return (int)(min_priority + stepsperquanta * priority);
- } else {
- // should never get here for priority increments.
- assert(false);
- return (int)(min_priority + stepsperquanta * NORMAL);
- }
- }
-
-public:
- Impl(POLICY policy, PRIORITY priority, int stackSize, bool detached)
- : policy_(policy), priority_(priority), stackSize_(stackSize), detached_(detached) {}
-
- /**
- * Creates a new POSIX thread to run the runnable object
- *
- * @param runnable A runnable object
- */
- shared_ptr<Thread> newThread(shared_ptr<Runnable> runnable) const {
- shared_ptr<PthreadThread> result
- = shared_ptr<PthreadThread>(new PthreadThread(toPthreadPolicy(policy_),
- toPthreadPriority(policy_, priority_),
- stackSize_,
- detached_,
- runnable));
- result->weakRef(result);
- runnable->thread(result);
- return result;
- }
-
- int getStackSize() const { return stackSize_; }
-
- void setStackSize(int value) { stackSize_ = value; }
-
- PRIORITY getPriority() const { return priority_; }
-
- /**
- * Sets priority.
- *
- * XXX
- * Need to handle incremental priorities properly.
- */
- void setPriority(PRIORITY value) { priority_ = value; }
-
- bool isDetached() const { return detached_; }
-
- void setDetached(bool value) { detached_ = value; }
-
- Thread::id_t getCurrentThreadId() const {
-
-#ifndef _WIN32
- return (Thread::id_t)pthread_self();
-#else
- return (Thread::id_t)pthread_self().p;
-#endif // _WIN32
- }
-};
-
-PosixThreadFactory::PosixThreadFactory(POLICY policy,
- PRIORITY priority,
- int stackSize,
- bool detached)
- : impl_(new PosixThreadFactory::Impl(policy, priority, stackSize, detached)) {
-}
-
-shared_ptr<Thread> PosixThreadFactory::newThread(shared_ptr<Runnable> runnable) const {
- return impl_->newThread(runnable);
-}
-
-int PosixThreadFactory::getStackSize() const {
- return impl_->getStackSize();
-}
-
-void PosixThreadFactory::setStackSize(int value) {
- impl_->setStackSize(value);
-}
-
-PosixThreadFactory::PRIORITY PosixThreadFactory::getPriority() const {
- return impl_->getPriority();
-}
-
-void PosixThreadFactory::setPriority(PosixThreadFactory::PRIORITY value) {
- impl_->setPriority(value);
-}
-
-bool PosixThreadFactory::isDetached() const {
- return impl_->isDetached();
-}
-
-void PosixThreadFactory::setDetached(bool value) {
- impl_->setDetached(value);
-}
-
-Thread::id_t PosixThreadFactory::getCurrentThreadId() const {
- return impl_->getCurrentThreadId();
-}
-}
-}
-} // apache::thrift::concurrency
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/928eade3/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/PosixThreadFactory.h
----------------------------------------------------------------------
diff --git a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/PosixThreadFactory.h b/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/PosixThreadFactory.h
deleted file mode 100644
index b26d296..0000000
--- a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/PosixThreadFactory.h
+++ /dev/null
@@ -1,131 +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.
- */
-
-#ifndef _THRIFT_CONCURRENCY_POSIXTHREADFACTORY_H_
-#define _THRIFT_CONCURRENCY_POSIXTHREADFACTORY_H_ 1
-
-#include <thrift/concurrency/Thread.h>
-
-#include <boost/shared_ptr.hpp>
-
-namespace apache {
-namespace thrift {
-namespace concurrency {
-
-/**
- * A thread factory to create posix threads
- *
- * @version $Id:$
- */
-class PosixThreadFactory : public ThreadFactory {
-
-public:
- /**
- * POSIX Thread scheduler policies
- */
- enum POLICY { OTHER, FIFO, ROUND_ROBIN };
-
- /**
- * POSIX Thread scheduler relative priorities,
- *
- * Absolute priority is determined by scheduler policy and OS. This
- * enumeration specifies relative priorities such that one can specify a
- * priority within a giving scheduler policy without knowing the absolute
- * value of the priority.
- */
- enum PRIORITY {
- LOWEST = 0,
- LOWER = 1,
- LOW = 2,
- NORMAL = 3,
- HIGH = 4,
- HIGHER = 5,
- HIGHEST = 6,
- INCREMENT = 7,
- DECREMENT = 8
- };
-
- /**
- * Posix thread (pthread) factory. All threads created by a factory are reference-counted
- * via boost::shared_ptr and boost::weak_ptr. The factory guarantees that threads and
- * the Runnable tasks they host will be properly cleaned up once the last strong reference
- * to both is given up.
- *
- * Threads are created with the specified policy, priority, stack-size and detachable-mode
- * detached means the thread is free-running and will release all system resources the
- * when it completes. A detachable thread is not joinable. The join method
- * of a detachable thread will return immediately with no error.
- *
- * By default threads are not joinable.
- */
-
- PosixThreadFactory(POLICY policy = ROUND_ROBIN,
- PRIORITY priority = NORMAL,
- int stackSize = 1,
- bool detached = true);
-
- // From ThreadFactory;
- boost::shared_ptr<Thread> newThread(boost::shared_ptr<Runnable> runnable) const;
-
- // From ThreadFactory;
- Thread::id_t getCurrentThreadId() const;
-
- /**
- * Gets stack size for created threads
- *
- * @return int size in megabytes
- */
- virtual int getStackSize() const;
-
- /**
- * Sets stack size for created threads
- *
- * @param value size in megabytes
- */
- virtual void setStackSize(int value);
-
- /**
- * Gets priority relative to current policy
- */
- virtual PRIORITY getPriority() const;
-
- /**
- * Sets priority relative to current policy
- */
- virtual void setPriority(PRIORITY priority);
-
- /**
- * Sets detached mode of threads
- */
- virtual void setDetached(bool detached);
-
- /**
- * Gets current detached mode
- */
- virtual bool isDetached() const;
-
-private:
- class Impl;
- boost::shared_ptr<Impl> impl_;
-};
-}
-}
-} // apache::thrift::concurrency
-
-#endif // #ifndef _THRIFT_CONCURRENCY_POSIXTHREADFACTORY_H_
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/928eade3/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdMonitor.cpp
----------------------------------------------------------------------
diff --git a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdMonitor.cpp b/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdMonitor.cpp
deleted file mode 100644
index 7b3b209..0000000
--- a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdMonitor.cpp
+++ /dev/null
@@ -1,213 +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 <thrift/thrift-config.h>
-
-#include <thrift/concurrency/Monitor.h>
-#include <thrift/concurrency/Exception.h>
-#include <thrift/concurrency/Util.h>
-#include <thrift/transport/PlatformSocket.h>
-#include <assert.h>
-
-#include <condition_variable>
-#include <chrono>
-#include <thread>
-#include <mutex>
-
-namespace apache {
-namespace thrift {
-namespace concurrency {
-
-/**
- * Monitor implementation using the std thread library
- *
- * @version $Id:$
- */
-class Monitor::Impl {
-
-public:
- Impl() : ownedMutex_(new Mutex()), conditionVariable_(), mutex_(NULL) { init(ownedMutex_.get()); }
-
- Impl(Mutex* mutex) : ownedMutex_(), conditionVariable_(), mutex_(NULL) { init(mutex); }
-
- Impl(Monitor* monitor) : ownedMutex_(), conditionVariable_(), mutex_(NULL) {
- init(&(monitor->mutex()));
- }
-
- Mutex& mutex() { return *mutex_; }
- void lock() { mutex_->lock(); }
- void unlock() { mutex_->unlock(); }
-
- /**
- * Exception-throwing version of waitForTimeRelative(), called simply
- * wait(int64) for historical reasons. Timeout is in milliseconds.
- *
- * If the condition occurs, this function returns cleanly; on timeout or
- * error an exception is thrown.
- */
- void wait(int64_t timeout_ms) {
- int result = waitForTimeRelative(timeout_ms);
- if (result == THRIFT_ETIMEDOUT) {
- throw TimedOutException();
- } else if (result != 0) {
- throw TException("Monitor::wait() failed");
- }
- }
-
- /**
- * Waits until the specified timeout in milliseconds for the condition to
- * occur, or waits forever if timeout_ms == 0.
- *
- * Returns 0 if condition occurs, THRIFT_ETIMEDOUT on timeout, or an error code.
- */
- int waitForTimeRelative(int64_t timeout_ms) {
- if (timeout_ms == 0LL) {
- return waitForever();
- }
-
- assert(mutex_);
- std::timed_mutex* mutexImpl = static_cast<std::timed_mutex*>(mutex_->getUnderlyingImpl());
- assert(mutexImpl);
-
- std::unique_lock<std::timed_mutex> lock(*mutexImpl, std::adopt_lock);
- bool timedout = (conditionVariable_.wait_for(lock, std::chrono::milliseconds(timeout_ms))
- == std::cv_status::timeout);
- lock.release();
- return (timedout ? THRIFT_ETIMEDOUT : 0);
- }
-
- /**
- * Waits until the absolute time specified using struct THRIFT_TIMESPEC.
- * Returns 0 if condition occurs, THRIFT_ETIMEDOUT on timeout, or an error code.
- */
- int waitForTime(const THRIFT_TIMESPEC* abstime) {
- struct timeval temp;
- temp.tv_sec = static_cast<long>(abstime->tv_sec);
- temp.tv_usec = static_cast<long>(abstime->tv_nsec) / 1000;
- return waitForTime(&temp);
- }
-
- /**
- * Waits until the absolute time specified using struct timeval.
- * Returns 0 if condition occurs, THRIFT_ETIMEDOUT on timeout, or an error code.
- */
- int waitForTime(const struct timeval* abstime) {
- assert(mutex_);
- std::timed_mutex* mutexImpl = static_cast<std::timed_mutex*>(mutex_->getUnderlyingImpl());
- assert(mutexImpl);
-
- struct timeval currenttime;
- Util::toTimeval(currenttime, Util::currentTime());
-
- long tv_sec = static_cast<long>(abstime->tv_sec - currenttime.tv_sec);
- long tv_usec = static_cast<long>(abstime->tv_usec - currenttime.tv_usec);
- if (tv_sec < 0)
- tv_sec = 0;
- if (tv_usec < 0)
- tv_usec = 0;
-
- std::unique_lock<std::timed_mutex> lock(*mutexImpl, std::adopt_lock);
- bool timedout = (conditionVariable_.wait_for(lock,
- std::chrono::seconds(tv_sec)
- + std::chrono::microseconds(tv_usec))
- == std::cv_status::timeout);
- lock.release();
- return (timedout ? THRIFT_ETIMEDOUT : 0);
- }
-
- /**
- * Waits forever until the condition occurs.
- * Returns 0 if condition occurs, or an error code otherwise.
- */
- int waitForever() {
- assert(mutex_);
- std::timed_mutex* mutexImpl = static_cast<std::timed_mutex*>(mutex_->getUnderlyingImpl());
- assert(mutexImpl);
-
- std::unique_lock<std::timed_mutex> lock(*mutexImpl, std::adopt_lock);
- conditionVariable_.wait(lock);
- lock.release();
- return 0;
- }
-
- void notify() { conditionVariable_.notify_one(); }
-
- void notifyAll() { conditionVariable_.notify_all(); }
-
-private:
- void init(Mutex* mutex) { mutex_ = mutex; }
-
- const std::unique_ptr<Mutex> ownedMutex_;
- std::condition_variable_any conditionVariable_;
- Mutex* mutex_;
-};
-
-Monitor::Monitor() : impl_(new Monitor::Impl()) {
-}
-Monitor::Monitor(Mutex* mutex) : impl_(new Monitor::Impl(mutex)) {
-}
-Monitor::Monitor(Monitor* monitor) : impl_(new Monitor::Impl(monitor)) {
-}
-
-Monitor::~Monitor() {
- delete impl_;
-}
-
-Mutex& Monitor::mutex() const {
- return const_cast<Monitor::Impl*>(impl_)->mutex();
-}
-
-void Monitor::lock() const {
- const_cast<Monitor::Impl*>(impl_)->lock();
-}
-
-void Monitor::unlock() const {
- const_cast<Monitor::Impl*>(impl_)->unlock();
-}
-
-void Monitor::wait(int64_t timeout) const {
- const_cast<Monitor::Impl*>(impl_)->wait(timeout);
-}
-
-int Monitor::waitForTime(const THRIFT_TIMESPEC* abstime) const {
- return const_cast<Monitor::Impl*>(impl_)->waitForTime(abstime);
-}
-
-int Monitor::waitForTime(const timeval* abstime) const {
- return const_cast<Monitor::Impl*>(impl_)->waitForTime(abstime);
-}
-
-int Monitor::waitForTimeRelative(int64_t timeout_ms) const {
- return const_cast<Monitor::Impl*>(impl_)->waitForTimeRelative(timeout_ms);
-}
-
-int Monitor::waitForever() const {
- return const_cast<Monitor::Impl*>(impl_)->waitForever();
-}
-
-void Monitor::notify() const {
- const_cast<Monitor::Impl*>(impl_)->notify();
-}
-
-void Monitor::notifyAll() const {
- const_cast<Monitor::Impl*>(impl_)->notifyAll();
-}
-}
-}
-} // apache::thrift::concurrency
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/928eade3/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdMutex.cpp
----------------------------------------------------------------------
diff --git a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdMutex.cpp b/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdMutex.cpp
deleted file mode 100644
index 69678a2..0000000
--- a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdMutex.cpp
+++ /dev/null
@@ -1,67 +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 <thrift/thrift-config.h>
-
-#include <thrift/concurrency/Mutex.h>
-#include <thrift/concurrency/Util.h>
-
-#include <cassert>
-#include <chrono>
-#include <mutex>
-
-namespace apache {
-namespace thrift {
-namespace concurrency {
-
-/**
- * Implementation of Mutex class using C++11 std::timed_mutex
- *
- * @version $Id:$
- */
-class Mutex::impl : public std::timed_mutex {};
-
-Mutex::Mutex(Initializer init) : impl_(new Mutex::impl()) {
-}
-
-void* Mutex::getUnderlyingImpl() const {
- return impl_.get();
-}
-
-void Mutex::lock() const {
- impl_->lock();
-}
-
-bool Mutex::trylock() const {
- return impl_->try_lock();
-}
-
-bool Mutex::timedlock(int64_t ms) const {
- return impl_->try_lock_for(std::chrono::milliseconds(ms));
-}
-
-void Mutex::unlock() const {
- impl_->unlock();
-}
-
-void Mutex::DEFAULT_INITIALIZER(void* arg) {
-}
-}
-}
-} // apache::thrift::concurrency
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/928eade3/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdThreadFactory.cpp
----------------------------------------------------------------------
diff --git a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdThreadFactory.cpp b/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdThreadFactory.cpp
deleted file mode 100644
index d57e7ec..0000000
--- a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdThreadFactory.cpp
+++ /dev/null
@@ -1,171 +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 <thrift/thrift-config.h>
-
-#if USE_STD_THREAD
-
-#include <thrift/concurrency/StdThreadFactory.h>
-#include <thrift/concurrency/Exception.h>
-
-#include <cassert>
-
-#include <boost/enable_shared_from_this.hpp>
-#include <boost/weak_ptr.hpp>
-#include <thread>
-
-namespace apache {
-namespace thrift {
-namespace concurrency {
-
-/**
- * The C++11 thread class.
- *
- * Note that we use boost shared_ptr rather than std shared_ptrs here
- * because the Thread/Runnable classes use those and we don't want to
- * mix them.
- *
- * @version $Id:$
- */
-class StdThread : public Thread, public boost::enable_shared_from_this<StdThread> {
-public:
- enum STATE { uninitialized, starting, started, stopping, stopped };
-
- static void threadMain(boost::shared_ptr<StdThread> thread);
-
-private:
- std::unique_ptr<std::thread> thread_;
- STATE state_;
- bool detached_;
-
-public:
- StdThread(bool detached, boost::shared_ptr<Runnable> runnable)
- : state_(uninitialized), detached_(detached) {
- this->Thread::runnable(runnable);
- }
-
- ~StdThread() {
- if (!detached_) {
- try {
- join();
- } catch (...) {
- // We're really hosed.
- }
- }
- }
-
- void start() {
- if (state_ != uninitialized) {
- return;
- }
-
- boost::shared_ptr<StdThread> selfRef = shared_from_this();
- state_ = starting;
-
- thread_ = std::unique_ptr<std::thread>(new std::thread(threadMain, selfRef));
-
- if (detached_)
- thread_->detach();
- }
-
- void join() {
- if (!detached_ && state_ != uninitialized) {
- thread_->join();
- }
- }
-
- Thread::id_t getId() { return thread_.get() ? thread_->get_id() : std::thread::id(); }
-
- boost::shared_ptr<Runnable> runnable() const { return Thread::runnable(); }
-
- void runnable(boost::shared_ptr<Runnable> value) { Thread::runnable(value); }
-};
-
-void StdThread::threadMain(boost::shared_ptr<StdThread> thread) {
- if (thread == NULL) {
- return;
- }
-
- if (thread->state_ != starting) {
- return;
- }
-
- thread->state_ = started;
- thread->runnable()->run();
-
- if (thread->state_ != stopping && thread->state_ != stopped) {
- thread->state_ = stopping;
- }
-
- return;
-}
-
-/**
- * std::thread factory implementation
- */
-class StdThreadFactory::Impl {
-
-private:
- bool detached_;
-
-public:
- Impl(bool detached) : detached_(detached) {}
-
- /**
- * Creates a new std::thread to run the runnable object
- *
- * @param runnable A runnable object
- */
- boost::shared_ptr<Thread> newThread(boost::shared_ptr<Runnable> runnable) const {
- boost::shared_ptr<StdThread> result
- = boost::shared_ptr<StdThread>(new StdThread(detached_, runnable));
- runnable->thread(result);
- return result;
- }
-
- bool isDetached() const { return detached_; }
-
- void setDetached(bool value) { detached_ = value; }
-
- Thread::id_t getCurrentThreadId() const { return std::this_thread::get_id(); }
-};
-
-StdThreadFactory::StdThreadFactory(bool detached) : impl_(new StdThreadFactory::Impl(detached)) {
-}
-
-boost::shared_ptr<Thread> StdThreadFactory::newThread(boost::shared_ptr<Runnable> runnable) const {
- return impl_->newThread(runnable);
-}
-
-bool StdThreadFactory::isDetached() const {
- return impl_->isDetached();
-}
-
-void StdThreadFactory::setDetached(bool value) {
- impl_->setDetached(value);
-}
-
-Thread::id_t StdThreadFactory::getCurrentThreadId() const {
- return impl_->getCurrentThreadId();
-}
-}
-}
-} // apache::thrift::concurrency
-
-#endif // USE_STD_THREAD
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/928eade3/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdThreadFactory.h
----------------------------------------------------------------------
diff --git a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdThreadFactory.h b/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdThreadFactory.h
deleted file mode 100644
index fb86bbf..0000000
--- a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/StdThreadFactory.h
+++ /dev/null
@@ -1,74 +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.
- */
-
-#ifndef _THRIFT_CONCURRENCY_STDTHREADFACTORY_H_
-#define _THRIFT_CONCURRENCY_STDTHREADFACTORY_H_ 1
-
-#include <thrift/concurrency/Thread.h>
-
-#include <boost/shared_ptr.hpp>
-
-namespace apache {
-namespace thrift {
-namespace concurrency {
-
-/**
- * A thread factory to create std::threads.
- *
- * @version $Id:$
- */
-class StdThreadFactory : public ThreadFactory {
-
-public:
- /**
- * Std thread factory. All threads created by a factory are reference-counted
- * via boost::shared_ptr and boost::weak_ptr. The factory guarantees that threads and
- * the Runnable tasks they host will be properly cleaned up once the last strong reference
- * to both is given up.
- *
- * By default threads are not joinable.
- */
-
- StdThreadFactory(bool detached = true);
-
- // From ThreadFactory;
- boost::shared_ptr<Thread> newThread(boost::shared_ptr<Runnable> runnable) const;
-
- // From ThreadFactory;
- Thread::id_t getCurrentThreadId() const;
-
- /**
- * Sets detached mode of threads
- */
- virtual void setDetached(bool detached);
-
- /**
- * Gets current detached mode
- */
- virtual bool isDetached() const;
-
-private:
- class Impl;
- boost::shared_ptr<Impl> impl_;
-};
-}
-}
-} // apache::thrift::concurrency
-
-#endif // #ifndef _THRIFT_CONCURRENCY_STDTHREADFACTORY_H_
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/928eade3/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/Thread.h
----------------------------------------------------------------------
diff --git a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/Thread.h b/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/Thread.h
deleted file mode 100644
index f5eb3a8..0000000
--- a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/Thread.h
+++ /dev/null
@@ -1,154 +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.
- */
-
-#ifndef _THRIFT_CONCURRENCY_THREAD_H_
-#define _THRIFT_CONCURRENCY_THREAD_H_ 1
-
-#include <stdint.h>
-#include <boost/shared_ptr.hpp>
-#include <boost/weak_ptr.hpp>
-
-#include <thrift/thrift-config.h>
-
-#if USE_BOOST_THREAD
-#include <boost/thread.hpp>
-#elif USE_STD_THREAD
-#include <thread>
-#else
-#ifdef HAVE_PTHREAD_H
-#include <pthread.h>
-#endif
-#endif
-
-namespace apache {
-namespace thrift {
-namespace concurrency {
-
-class Thread;
-
-/**
- * Minimal runnable class. More or less analogous to java.lang.Runnable.
- *
- * @version $Id:$
- */
-class Runnable {
-
-public:
- virtual ~Runnable(){};
- virtual void run() = 0;
-
- /**
- * Gets the thread object that is hosting this runnable object - can return
- * an empty boost::shared pointer if no references remain on that thread object
- */
- virtual boost::shared_ptr<Thread> thread() { return thread_.lock(); }
-
- /**
- * Sets the thread that is executing this object. This is only meant for
- * use by concrete implementations of Thread.
- */
- virtual void thread(boost::shared_ptr<Thread> value) { thread_ = value; }
-
-private:
- boost::weak_ptr<Thread> thread_;
-};
-
-/**
- * Minimal thread class. Returned by thread factory bound to a Runnable object
- * and ready to start execution. More or less analogous to java.lang.Thread
- * (minus all the thread group, priority, mode and other baggage, since that
- * is difficult to abstract across platforms and is left for platform-specific
- * ThreadFactory implemtations to deal with
- *
- * @see apache::thrift::concurrency::ThreadFactory)
- */
-class Thread {
-
-public:
-#if USE_BOOST_THREAD
- typedef boost::thread::id id_t;
-
- static inline bool is_current(id_t t) { return t == boost::this_thread::get_id(); }
- static inline id_t get_current() { return boost::this_thread::get_id(); }
-#elif USE_STD_THREAD
- typedef std::thread::id id_t;
-
- static inline bool is_current(id_t t) { return t == std::this_thread::get_id(); }
- static inline id_t get_current() { return std::this_thread::get_id(); }
-#else
- typedef pthread_t id_t;
-
- static inline bool is_current(id_t t) { return pthread_equal(pthread_self(), t); }
- static inline id_t get_current() { return pthread_self(); }
-#endif
-
- virtual ~Thread(){};
-
- /**
- * Starts the thread. Does platform specific thread creation and
- * configuration then invokes the run method of the Runnable object bound
- * to this thread.
- */
- virtual void start() = 0;
-
- /**
- * Join this thread. Current thread blocks until this target thread
- * completes.
- */
- virtual void join() = 0;
-
- /**
- * Gets the thread's platform-specific ID
- */
- virtual id_t getId() = 0;
-
- /**
- * Gets the runnable object this thread is hosting
- */
- virtual boost::shared_ptr<Runnable> runnable() const { return _runnable; }
-
-protected:
- virtual void runnable(boost::shared_ptr<Runnable> value) { _runnable = value; }
-
-private:
- boost::shared_ptr<Runnable> _runnable;
-};
-
-/**
- * Factory to create platform-specific thread object and bind them to Runnable
- * object for execution
- */
-class ThreadFactory {
-
-public:
- virtual ~ThreadFactory() {}
- virtual boost::shared_ptr<Thread> newThread(boost::shared_ptr<Runnable> runnable) const = 0;
-
- /** Gets the current thread id or unknown_thread_id if the current thread is not a thrift thread
- */
-
- static const Thread::id_t unknown_thread_id;
-
- virtual Thread::id_t getCurrentThreadId() const = 0;
-};
-}
-}
-} // apache::thrift::concurrency
-
-#endif // #ifndef _THRIFT_CONCURRENCY_THREAD_H_
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/928eade3/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/ThreadManager.cpp
----------------------------------------------------------------------
diff --git a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/ThreadManager.cpp b/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/ThreadManager.cpp
deleted file mode 100644
index a2b44d4..0000000
--- a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/ThreadManager.cpp
+++ /dev/null
@@ -1,561 +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 <thrift/thrift-config.h>
-
-#include <thrift/concurrency/ThreadManager.h>
-#include <thrift/concurrency/Exception.h>
-#include <thrift/concurrency/Monitor.h>
-#include <thrift/concurrency/Util.h>
-
-#include <boost/shared_ptr.hpp>
-
-#include <assert.h>
-#include <queue>
-#include <set>
-
-#if defined(DEBUG)
-#include <iostream>
-#endif // defined(DEBUG)
-
-namespace apache {
-namespace thrift {
-namespace concurrency {
-
-using boost::shared_ptr;
-using boost::dynamic_pointer_cast;
-
-/**
- * ThreadManager class
- *
- * This class manages a pool of threads. It uses a ThreadFactory to create
- * threads. It never actually creates or destroys worker threads, rather
- * it maintains statistics on number of idle threads, number of active threads,
- * task backlog, and average wait and service times.
- *
- * @version $Id:$
- */
-class ThreadManager::Impl : public ThreadManager {
-
-public:
- Impl()
- : workerCount_(0),
- workerMaxCount_(0),
- idleCount_(0),
- pendingTaskCountMax_(0),
- expiredCount_(0),
- state_(ThreadManager::UNINITIALIZED),
- monitor_(&mutex_),
- maxMonitor_(&mutex_) {}
-
- ~Impl() { stop(); }
-
- void start();
-
- void stop() { stopImpl(false); }
-
- void join() { stopImpl(true); }
-
- ThreadManager::STATE state() const { return state_; }
-
- shared_ptr<ThreadFactory> threadFactory() const {
- Synchronized s(monitor_);
- return threadFactory_;
- }
-
- void threadFactory(shared_ptr<ThreadFactory> value) {
- Synchronized s(monitor_);
- threadFactory_ = value;
- }
-
- void addWorker(size_t value);
-
- void removeWorker(size_t value);
-
- size_t idleWorkerCount() const { return idleCount_; }
-
- size_t workerCount() const {
- Synchronized s(monitor_);
- return workerCount_;
- }
-
- size_t pendingTaskCount() const {
- Synchronized s(monitor_);
- return tasks_.size();
- }
-
- size_t totalTaskCount() const {
- Synchronized s(monitor_);
- return tasks_.size() + workerCount_ - idleCount_;
- }
-
- size_t pendingTaskCountMax() const {
- Synchronized s(monitor_);
- return pendingTaskCountMax_;
- }
-
- size_t expiredTaskCount() {
- Synchronized s(monitor_);
- size_t result = expiredCount_;
- expiredCount_ = 0;
- return result;
- }
-
- void pendingTaskCountMax(const size_t value) {
- Synchronized s(monitor_);
- pendingTaskCountMax_ = value;
- }
-
- bool canSleep();
-
- void add(shared_ptr<Runnable> value, int64_t timeout, int64_t expiration);
-
- void remove(shared_ptr<Runnable> task);
-
- shared_ptr<Runnable> removeNextPending();
-
- void removeExpiredTasks();
-
- void setExpireCallback(ExpireCallback expireCallback);
-
-private:
- void stopImpl(bool join);
-
- size_t workerCount_;
- size_t workerMaxCount_;
- size_t idleCount_;
- size_t pendingTaskCountMax_;
- size_t expiredCount_;
- ExpireCallback expireCallback_;
-
- ThreadManager::STATE state_;
- shared_ptr<ThreadFactory> threadFactory_;
-
- friend class ThreadManager::Task;
- std::queue<shared_ptr<Task> > tasks_;
- Mutex mutex_;
- Monitor monitor_;
- Monitor maxMonitor_;
- Monitor workerMonitor_;
-
- friend class ThreadManager::Worker;
- std::set<shared_ptr<Thread> > workers_;
- std::set<shared_ptr<Thread> > deadWorkers_;
- std::map<const Thread::id_t, shared_ptr<Thread> > idMap_;
-};
-
-class ThreadManager::Task : public Runnable {
-
-public:
- enum STATE { WAITING, EXECUTING, CANCELLED, COMPLETE };
-
- Task(shared_ptr<Runnable> runnable, int64_t expiration = 0LL)
- : runnable_(runnable),
- state_(WAITING),
- expireTime_(expiration != 0LL ? Util::currentTime() + expiration : 0LL) {}
-
- ~Task() {}
-
- void run() {
- if (state_ == EXECUTING) {
- runnable_->run();
- state_ = COMPLETE;
- }
- }
-
- shared_ptr<Runnable> getRunnable() { return runnable_; }
-
- int64_t getExpireTime() const { return expireTime_; }
-
-private:
- shared_ptr<Runnable> runnable_;
- friend class ThreadManager::Worker;
- STATE state_;
- int64_t expireTime_;
-};
-
-class ThreadManager::Worker : public Runnable {
- enum STATE { UNINITIALIZED, STARTING, STARTED, STOPPING, STOPPED };
-
-public:
- Worker(ThreadManager::Impl* manager) : manager_(manager), state_(UNINITIALIZED), idle_(false) {}
-
- ~Worker() {}
-
-private:
- bool isActive() const {
- return (manager_->workerCount_ <= manager_->workerMaxCount_)
- || (manager_->state_ == JOINING && !manager_->tasks_.empty());
- }
-
-public:
- /**
- * Worker entry point
- *
- * As long as worker thread is running, pull tasks off the task queue and
- * execute.
- */
- void run() {
- bool active = false;
- bool notifyManager = false;
-
- /**
- * Increment worker semaphore and notify manager if worker count reached
- * desired max
- *
- * Note: We have to release the monitor and acquire the workerMonitor
- * since that is what the manager blocks on for worker add/remove
- */
- {
- Synchronized s(manager_->monitor_);
- active = manager_->workerCount_ < manager_->workerMaxCount_;
- if (active) {
- manager_->workerCount_++;
- notifyManager = manager_->workerCount_ == manager_->workerMaxCount_;
- }
- }
-
- if (notifyManager) {
- Synchronized s(manager_->workerMonitor_);
- manager_->workerMonitor_.notify();
- notifyManager = false;
- }
-
- while (active) {
- shared_ptr<ThreadManager::Task> task;
-
- /**
- * While holding manager monitor block for non-empty task queue (Also
- * check that the thread hasn't been requested to stop). Once the queue
- * is non-empty, dequeue a task, release monitor, and execute. If the
- * worker max count has been decremented such that we exceed it, mark
- * ourself inactive, decrement the worker count and notify the manager
- * (technically we're notifying the next blocked thread but eventually
- * the manager will see it.
- */
- {
- Guard g(manager_->mutex_);
- active = isActive();
-
- while (active && manager_->tasks_.empty()) {
- manager_->idleCount_++;
- idle_ = true;
- manager_->monitor_.wait();
- active = isActive();
- idle_ = false;
- manager_->idleCount_--;
- }
-
- if (active) {
- manager_->removeExpiredTasks();
-
- if (!manager_->tasks_.empty()) {
- task = manager_->tasks_.front();
- manager_->tasks_.pop();
- if (task->state_ == ThreadManager::Task::WAITING) {
- task->state_ = ThreadManager::Task::EXECUTING;
- }
- }
-
- /* If we have a pending task max and we just dropped below it, wakeup any
- thread that might be blocked on add. */
- if (manager_->pendingTaskCountMax_ != 0
- && manager_->tasks_.size() <= manager_->pendingTaskCountMax_ - 1) {
- manager_->maxMonitor_.notify();
- }
- } else {
- idle_ = true;
- manager_->workerCount_--;
- notifyManager = (manager_->workerCount_ == manager_->workerMaxCount_);
- }
- }
-
- if (task) {
- if (task->state_ == ThreadManager::Task::EXECUTING) {
- try {
- task->run();
- } catch (const std::exception& e) {
- GlobalOutput.printf("[ERROR] task->run() raised an exception: %s", e.what());
- } catch (...) {
- GlobalOutput.printf("[ERROR] task->run() raised an unknown exception");
- }
- }
- }
- }
-
- {
- Synchronized s(manager_->workerMonitor_);
- manager_->deadWorkers_.insert(this->thread());
- if (notifyManager) {
- manager_->workerMonitor_.notify();
- }
- }
-
- return;
- }
-
-private:
- ThreadManager::Impl* manager_;
- friend class ThreadManager::Impl;
- STATE state_;
- bool idle_;
-};
-
-void ThreadManager::Impl::addWorker(size_t value) {
- std::set<shared_ptr<Thread> > newThreads;
- for (size_t ix = 0; ix < value; ix++) {
- shared_ptr<ThreadManager::Worker> worker
- = shared_ptr<ThreadManager::Worker>(new ThreadManager::Worker(this));
- newThreads.insert(threadFactory_->newThread(worker));
- }
-
- {
- Synchronized s(monitor_);
- workerMaxCount_ += value;
- workers_.insert(newThreads.begin(), newThreads.end());
- }
-
- for (std::set<shared_ptr<Thread> >::iterator ix = newThreads.begin(); ix != newThreads.end();
- ++ix) {
- shared_ptr<ThreadManager::Worker> worker
- = dynamic_pointer_cast<ThreadManager::Worker, Runnable>((*ix)->runnable());
- worker->state_ = ThreadManager::Worker::STARTING;
- (*ix)->start();
- idMap_.insert(std::pair<const Thread::id_t, shared_ptr<Thread> >((*ix)->getId(), *ix));
- }
-
- {
- Synchronized s(workerMonitor_);
- while (workerCount_ != workerMaxCount_) {
- workerMonitor_.wait();
- }
- }
-}
-
-void ThreadManager::Impl::start() {
-
- if (state_ == ThreadManager::STOPPED) {
- return;
- }
-
- {
- Synchronized s(monitor_);
- if (state_ == ThreadManager::UNINITIALIZED) {
- if (!threadFactory_) {
- throw InvalidArgumentException();
- }
- state_ = ThreadManager::STARTED;
- monitor_.notifyAll();
- }
-
- while (state_ == STARTING) {
- monitor_.wait();
- }
- }
-}
-
-void ThreadManager::Impl::stopImpl(bool join) {
- bool doStop = false;
- if (state_ == ThreadManager::STOPPED) {
- return;
- }
-
- {
- Synchronized s(monitor_);
- if (state_ != ThreadManager::STOPPING && state_ != ThreadManager::JOINING
- && state_ != ThreadManager::STOPPED) {
- doStop = true;
- state_ = join ? ThreadManager::JOINING : ThreadManager::STOPPING;
- }
- }
-
- if (doStop) {
- removeWorker(workerCount_);
- }
-
- // XXX
- // should be able to block here for transition to STOPPED since we're no
- // using shared_ptrs
-
- {
- Synchronized s(monitor_);
- state_ = ThreadManager::STOPPED;
- }
-}
-
-void ThreadManager::Impl::removeWorker(size_t value) {
- std::set<shared_ptr<Thread> > removedThreads;
- {
- Synchronized s(monitor_);
- if (value > workerMaxCount_) {
- throw InvalidArgumentException();
- }
-
- workerMaxCount_ -= value;
-
- if (idleCount_ < value) {
- for (size_t ix = 0; ix < idleCount_; ix++) {
- monitor_.notify();
- }
- } else {
- monitor_.notifyAll();
- }
- }
-
- {
- Synchronized s(workerMonitor_);
-
- while (workerCount_ != workerMaxCount_) {
- workerMonitor_.wait();
- }
-
- for (std::set<shared_ptr<Thread> >::iterator ix = deadWorkers_.begin();
- ix != deadWorkers_.end();
- ++ix) {
- idMap_.erase((*ix)->getId());
- workers_.erase(*ix);
- }
-
- deadWorkers_.clear();
- }
-}
-
-bool ThreadManager::Impl::canSleep() {
- const Thread::id_t id = threadFactory_->getCurrentThreadId();
- return idMap_.find(id) == idMap_.end();
-}
-
-void ThreadManager::Impl::add(shared_ptr<Runnable> value, int64_t timeout, int64_t expiration) {
- Guard g(mutex_, timeout);
-
- if (!g) {
- throw TimedOutException();
- }
-
- if (state_ != ThreadManager::STARTED) {
- throw IllegalStateException(
- "ThreadManager::Impl::add ThreadManager "
- "not started");
- }
-
- removeExpiredTasks();
- if (pendingTaskCountMax_ > 0 && (tasks_.size() >= pendingTaskCountMax_)) {
- if (canSleep() && timeout >= 0) {
- while (pendingTaskCountMax_ > 0 && tasks_.size() >= pendingTaskCountMax_) {
- // This is thread safe because the mutex is shared between monitors.
- maxMonitor_.wait(timeout);
- }
- } else {
- throw TooManyPendingTasksException();
- }
- }
-
- tasks_.push(shared_ptr<ThreadManager::Task>(new ThreadManager::Task(value, expiration)));
-
- // If idle thread is available notify it, otherwise all worker threads are
- // running and will get around to this task in time.
- if (idleCount_ > 0) {
- monitor_.notify();
- }
-}
-
-void ThreadManager::Impl::remove(shared_ptr<Runnable> task) {
- (void)task;
- Synchronized s(monitor_);
- if (state_ != ThreadManager::STARTED) {
- throw IllegalStateException(
- "ThreadManager::Impl::remove ThreadManager not "
- "started");
- }
-}
-
-boost::shared_ptr<Runnable> ThreadManager::Impl::removeNextPending() {
- Guard g(mutex_);
- if (state_ != ThreadManager::STARTED) {
- throw IllegalStateException(
- "ThreadManager::Impl::removeNextPending "
- "ThreadManager not started");
- }
-
- if (tasks_.empty()) {
- return boost::shared_ptr<Runnable>();
- }
-
- shared_ptr<ThreadManager::Task> task = tasks_.front();
- tasks_.pop();
-
- return task->getRunnable();
-}
-
-void ThreadManager::Impl::removeExpiredTasks() {
- int64_t now = 0LL; // we won't ask for the time untile we need it
-
- // note that this loop breaks at the first non-expiring task
- while (!tasks_.empty()) {
- shared_ptr<ThreadManager::Task> task = tasks_.front();
- if (task->getExpireTime() == 0LL) {
- break;
- }
- if (now == 0LL) {
- now = Util::currentTime();
- }
- if (task->getExpireTime() > now) {
- break;
- }
- if (expireCallback_) {
- expireCallback_(task->getRunnable());
- }
- tasks_.pop();
- expiredCount_++;
- }
-}
-
-void ThreadManager::Impl::setExpireCallback(ExpireCallback expireCallback) {
- expireCallback_ = expireCallback;
-}
-
-class SimpleThreadManager : public ThreadManager::Impl {
-
-public:
- SimpleThreadManager(size_t workerCount = 4, size_t pendingTaskCountMax = 0)
- : workerCount_(workerCount), pendingTaskCountMax_(pendingTaskCountMax) {}
-
- void start() {
- ThreadManager::Impl::pendingTaskCountMax(pendingTaskCountMax_);
- ThreadManager::Impl::start();
- addWorker(workerCount_);
- }
-
-private:
- const size_t workerCount_;
- const size_t pendingTaskCountMax_;
- Monitor monitor_;
-};
-
-shared_ptr<ThreadManager> ThreadManager::newThreadManager() {
- return shared_ptr<ThreadManager>(new ThreadManager::Impl());
-}
-
-shared_ptr<ThreadManager> ThreadManager::newSimpleThreadManager(size_t count,
- size_t pendingTaskCountMax) {
- return shared_ptr<ThreadManager>(new SimpleThreadManager(count, pendingTaskCountMax));
-}
-}
-}
-} // apache::thrift::concurrency
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/928eade3/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/ThreadManager.h
----------------------------------------------------------------------
diff --git a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/ThreadManager.h b/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/ThreadManager.h
deleted file mode 100644
index 2112845..0000000
--- a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/ThreadManager.h
+++ /dev/null
@@ -1,198 +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.
- */
-
-#ifndef _THRIFT_CONCURRENCY_THREADMANAGER_H_
-#define _THRIFT_CONCURRENCY_THREADMANAGER_H_ 1
-
-#include <boost/shared_ptr.hpp>
-#include <thrift/cxxfunctional.h>
-#include <sys/types.h>
-#include <thrift/concurrency/Thread.h>
-
-namespace apache {
-namespace thrift {
-namespace concurrency {
-
-/**
- * Thread Pool Manager and related classes
- *
- * @version $Id:$
- */
-class ThreadManager;
-
-/**
- * ThreadManager class
- *
- * This class manages a pool of threads. It uses a ThreadFactory to create
- * threads. It never actually creates or destroys worker threads, rather
- * it maintains statistics on number of idle threads, number of active threads,
- * task backlog, and average wait and service times and informs the PoolPolicy
- * object bound to instances of this manager of interesting transitions. It is
- * then up the PoolPolicy object to decide if the thread pool size needs to be
- * adjusted and call this object addWorker and removeWorker methods to make
- * changes.
- *
- * This design allows different policy implementations to use this code to
- * handle basic worker thread management and worker task execution and focus on
- * policy issues. The simplest policy, StaticPolicy, does nothing other than
- * create a fixed number of threads.
- */
-class ThreadManager {
-
-protected:
- ThreadManager() {}
-
-public:
- typedef apache::thrift::stdcxx::function<void(boost::shared_ptr<Runnable>)> ExpireCallback;
-
- virtual ~ThreadManager() {}
-
- /**
- * Starts the thread manager. Verifies all attributes have been properly
- * initialized, then allocates necessary resources to begin operation
- */
- virtual void start() = 0;
-
- /**
- * Stops the thread manager. Aborts all remaining unprocessed task, shuts
- * down all created worker threads, and realeases all allocated resources.
- * This method blocks for all worker threads to complete, thus it can
- * potentially block forever if a worker thread is running a task that
- * won't terminate.
- */
- virtual void stop() = 0;
-
- /**
- * Joins the thread manager. This is the same as stop, except that it will
- * block until all the workers have finished their work. At that point
- * the ThreadManager will transition into the STOPPED state.
- */
- virtual void join() = 0;
-
- enum STATE { UNINITIALIZED, STARTING, STARTED, JOINING, STOPPING, STOPPED };
-
- virtual STATE state() const = 0;
-
- virtual boost::shared_ptr<ThreadFactory> threadFactory() const = 0;
-
- virtual void threadFactory(boost::shared_ptr<ThreadFactory> value) = 0;
-
- virtual void addWorker(size_t value = 1) = 0;
-
- virtual void removeWorker(size_t value = 1) = 0;
-
- /**
- * Gets the current number of idle worker threads
- */
- virtual size_t idleWorkerCount() const = 0;
-
- /**
- * Gets the current number of total worker threads
- */
- virtual size_t workerCount() const = 0;
-
- /**
- * Gets the current number of pending tasks
- */
- virtual size_t pendingTaskCount() const = 0;
-
- /**
- * Gets the current number of pending and executing tasks
- */
- virtual size_t totalTaskCount() const = 0;
-
- /**
- * Gets the maximum pending task count. 0 indicates no maximum
- */
- virtual size_t pendingTaskCountMax() const = 0;
-
- /**
- * Gets the number of tasks which have been expired without being run.
- */
- virtual size_t expiredTaskCount() = 0;
-
- /**
- * Adds a task to be executed at some time in the future by a worker thread.
- *
- * This method will block if pendingTaskCountMax() in not zero and pendingTaskCount()
- * is greater than or equalt to pendingTaskCountMax(). If this method is called in the
- * context of a ThreadManager worker thread it will throw a
- * TooManyPendingTasksException
- *
- * @param task The task to queue for execution
- *
- * @param timeout Time to wait in milliseconds to add a task when a pending-task-count
- * is specified. Specific cases:
- * timeout = 0 : Wait forever to queue task.
- * timeout = -1 : Return immediately if pending task count exceeds specified max
- * @param expiration when nonzero, the number of milliseconds the task is valid
- * to be run; if exceeded, the task will be dropped off the queue and not run.
- *
- * @throws TooManyPendingTasksException Pending task count exceeds max pending task count
- */
- virtual void add(boost::shared_ptr<Runnable> task,
- int64_t timeout = 0LL,
- int64_t expiration = 0LL) = 0;
-
- /**
- * Removes a pending task
- */
- virtual void remove(boost::shared_ptr<Runnable> task) = 0;
-
- /**
- * Remove the next pending task which would be run.
- *
- * @return the task removed.
- */
- virtual boost::shared_ptr<Runnable> removeNextPending() = 0;
-
- /**
- * Remove tasks from front of task queue that have expired.
- */
- virtual void removeExpiredTasks() = 0;
-
- /**
- * Set a callback to be called when a task is expired and not run.
- *
- * @param expireCallback a function called with the shared_ptr<Runnable> for
- * the expired task.
- */
- virtual void setExpireCallback(ExpireCallback expireCallback) = 0;
-
- static boost::shared_ptr<ThreadManager> newThreadManager();
-
- /**
- * Creates a simple thread manager the uses count number of worker threads and has
- * a pendingTaskCountMax maximum pending tasks. The default, 0, specified no limit
- * on pending tasks
- */
- static boost::shared_ptr<ThreadManager> newSimpleThreadManager(size_t count = 4,
- size_t pendingTaskCountMax = 0);
-
- class Task;
-
- class Worker;
-
- class Impl;
-};
-}
-}
-} // apache::thrift::concurrency
-
-#endif // #ifndef _THRIFT_CONCURRENCY_THREADMANAGER_H_
http://git-wip-us.apache.org/repos/asf/incubator-hawq/blob/928eade3/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/TimerManager.cpp
----------------------------------------------------------------------
diff --git a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/TimerManager.cpp b/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/TimerManager.cpp
deleted file mode 100644
index 122d26e..0000000
--- a/depends/thirdparty/thrift/lib/cpp/src/thrift/concurrency/TimerManager.cpp
+++ /dev/null
@@ -1,306 +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 <thrift/concurrency/TimerManager.h>
-#include <thrift/concurrency/Exception.h>
-#include <thrift/concurrency/Util.h>
-
-#include <assert.h>
-#include <iostream>
-#include <set>
-
-namespace apache {
-namespace thrift {
-namespace concurrency {
-
-using boost::shared_ptr;
-
-/**
- * TimerManager class
- *
- * @version $Id:$
- */
-class TimerManager::Task : public Runnable {
-
-public:
- enum STATE { WAITING, EXECUTING, CANCELLED, COMPLETE };
-
- Task(shared_ptr<Runnable> runnable) : runnable_(runnable), state_(WAITING) {}
-
- ~Task() {}
-
- void run() {
- if (state_ == EXECUTING) {
- runnable_->run();
- state_ = COMPLETE;
- }
- }
-
-private:
- shared_ptr<Runnable> runnable_;
- friend class TimerManager::Dispatcher;
- STATE state_;
-};
-
-class TimerManager::Dispatcher : public Runnable {
-
-public:
- Dispatcher(TimerManager* manager) : manager_(manager) {}
-
- ~Dispatcher() {}
-
- /**
- * Dispatcher entry point
- *
- * As long as dispatcher thread is running, pull tasks off the task taskMap_
- * and execute.
- */
- void run() {
- {
- Synchronized s(manager_->monitor_);
- if (manager_->state_ == TimerManager::STARTING) {
- manager_->state_ = TimerManager::STARTED;
- manager_->monitor_.notifyAll();
- }
- }
-
- do {
- std::set<shared_ptr<TimerManager::Task> > expiredTasks;
- {
- Synchronized s(manager_->monitor_);
- task_iterator expiredTaskEnd;
- int64_t now = Util::currentTime();
- while (manager_->state_ == TimerManager::STARTED
- && (expiredTaskEnd = manager_->taskMap_.upper_bound(now))
- == manager_->taskMap_.begin()) {
- int64_t timeout = 0LL;
- if (!manager_->taskMap_.empty()) {
- timeout = manager_->taskMap_.begin()->first - now;
- }
- assert((timeout != 0 && manager_->taskCount_ > 0)
- || (timeout == 0 && manager_->taskCount_ == 0));
- try {
- manager_->monitor_.wait(timeout);
- } catch (TimedOutException&) {
- }
- now = Util::currentTime();
- }
-
- if (manager_->state_ == TimerManager::STARTED) {
- for (task_iterator ix = manager_->taskMap_.begin(); ix != expiredTaskEnd; ix++) {
- shared_ptr<TimerManager::Task> task = ix->second;
- expiredTasks.insert(task);
- if (task->state_ == TimerManager::Task::WAITING) {
- task->state_ = TimerManager::Task::EXECUTING;
- }
- manager_->taskCount_--;
- }
- manager_->taskMap_.erase(manager_->taskMap_.begin(), expiredTaskEnd);
- }
- }
-
- for (std::set<shared_ptr<Task> >::iterator ix = expiredTasks.begin();
- ix != expiredTasks.end();
- ++ix) {
- (*ix)->run();
- }
-
- } while (manager_->state_ == TimerManager::STARTED);
-
- {
- Synchronized s(manager_->monitor_);
- if (manager_->state_ == TimerManager::STOPPING) {
- manager_->state_ = TimerManager::STOPPED;
- manager_->monitor_.notify();
- }
- }
- return;
- }
-
-private:
- TimerManager* manager_;
- friend class TimerManager;
-};
-
-#if defined(_MSC_VER)
-#pragma warning(push)
-#pragma warning(disable : 4355) // 'this' used in base member initializer list
-#endif
-
-TimerManager::TimerManager()
- : taskCount_(0),
- state_(TimerManager::UNINITIALIZED),
- dispatcher_(shared_ptr<Dispatcher>(new Dispatcher(this))) {
-}
-
-#if defined(_MSC_VER)
-#pragma warning(pop)
-#endif
-
-TimerManager::~TimerManager() {
-
- // If we haven't been explicitly stopped, do so now. We don't need to grab
- // the monitor here, since stop already takes care of reentrancy.
-
- if (state_ != STOPPED) {
- try {
- stop();
- } catch (...) {
- throw;
- // uhoh
- }
- }
-}
-
-void TimerManager::start() {
- bool doStart = false;
- {
- Synchronized s(monitor_);
- if (!threadFactory_) {
- throw InvalidArgumentException();
- }
- if (state_ == TimerManager::UNINITIALIZED) {
- state_ = TimerManager::STARTING;
- doStart = true;
- }
- }
-
- if (doStart) {
- dispatcherThread_ = threadFactory_->newThread(dispatcher_);
- dispatcherThread_->start();
- }
-
- {
- Synchronized s(monitor_);
- while (state_ == TimerManager::STARTING) {
- monitor_.wait();
- }
- assert(state_ != TimerManager::STARTING);
- }
-}
-
-void TimerManager::stop() {
- bool doStop = false;
- {
- Synchronized s(monitor_);
- if (state_ == TimerManager::UNINITIALIZED) {
- state_ = TimerManager::STOPPED;
- } else if (state_ != STOPPING && state_ != STOPPED) {
- doStop = true;
- state_ = STOPPING;
- monitor_.notifyAll();
- }
- while (state_ != STOPPED) {
- monitor_.wait();
- }
- }
-
- if (doStop) {
- // Clean up any outstanding tasks
- taskMap_.clear();
-
- // Remove dispatcher's reference to us.
- dispatcher_->manager_ = NULL;
- }
-}
-
-shared_ptr<const ThreadFactory> TimerManager::threadFactory() const {
- Synchronized s(monitor_);
- return threadFactory_;
-}
-
-void TimerManager::threadFactory(shared_ptr<const ThreadFactory> value) {
- Synchronized s(monitor_);
- threadFactory_ = value;
-}
-
-size_t TimerManager::taskCount() const {
- return taskCount_;
-}
-
-void TimerManager::add(shared_ptr<Runnable> task, int64_t timeout) {
- int64_t now = Util::currentTime();
- timeout += now;
-
- {
- Synchronized s(monitor_);
- if (state_ != TimerManager::STARTED) {
- throw IllegalStateException();
- }
-
- // If the task map is empty, we will kick the dispatcher for sure. Otherwise, we kick him
- // if the expiration time is shorter than the current value. Need to test before we insert,
- // because the new task might insert at the front.
- bool notifyRequired = (taskCount_ == 0) ? true : timeout < taskMap_.begin()->first;
-
- taskCount_++;
- taskMap_.insert(
- std::pair<int64_t, shared_ptr<Task> >(timeout, shared_ptr<Task>(new Task(task))));
-
- // If the task map was empty, or if we have an expiration that is earlier
- // than any previously seen, kick the dispatcher so it can update its
- // timeout
- if (notifyRequired) {
- monitor_.notify();
- }
- }
-}
-
-void TimerManager::add(shared_ptr<Runnable> task, const struct THRIFT_TIMESPEC& value) {
-
- int64_t expiration;
- Util::toMilliseconds(expiration, value);
-
- int64_t now = Util::currentTime();
-
- if (expiration < now) {
- throw InvalidArgumentException();
- }
-
- add(task, expiration - now);
-}
-
-void TimerManager::add(shared_ptr<Runnable> task, const struct timeval& value) {
-
- int64_t expiration;
- Util::toMilliseconds(expiration, value);
-
- int64_t now = Util::currentTime();
-
- if (expiration < now) {
- throw InvalidArgumentException();
- }
-
- add(task, expiration - now);
-}
-
-void TimerManager::remove(shared_ptr<Runnable> task) {
- (void)task;
- Synchronized s(monitor_);
- if (state_ != TimerManager::STARTED) {
- throw IllegalStateException();
- }
-}
-
-TimerManager::STATE TimerManager::state() const {
- return state_;
-}
-}
-}
-} // apache::thrift::concurrency