# Monitor asynchronous tasks, tracking their running times

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 <future>
#include <chrono>
#include <vector>
#include <random>

typedef std::future<int> FutureResultInt;

{
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::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++)

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;
}


## 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;

{
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;


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: