THRIFT-3932 fixed ThreadManager concurrency issues, added more tests in that area, did a little refactoring and prettying up along the way
Client: C++
This closes #1103
diff --git a/lib/cpp/test/concurrency/Tests.cpp b/lib/cpp/test/concurrency/Tests.cpp
index 0d81d7e..33af392 100644
--- a/lib/cpp/test/concurrency/Tests.cpp
+++ b/lib/cpp/test/concurrency/Tests.cpp
@@ -45,25 +45,38 @@
std::cout << "ThreadFactory tests..." << std::endl;
- size_t count = 1000;
- size_t floodLoops = 1;
- size_t floodCount = 100000;
+ int reapLoops = 20;
+ int reapCount = 1000;
+ size_t floodLoops = 3;
+ size_t floodCount = 20000;
- std::cout << "\t\tThreadFactory reap N threads test: N = " << count << std::endl;
+ std::cout << "\t\tThreadFactory reap N threads test: N = " << reapLoops << "x" << reapCount << std::endl;
- assert(threadFactoryTests.reapNThreads(count));
+ if (!threadFactoryTests.reapNThreads(reapLoops, reapCount)) {
+ std::cerr << "\t\ttThreadFactory reap N threads FAILED" << std::endl;
+ return 1;
+ }
- std::cout << "\t\tThreadFactory floodN threads test: N = " << floodCount << std::endl;
+ std::cout << "\t\tThreadFactory flood N threads test: N = " << floodLoops << "x" << floodCount << std::endl;
- assert(threadFactoryTests.floodNTest(floodLoops, floodCount));
+ if (!threadFactoryTests.floodNTest(floodLoops, floodCount)) {
+ std::cerr << "\t\ttThreadFactory flood N threads FAILED" << std::endl;
+ return 1;
+ }
std::cout << "\t\tThreadFactory synchronous start test" << std::endl;
- assert(threadFactoryTests.synchStartTest());
+ if (!threadFactoryTests.synchStartTest()) {
+ std::cerr << "\t\ttThreadFactory synchronous start FAILED" << std::endl;
+ return 1;
+ }
std::cout << "\t\tThreadFactory monitor timeout test" << std::endl;
- assert(threadFactoryTests.monitorTimeoutTest());
+ if (!threadFactoryTests.monitorTimeoutTest()) {
+ std::cerr << "\t\ttThreadFactory monitor timeout FAILED" << std::endl;
+ return 1;
+ }
}
if (runAll || args[0].compare("util") == 0) {
@@ -97,7 +110,10 @@
TimerManagerTests timerManagerTests;
- assert(timerManagerTests.test00());
+ if (!timerManagerTests.test00()) {
+ std::cerr << "\t\tTimerManager tests FAILED" << std::endl;
+ return 1;
+ }
}
if (runAll || args[0].compare("thread-manager") == 0) {
@@ -105,24 +121,34 @@
std::cout << "ThreadManager tests..." << std::endl;
{
-
size_t workerCount = 100;
-
- size_t taskCount = 100000;
-
+ size_t taskCount = 50000;
int64_t delay = 10LL;
+ ThreadManagerTests threadManagerTests;
+
+ std::cout << "\t\tThreadManager api test:" << std::endl;
+
+ if (!threadManagerTests.apiTest()) {
+ std::cerr << "\t\tThreadManager apiTest FAILED" << std::endl;
+ return 1;
+ }
+
std::cout << "\t\tThreadManager load test: worker count: " << workerCount
<< " task count: " << taskCount << " delay: " << delay << std::endl;
- ThreadManagerTests threadManagerTests;
-
- assert(threadManagerTests.loadTest(taskCount, delay, workerCount));
+ if (!threadManagerTests.loadTest(taskCount, delay, workerCount)) {
+ std::cerr << "\t\tThreadManager loadTest FAILED" << std::endl;
+ return 1;
+ }
std::cout << "\t\tThreadManager block test: worker count: " << workerCount
<< " delay: " << delay << std::endl;
- assert(threadManagerTests.blockTest(delay, workerCount));
+ if (!threadManagerTests.blockTest(delay, workerCount)) {
+ std::cerr << "\t\tThreadManager blockTest FAILED" << std::endl;
+ return 1;
+ }
}
}
@@ -134,13 +160,13 @@
size_t minWorkerCount = 2;
- size_t maxWorkerCount = 512;
+ size_t maxWorkerCount = 64;
size_t tasksPerWorker = 1000;
- int64_t delay = 10LL;
+ int64_t delay = 5LL;
- for (size_t workerCount = minWorkerCount; workerCount < maxWorkerCount; workerCount *= 2) {
+ for (size_t workerCount = minWorkerCount; workerCount < maxWorkerCount; workerCount *= 4) {
size_t taskCount = workerCount * tasksPerWorker;
@@ -149,8 +175,15 @@
ThreadManagerTests threadManagerTests;
- threadManagerTests.loadTest(taskCount, delay, workerCount);
+ if (!threadManagerTests.loadTest(taskCount, delay, workerCount))
+ {
+ std::cerr << "\t\tThreadManager loadTest FAILED" << std::endl;
+ return 1;
+ }
}
}
}
+
+ std::cout << "ALL TESTS PASSED" << std::endl;
+ return 0;
}
diff --git a/lib/cpp/test/concurrency/ThreadFactoryTests.h b/lib/cpp/test/concurrency/ThreadFactoryTests.h
index 3ad14ca..4fc688c 100644
--- a/lib/cpp/test/concurrency/ThreadFactoryTests.h
+++ b/lib/cpp/test/concurrency/ThreadFactoryTests.h
@@ -43,36 +43,6 @@
class ThreadFactoryTests {
public:
- static const double TEST_TOLERANCE;
-
- class Task : public Runnable {
-
- public:
- Task() {}
-
- void run() { std::cout << "\t\t\tHello World" << std::endl; }
- };
-
- /**
- * Hello world test
- */
- bool helloWorldTest() {
-
- PlatformThreadFactory threadFactory = PlatformThreadFactory();
-
- shared_ptr<Task> task = shared_ptr<Task>(new ThreadFactoryTests::Task());
-
- shared_ptr<Thread> thread = threadFactory.newThread(task);
-
- thread->start();
-
- thread->join();
-
- std::cout << "\t\t\tSuccess!" << std::endl;
-
- return true;
- }
-
/**
* Reap N threads
*/
@@ -244,15 +214,22 @@
return true;
}
- /** See how accurate monitor timeout is. */
+ /**
+ * The only guarantee a monitor timeout can give you is that
+ * it will take "at least" as long as the timeout, no less.
+ * There is absolutely no guarantee around regaining execution
+ * near the timeout. On a busy system (like inside a third party
+ * CI environment) it could take quite a bit longer than the
+ * requested timeout, and that's ok.
+ */
- bool monitorTimeoutTest(size_t count = 1000, int64_t timeout = 10) {
+ bool monitorTimeoutTest(int64_t count = 1000, int64_t timeout = 2) {
Monitor monitor;
int64_t startTime = Util::currentTime();
- for (size_t ix = 0; ix < count; ix++) {
+ for (int64_t ix = 0; ix < count; ix++) {
{
Synchronized s(monitor);
try {
@@ -264,18 +241,11 @@
int64_t endTime = Util::currentTime();
- double error = ((endTime - startTime) - (count * timeout)) / (double)(count * timeout);
-
- if (error < 0.0) {
-
- error *= 1.0;
- }
-
- bool success = error < ThreadFactoryTests::TEST_TOLERANCE;
+ bool success = (endTime - startTime) >= (count * timeout);
std::cout << "\t\t\t" << (success ? "Success" : "Failure")
- << "! expected time: " << count * timeout
- << "ms elapsed time: " << endTime - startTime << "ms error%: " << error * 100.0
+ << ": minimum required time to elapse " << count * timeout
+ << "ms; actual elapsed time " << endTime - startTime << "ms"
<< std::endl;
return success;
@@ -285,17 +255,15 @@
public:
FloodTask(const size_t id) : _id(id) {}
~FloodTask() {
- if (_id % 1000 == 0) {
+ if (_id % 10000 == 0) {
std::cout << "\t\tthread " << _id << " done" << std::endl;
}
}
void run() {
- if (_id % 1000 == 0) {
+ if (_id % 10000 == 0) {
std::cout << "\t\tthread " << _id << " started" << std::endl;
}
-
- THRIFT_SLEEP_USEC(1);
}
const size_t _id;
};
@@ -321,8 +289,6 @@
thread->start();
- THRIFT_SLEEP_USEC(1);
-
} catch (TException& e) {
std::cout << "\t\t\tfailed to start " << lix* count + tix << " thread " << e.what()
@@ -341,7 +307,6 @@
}
};
-const double ThreadFactoryTests::TEST_TOLERANCE = .20;
}
}
}
diff --git a/lib/cpp/test/concurrency/ThreadManagerTests.h b/lib/cpp/test/concurrency/ThreadManagerTests.h
index b196813..b5925ac 100644
--- a/lib/cpp/test/concurrency/ThreadManagerTests.h
+++ b/lib/cpp/test/concurrency/ThreadManagerTests.h
@@ -24,9 +24,9 @@
#include <thrift/concurrency/Util.h>
#include <assert.h>
+#include <deque>
#include <set>
#include <iostream>
-#include <set>
#include <stdint.h>
namespace apache {
@@ -36,9 +36,26 @@
using namespace apache::thrift::concurrency;
-class ThreadManagerTests {
+static std::deque<boost::shared_ptr<Runnable> > m_expired;
+static void expiredNotifier(boost::shared_ptr<Runnable> runnable)
+{
+ m_expired.push_back(runnable);
+}
- static const double TEST_TOLERANCE;
+static void sleep_(int64_t millisec) {
+ Monitor _sleep;
+ Synchronized s(_sleep);
+
+ try {
+ _sleep.wait(millisec);
+ } catch (TimedOutException&) {
+ ;
+ } catch (...) {
+ assert(0);
+ }
+}
+
+class ThreadManagerTests {
public:
class Task : public Runnable {
@@ -51,17 +68,7 @@
_startTime = Util::currentTime();
- {
- Synchronized s(_sleep);
-
- try {
- _sleep.wait(_timeout);
- } catch (TimedOutException&) {
- ;
- } catch (...) {
- assert(0);
- }
- }
+ sleep_(_timeout);
_endTime = Util::currentTime();
@@ -73,9 +80,7 @@
// std::cout << "Thread " << _count << " completed " << std::endl;
_count--;
-
- if (_count == 0) {
-
+ if (_count % 10000 == 0) {
_monitor.notify();
}
}
@@ -130,11 +135,13 @@
threadManager->add(*ix);
}
+ std::cout << "\t\t\t\tloaded " << count << " tasks to execute" << std::endl;
+
{
Synchronized s(monitor);
while (activeCount > 0) {
-
+ std::cout << "\t\t\t\tactiveCount = " << activeCount << std::endl;
monitor.wait();
}
}
@@ -179,23 +186,15 @@
averageTime /= count;
- std::cout << "\t\t\tfirst start: " << firstTime << "ms Last end: " << lastTime
- << "ms min: " << minTime << "ms max: " << maxTime << "ms average: " << averageTime
+ std::cout << "\t\t\tfirst start: " << firstTime << " Last end: " << lastTime
+ << " min: " << minTime << "ms max: " << maxTime << "ms average: " << averageTime
<< "ms" << std::endl;
- double expectedTime = (double(count + (workerCount - 1)) / workerCount) * timeout;
-
- double error = ((time01 - time00) - expectedTime) / expectedTime;
-
- if (error < 0) {
- error *= -1.0;
- }
-
- bool success = error < TEST_TOLERANCE;
+ bool success = (time01 - time00) >= ((int64_t)count * timeout) / (int64_t)workerCount;
std::cout << "\t\t\t" << (success ? "Success" : "Failure")
- << "! expected time: " << expectedTime << "ms elapsed time: " << time01 - time00
- << "ms error%: " << error * 100.0 << std::endl;
+ << "! expected time: " << ((int64_t)count * timeout) / (int64_t)workerCount << "ms elapsed time: " << time01 - time00
+ << "ms" << std::endl;
return success;
}
@@ -203,30 +202,36 @@
class BlockTask : public Runnable {
public:
- BlockTask(Monitor& monitor, Monitor& bmonitor, size_t& count)
- : _monitor(monitor), _bmonitor(bmonitor), _count(count) {}
+ BlockTask(Monitor& entryMonitor, Monitor& blockMonitor, bool& blocked, Monitor& doneMonitor, size_t& count)
+ : _entryMonitor(entryMonitor), _entered(false), _blockMonitor(blockMonitor), _blocked(blocked), _doneMonitor(doneMonitor), _count(count) {}
void run() {
{
- Synchronized s(_bmonitor);
-
- _bmonitor.wait();
+ Synchronized s(_entryMonitor);
+ _entered = true;
+ _entryMonitor.notify();
}
{
- Synchronized s(_monitor);
+ Synchronized s(_blockMonitor);
+ while (_blocked) {
+ _blockMonitor.wait();
+ }
+ }
- _count--;
-
- if (_count == 0) {
-
- _monitor.notify();
+ {
+ Synchronized s(_doneMonitor);
+ if (--_count == 0) {
+ _doneMonitor.notify();
}
}
}
- Monitor& _monitor;
- Monitor& _bmonitor;
+ Monitor& _entryMonitor;
+ bool _entered;
+ Monitor& _blockMonitor;
+ bool& _blocked;
+ Monitor& _doneMonitor;
size_t& _count;
};
@@ -240,8 +245,10 @@
try {
- Monitor bmonitor;
- Monitor monitor;
+ Monitor entryMonitor; // not used by this test
+ Monitor blockMonitor;
+ bool blocked[] = {true, true, true};
+ Monitor doneMonitor;
size_t pendingTaskMaxCount = workerCount;
@@ -260,21 +267,22 @@
threadManager->start();
- std::set<shared_ptr<ThreadManagerTests::BlockTask> > tasks;
+ std::vector<shared_ptr<ThreadManagerTests::BlockTask> > tasks;
+ tasks.reserve(workerCount + pendingTaskMaxCount);
for (size_t ix = 0; ix < workerCount; ix++) {
- tasks.insert(shared_ptr<ThreadManagerTests::BlockTask>(
- new ThreadManagerTests::BlockTask(monitor, bmonitor, activeCounts[0])));
+ tasks.push_back(shared_ptr<ThreadManagerTests::BlockTask>(
+ new ThreadManagerTests::BlockTask(entryMonitor, blockMonitor, blocked[0], doneMonitor, activeCounts[0])));
}
for (size_t ix = 0; ix < pendingTaskMaxCount; ix++) {
- tasks.insert(shared_ptr<ThreadManagerTests::BlockTask>(
- new ThreadManagerTests::BlockTask(monitor, bmonitor, activeCounts[1])));
+ tasks.push_back(shared_ptr<ThreadManagerTests::BlockTask>(
+ new ThreadManagerTests::BlockTask(entryMonitor, blockMonitor, blocked[1], doneMonitor, activeCounts[1])));
}
- for (std::set<shared_ptr<ThreadManagerTests::BlockTask> >::iterator ix = tasks.begin();
+ for (std::vector<shared_ptr<ThreadManagerTests::BlockTask> >::iterator ix = tasks.begin();
ix != tasks.end();
ix++) {
threadManager->add(*ix);
@@ -285,7 +293,7 @@
}
shared_ptr<ThreadManagerTests::BlockTask> extraTask(
- new ThreadManagerTests::BlockTask(monitor, bmonitor, activeCounts[2]));
+ new ThreadManagerTests::BlockTask(entryMonitor, blockMonitor, blocked[2], doneMonitor, activeCounts[2]));
try {
threadManager->add(extraTask, 1);
@@ -309,16 +317,15 @@
<< "Pending tasks " << threadManager->pendingTaskCount() << std::endl;
{
- Synchronized s(bmonitor);
-
- bmonitor.notifyAll();
+ Synchronized s(blockMonitor);
+ blocked[0] = false;
+ blockMonitor.notifyAll();
}
{
- Synchronized s(monitor);
-
+ Synchronized s(doneMonitor);
while (activeCounts[0] != 0) {
- monitor.wait();
+ doneMonitor.wait();
}
}
@@ -341,37 +348,37 @@
// Wake up tasks that were pending before and wait for them to complete
{
- Synchronized s(bmonitor);
-
- bmonitor.notifyAll();
+ Synchronized s(blockMonitor);
+ blocked[1] = false;
+ blockMonitor.notifyAll();
}
{
- Synchronized s(monitor);
-
+ Synchronized s(doneMonitor);
while (activeCounts[1] != 0) {
- monitor.wait();
+ doneMonitor.wait();
}
}
// Wake up the extra task and wait for it to complete
{
- Synchronized s(bmonitor);
-
- bmonitor.notifyAll();
+ Synchronized s(blockMonitor);
+ blocked[2] = false;
+ blockMonitor.notifyAll();
}
{
- Synchronized s(monitor);
-
+ Synchronized s(doneMonitor);
while (activeCounts[2] != 0) {
- monitor.wait();
+ doneMonitor.wait();
}
}
+ threadManager->stop();
+
if (!(success = (threadManager->totalTaskCount() == 0))) {
- throw TException("Unexpected pending task count");
+ throw TException("Unexpected total task count");
}
} catch (TException& e) {
@@ -381,9 +388,295 @@
std::cout << "\t\t\t" << (success ? "Success" : "Failure") << std::endl;
return success;
}
+
+
+ bool apiTest() {
+
+ // prove currentTime has milliseconds granularity since many other things depend on it
+ int64_t a = Util::currentTime();
+ sleep_(100);
+ int64_t b = Util::currentTime();
+ if (b - a < 50 || b - a > 150) {
+ std::cerr << "\t\t\texpected 100ms gap, found " << (b-a) << "ms gap instead." << std::endl;
+ return false;
+ }
+
+#if !USE_BOOST_THREAD && !USE_STD_THREAD
+ // test once with a detached thread factory and once with a joinable thread factory
+
+ shared_ptr<PosixThreadFactory> threadFactory
+ = shared_ptr<PosixThreadFactory>(new PosixThreadFactory(false));
+
+ std::cout << "\t\t\tapiTest with joinable thread factory" << std::endl;
+ if (!apiTestWithThreadFactory(threadFactory)) {
+ return false;
+ }
+
+ threadFactory.reset(new PosixThreadFactory(true));
+ std::cout << "\t\t\tapiTest with detached thread factory" << std::endl;
+ return apiTestWithThreadFactory(threadFactory);
+#else
+ return apiTestWithThreadFactory(shared_ptr<PlatformThreadFactory>(new PlatformThreadFactory()));
+#endif
+
+ }
+
+ bool apiTestWithThreadFactory(shared_ptr<PlatformThreadFactory> threadFactory)
+ {
+ shared_ptr<ThreadManager> threadManager = ThreadManager::newSimpleThreadManager(1);
+ threadManager->threadFactory(threadFactory);
+
+#if !USE_BOOST_THREAD && !USE_STD_THREAD
+ threadFactory->setPriority(PosixThreadFactory::HIGHEST);
+
+ // verify we cannot change the thread factory to one with the opposite detached setting
+ shared_ptr<PlatformThreadFactory> threadFactory2
+ = shared_ptr<PosixThreadFactory>(new PlatformThreadFactory(
+ PosixThreadFactory::ROUND_ROBIN,
+ PosixThreadFactory::NORMAL,
+ 1,
+ !threadFactory->isDetached()));
+ try {
+ threadManager->threadFactory(threadFactory2);
+ // if the call succeeded we changed the thread factory to one that had the opposite setting for "isDetached()".
+ // this is bad, because the thread manager checks with the thread factory to see if it should join threads
+ // as they are leaving - so the detached status of new threads cannot change while there are existing threads.
+ std::cerr << "\t\t\tShould not be able to change thread factory detached disposition" << std::endl;
+ return false;
+ }
+ catch (InvalidArgumentException& ex) {
+ /* expected */
+ }
+#endif
+
+ std::cout << "\t\t\t\tstarting.. " << std::endl;
+
+ threadManager->start();
+ threadManager->setExpireCallback(expiredNotifier); // apache::thrift::stdcxx::bind(&ThreadManagerTests::expiredNotifier, this));
+
+#define EXPECT(FUNC, COUNT) { size_t c = FUNC; if (c != COUNT) { std::cerr << "expected " #FUNC" to be " #COUNT ", but was " << c << std::endl; return false; } }
+
+ EXPECT(threadManager->workerCount(), 1);
+ EXPECT(threadManager->idleWorkerCount(), 1);
+ EXPECT(threadManager->pendingTaskCount(), 0);
+
+ std::cout << "\t\t\t\tadd 2nd worker.. " << std::endl;
+
+ threadManager->addWorker();
+
+ EXPECT(threadManager->workerCount(), 2);
+ EXPECT(threadManager->idleWorkerCount(), 2);
+ EXPECT(threadManager->pendingTaskCount(), 0);
+
+ std::cout << "\t\t\t\tremove 2nd worker.. " << std::endl;
+
+ threadManager->removeWorker();
+
+ EXPECT(threadManager->workerCount(), 1);
+ EXPECT(threadManager->idleWorkerCount(), 1);
+ EXPECT(threadManager->pendingTaskCount(), 0);
+
+ std::cout << "\t\t\t\tremove 1st worker.. " << std::endl;
+
+ threadManager->removeWorker();
+
+ EXPECT(threadManager->workerCount(), 0);
+ EXPECT(threadManager->idleWorkerCount(), 0);
+ EXPECT(threadManager->pendingTaskCount(), 0);
+
+ std::cout << "\t\t\t\tadd blocking task.. " << std::endl;
+
+ // We're going to throw a blocking task into the mix
+ Monitor entryMonitor; // signaled when task is running
+ Monitor blockMonitor; // to be signaled to unblock the task
+ bool blocked(true); // set to false before notifying
+ Monitor doneMonitor; // signaled when count reaches zero
+ size_t activeCount = 1;
+ shared_ptr<ThreadManagerTests::BlockTask> blockingTask(
+ new ThreadManagerTests::BlockTask(entryMonitor, blockMonitor, blocked, doneMonitor, activeCount));
+ threadManager->add(blockingTask);
+
+ EXPECT(threadManager->workerCount(), 0);
+ EXPECT(threadManager->idleWorkerCount(), 0);
+ EXPECT(threadManager->pendingTaskCount(), 1);
+
+ std::cout << "\t\t\t\tadd other task.. " << std::endl;
+
+ shared_ptr<ThreadManagerTests::Task> otherTask(
+ new ThreadManagerTests::Task(doneMonitor, activeCount, 0));
+
+ threadManager->add(otherTask);
+
+ EXPECT(threadManager->workerCount(), 0);
+ EXPECT(threadManager->idleWorkerCount(), 0);
+ EXPECT(threadManager->pendingTaskCount(), 2);
+
+ std::cout << "\t\t\t\tremove blocking task specifically.. " << std::endl;
+
+ threadManager->remove(blockingTask);
+
+ EXPECT(threadManager->workerCount(), 0);
+ EXPECT(threadManager->idleWorkerCount(), 0);
+ EXPECT(threadManager->pendingTaskCount(), 1);
+
+ std::cout << "\t\t\t\tremove next pending task.." << std::endl;
+
+ shared_ptr<Runnable> nextTask = threadManager->removeNextPending();
+ if (nextTask != otherTask) {
+ std::cerr << "\t\t\t\t\texpected removeNextPending to return otherTask" << std::endl;
+ return false;
+ }
+
+ EXPECT(threadManager->workerCount(), 0);
+ EXPECT(threadManager->idleWorkerCount(), 0);
+ EXPECT(threadManager->pendingTaskCount(), 0);
+
+ std::cout << "\t\t\t\tremove next pending task (none left).." << std::endl;
+
+ nextTask = threadManager->removeNextPending();
+ if (nextTask) {
+ std::cerr << "\t\t\t\t\texpected removeNextPending to return an empty Runnable" << std::endl;
+ return false;
+ }
+
+ std::cout << "\t\t\t\tadd 2 expired tasks and 1 not.." << std::endl;
+
+ shared_ptr<ThreadManagerTests::Task> expiredTask(
+ new ThreadManagerTests::Task(doneMonitor, activeCount, 0));
+
+ threadManager->add(expiredTask, 0, 1);
+ threadManager->add(blockingTask); // add one that hasn't expired to make sure it gets skipped
+ threadManager->add(expiredTask, 0, 1); // add a second expired to ensure removeExpiredTasks removes both
+
+ sleep_(50); // make sure enough time elapses for it to expire - the shortest expiration time is 1 millisecond
+
+ EXPECT(threadManager->workerCount(), 0);
+ EXPECT(threadManager->idleWorkerCount(), 0);
+ EXPECT(threadManager->pendingTaskCount(), 3);
+ EXPECT(threadManager->expiredTaskCount(), 0);
+
+ std::cout << "\t\t\t\tremove expired tasks.." << std::endl;
+
+ if (!m_expired.empty()) {
+ std::cerr << "\t\t\t\t\texpected m_expired to be empty" << std::endl;
+ return false;
+ }
+
+ threadManager->removeExpiredTasks();
+
+ if (m_expired.size() != 2) {
+ std::cerr << "\t\t\t\t\texpected m_expired to be set" << std::endl;
+ return false;
+ }
+
+ if (m_expired.front() != expiredTask) {
+ std::cerr << "\t\t\t\t\texpected m_expired[0] to be the expired task" << std::endl;
+ return false;
+ }
+ m_expired.pop_front();
+
+ if (m_expired.front() != expiredTask) {
+ std::cerr << "\t\t\t\t\texpected m_expired[1] to be the expired task" << std::endl;
+ return false;
+ }
+
+ m_expired.clear();
+
+ threadManager->remove(blockingTask);
+
+ EXPECT(threadManager->workerCount(), 0);
+ EXPECT(threadManager->idleWorkerCount(), 0);
+ EXPECT(threadManager->pendingTaskCount(), 0);
+ EXPECT(threadManager->expiredTaskCount(), 2);
+
+ std::cout << "\t\t\t\tadd expired task (again).." << std::endl;
+
+ threadManager->add(expiredTask, 0, 1); // expires in 1ms
+ sleep_(50); // make sure enough time elapses for it to expire - the shortest expiration time is 1ms
+
+ std::cout << "\t\t\t\tadd worker to consume expired task.." << std::endl;
+
+ threadManager->addWorker();
+ sleep_(100); // make sure it has time to spin up and expire the task
+
+ if (m_expired.empty()) {
+ std::cerr << "\t\t\t\t\texpected m_expired to be set" << std::endl;
+ return false;
+ }
+
+ if (m_expired.front() != expiredTask) {
+ std::cerr << "\t\t\t\t\texpected m_expired to be the expired task" << std::endl;
+ return false;
+ }
+
+ m_expired.clear();
+
+ EXPECT(threadManager->workerCount(), 1);
+ EXPECT(threadManager->idleWorkerCount(), 1);
+ EXPECT(threadManager->pendingTaskCount(), 0);
+ EXPECT(threadManager->expiredTaskCount(), 3);
+
+ std::cout << "\t\t\t\ttry to remove too many workers" << std::endl;
+ try {
+ threadManager->removeWorker(2);
+ std::cerr << "\t\t\t\t\texpected InvalidArgumentException" << std::endl;
+ return false;
+ } catch (const InvalidArgumentException&) {
+ /* expected */
+ }
+
+ std::cout << "\t\t\t\tremove worker.. " << std::endl;
+
+ threadManager->removeWorker();
+
+ EXPECT(threadManager->workerCount(), 0);
+ EXPECT(threadManager->idleWorkerCount(), 0);
+ EXPECT(threadManager->pendingTaskCount(), 0);
+ EXPECT(threadManager->expiredTaskCount(), 3);
+
+ std::cout << "\t\t\t\tadd blocking task.. " << std::endl;
+
+ threadManager->add(blockingTask);
+
+ EXPECT(threadManager->workerCount(), 0);
+ EXPECT(threadManager->idleWorkerCount(), 0);
+ EXPECT(threadManager->pendingTaskCount(), 1);
+
+ std::cout << "\t\t\t\tadd worker.. " << std::endl;
+
+ threadManager->addWorker();
+ {
+ Synchronized s(entryMonitor);
+ while (!blockingTask->_entered) {
+ entryMonitor.wait();
+ }
+ }
+
+ EXPECT(threadManager->workerCount(), 1);
+ EXPECT(threadManager->idleWorkerCount(), 0);
+ EXPECT(threadManager->pendingTaskCount(), 0);
+
+ std::cout << "\t\t\t\tunblock task and remove worker.. " << std::endl;
+
+ {
+ Synchronized s(blockMonitor);
+ blocked = false;
+ blockMonitor.notifyAll();
+ }
+ threadManager->removeWorker();
+
+ EXPECT(threadManager->workerCount(), 0);
+ EXPECT(threadManager->idleWorkerCount(), 0);
+ EXPECT(threadManager->pendingTaskCount(), 0);
+
+ std::cout << "\t\t\t\tcleanup.. " << std::endl;
+
+ blockingTask.reset();
+ threadManager.reset();
+ return true;
+ }
};
-const double ThreadManagerTests::TEST_TOLERANCE = .20;
}
}
}
diff --git a/lib/cpp/test/concurrency/TimerManagerTests.h b/lib/cpp/test/concurrency/TimerManagerTests.h
index c6fa4cf..32d3935 100644
--- a/lib/cpp/test/concurrency/TimerManagerTests.h
+++ b/lib/cpp/test/concurrency/TimerManagerTests.h
@@ -34,8 +34,6 @@
class TimerManagerTests {
- static const double TEST_TOLERANCE;
-
public:
class Task : public Runnable {
public:
@@ -52,25 +50,11 @@
void run() {
_endTime = Util::currentTime();
-
- // Figure out error percentage
-
- int64_t delta = _endTime - _startTime;
-
- delta = delta > _timeout ? delta - _timeout : _timeout - delta;
-
- double error = double(delta) / _timeout;
-
- if (error < TEST_TOLERANCE) {
- _success = true;
- }
-
- _done = true;
-
- std::cout << "\t\t\tTimerManagerTests::Task[" << this << "] done" << std::endl; // debug
+ _success = (_endTime - _startTime) >= _timeout;
{
Synchronized s(_monitor);
+ _done = true;
_monitor.notifyAll();
}
}
@@ -147,7 +131,6 @@
Monitor _monitor;
};
-const double TimerManagerTests::TEST_TOLERANCE = .20;
}
}
}