Monitor asynchronous tasks, tracking their running times

Question

Scenario: I'm getting requests through a (thread-safe) queue. Each request then needs to be handled in a separate thread. There is a chance that the function (which is actually calling a Java-program via popen, and polling its output) takes a very long time. From the main thread I need a mechanism to indicate that situation (basically, measuring thread-running-time).

In my example I'm trying to 'enrich' std::future with some time information. The sample runs seamlessly - but I'm uncertain if this is the correct way.

Here is a very simple demo that mimics what I'm trying to achieve (where ThreadFunc stands in for my real processing code):

#include <iostream>
#include <thread>
#include <future>
#include <chrono>
#include <vector>
#include <random>
typedef std::future<int> FutureResultInt;
int ThreadFunc( )
{
 std::random_device rd;
 std::mt19937 mt(rd());
 const int iRand = std::uniform_int_distribution<int>(2000, 6000)(mt);
 std::cout << "ThreadFunc waiting for [" << iRand << "] ms ... " << std::endl;
 std::this_thread::sleep_for(std::chrono::milliseconds(iRand));
 std::cout << "ThreadFunc [" << iRand << "] done" << std::endl;
 return iRand;
}
class CFutureTest
{
public:
 CFutureTest() = delete;
 CFutureTest(FutureResultInt&& fr)
 : m_start(std::chrono::system_clock::now())
 , m_result()
 {
 m_result = std::move(fr);
 };
 int GetAge() const
 {
 return std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now() - m_start).count();
 }
// private:
 FutureResultInt m_result;
 std::chrono::time_point<std::chrono::system_clock> m_start; 
};
int main()
{
 std::vector< CFutureTest > futures;
 for (int i = 0; i < 5; i++)
 futures.push_back(std::move(std::async(std::launch::async, ThreadFunc)));
 while (futures.size() > 0)
 {
 for (std::vector< CFutureTest >::iterator it = futures.begin(); it != futures.end(); ++it)
 {
 CFutureTest& future = *it;
 const std::future_status stat = future.m_result.wait_for(std::chrono::milliseconds(1));
 switch (stat)
 {
 case std::future_status::timeout:
 if (future.GetAge() > 4000)
 {
 std::cout << "Thread has exceeded the time limit" << std::endl;
 }
 continue;
 case std::future_status::deferred:
 std::cout << "std::future_status::deferred" << std::endl;
 continue;
 }
 const int iResult = future.m_result.get();
 std::cout << "future returned [" << iResult << "] (removing!)" << std::endl;
 futures.erase(it);
 if (futures.size() < 1)
 break;
 it = futures.begin();
 }
 }
 return 0;
}

• Mast: Cross-posted on Stack Overflow

Answer 1

Ugly typedef

I'm not a big fan of this:

typedef std::future<int> FutureResultInt;

It's not significantly shorter or easier to read, it doesn't isolate the user from an underlying type, and it only serves to slow down my reading every time I hit it. That's somewhat subjective, of course, but I don't believe it adds value. I'd be happier with it if it told us what it's for, rather than merely what it is. I'm thinking of something like:

using TimerTaskResult = std::future<int>;

Now the name conveys something that's not just duplicating the standard name.

---

Member initialization

My compiler warns me that the initializer list is in a misleading order:

203856.cpp:47:54: warning: ‘CFutureTest::m_start’ will be initialized after [-Wreorder]
203856.cpp:46:19: warning: ‘FutureResultInt CFutureTest::m_result’ [-Wreorder]
203856.cpp:33:3: warning: when initialized here [-Wreorder]
 CFutureTest(FutureResultInt&& fr)
 ^~~~~~~~~~~

I'd also recommend value-initializing m_result rather than default-constructing followed by move-assigning:

 CFutureTest(FutureResultInt&& fr)
 : m_result{std::move(fr)}
 , m_start{std::chrono::system_clock::now()}
 {
 }

---

Missing case

203856.cpp: In function ‘int main()’:
203856.cpp:63:14: warning: enumeration value ‘ready’ not handled in switch [-Wswitch]
 switch (stat)
 ^

I like to have this warning enabled. We could provide a minimal case std::​future_status::​ready: break;, or we could unify the control flow by changing the other branches from continue to break and inlining the following code into the ready case like this:

 switch (stat) {
 case std::future_status::timeout:
 if (future.GetAge() > 4000) {
 std::cout << "Thread has exceeded the time limit" << std::endl;
 }
 break;
 case std::future_status::deferred:
 std::cout << "std::future_status::deferred" << std::endl;
 break;
 case std::future_status::ready:
 {
 const int iResult = future.m_result.get();
 std::cout << "future returned [" << iResult << "] (removing!)" << std::endl;
 futures.erase(it);
 if (futures.empty()) {
 return 0;
 } else {
 it = futures.begin();
 }
 }
 }

Instead of returning to futures.begin() when we harvest a result, it's arguably better to keep going from the next element if there is one:

 it = futures.erase(it);
 if (it == futures.end()) {
 if (futures.empty()) {
 return 0;
 } else {
 it = futures.begin();
 }
 }

---

Interleaved output

I know it's not part of your production code, but it's quite irritating to have the different threads' output interrupting each other mid-line. We can provide a class to hold a lock while several items are written to a stream:

class LogStream
{
 static std::mutex mutex;
 std::lock_guard<std::mutex> guard{mutex};
 std::ostream& stream{std::clog};
public:
 LogStream() {}
 template<typename T>
 std::ostream& operator<<(T&& t) { return stream << std::forward<T>(t); }
};
std::mutex LogStream::mutex;

And use it like this:

LogStream() << "ThreadFunc waiting for [" << iRand << "] ms ... " << std::endl;

---

Be clear about units

I don't like this function name:

int GetAge() const

Perhaps change it to getAgeMillis(), or how about making it a template method? Like this:

template<typename TimeUnit = std::chrono::seconds>
int GetAge() const
{
 auto now = std::chrono::system_clock::now();
 return std::chrono::duration_cast<TimeUnit>(now - m_start).count();
}

That also helps us deal with the very long lines that are a perennial problem with duration_cast.

---

Avoid polling

This is the big one. It's very power-inefficient to poll every 1ms to see whether you have any results; that's a bad thing on battery-powered systems, but it's just as important in a server farm.

What we need to do instead is have each finishing thread notify a condition variable. When we're woken, we then examine all the futures, picking up any where is_ready().

Unfortunately, that means we can't use std::async(), as its future isn't made ready until the called function has returned. We'll need to implement our own version that can accept a std::condition_variable& to be notified after it sets the future's value.

You might be able to pick up some advice from Stack Overflow - here's a couple of starting points that might have something relevant:

• Signaling main thread when std::future is ready to be retrieved
• How to implement std::when_any without polling?