2
\$\begingroup\$

I'm working on simulating a number of processing elements (nodes) which do some work in parallel and communicate by exchanging messages. The message exchange is not in the scope of this post, but it restricts the modelling of the parallel tasks because they are not completely independent and thus cannot run in any order. The computation a node does takes some time and may result in an error.

The simulation must detect if or when all computations are done. In case an error occurs the simulation has to cancel all remaining tasks and quit gracefully. While the simulation is waiting for the asynchronous operations to finish it has to do some work on it's own, namely dispatch messages between nodes.

In my implementation each asynchronous operation is started with a call to std::async, resulting in a std::future. Because the simulation's thread should not block for too long and the number of nodes may be large iterating over all futures and giving each some time to finish with future::wait_for seems like a bad idea. On the other hand iterating over the collection and just checking the result feels like busy waiting. So my idea is to have another std::future which is available when all tasks are done - with or without error.

So this "watcher" task will monitor the nodes' tasks. It could iterate over the collection of futures, calling get() on each one. But since the order of completion of the tasks is not known there may be (and in general will be) a significant delay between the occurrence and detection of an error.

My solution to this problem is as follows: I create a queue of tasks that are about to finish. A node's computation is wrapped in a function which notifies the watcher by using a callback in the case the computation is done or an error is caught. This callback inserts the node's ID into a synchronised queue and wakes up the watcher by calling notify_all() on a std::condition_variable. Since the asynchronous operation will not terminate until the callback returns notify_all() will be called before the respective std::future is ready. So there is no guarantee that the tasks finish in the order given by the queue, but in my scenario I'm not interested in the exact completion order and it should be close enough.

In case of an error the remaining operations need to be terminated. std::future does not offer such functionality. Luckily the nodes I want to simulate don't do time consuming operations which are difficult to cancel, but they will often wait for incoming messages on a condition_variable and so can be notified when task termination is requested.

For the sake of simplicity in the example code a node computes some math. This will take some time in which the node's computation thread will sleep. If the input is invalid, an exception is raised. If the watcher detects an exception, all remaining tasks are terminated and the exception is rethrown.

Here is a working example:

#include <iostream>
#include <cmath>
#include <exception>
#include <mutex>
#include <chrono>
#include <map>
#include <queue>
#include <future>
#include <random>

class Node {
public:
  Node(int id, double value, std::chrono::milliseconds delay)
    : _id(id), _value(value), _delay(delay), _terminate(false) {

  }

  int getId() const {
    return _id;
  }

  double getValue() const {
    return _value;
  }

  double sqrt() const {
    // We just use this so we can wake this thread up conviniently.
    std::mutex mutex;
    std::unique_lock<std::mutex> lock(mutex);
    _wakeup.wait_for(lock, _delay, [&]() {
      return _terminate;
    });

    // Check if terminate() was called while we waited.
    if(_terminate) {
      throw std::runtime_error("Thread terminated");
    }

    if(_value < 0) {
      auto message = std::string("Node ") + std::to_string(_id) +
        std::string(" doesn't know how to compute sqrt of negative value");
      throw std::out_of_range(message);
    }

    return std::sqrt(_value);
  }

  void terminate() {
    _terminate = true;
    _wakeup.notify_all();
  }

private:
  int _id;
  double _value;
  std::chrono::milliseconds _delay;
  mutable std::condition_variable _wakeup;
  bool _terminate;
};

static double nodeThreadFunc(Node* node, std::function<void(int)> onEvent) {
  try {
    auto result = node->sqrt();
    onEvent(node->getId());
    return result;
  } catch(...) {
    // Catch all exceptions so we can notify the parent thread before rethrowing.
    onEvent(node->getId());
    throw;
  }
}

static std::chrono::milliseconds watchThreadFunc(const std::map<int, std::unique_ptr<Node>>* nodes) {
  auto start = std::chrono::high_resolution_clock::now();

  std::mutex futuresMutex;
  std::map<int, std::future<double>> futures;
  std::queue<int> completionOrder;
  std::condition_variable electionEvent;

  const auto& onEvent = [&futuresMutex, &completionOrder, &electionEvent](int nodeId) {
    // The async operation of the node with the given id is about to finish.
    // Since this function will be called from within the node's thread, the
    // actual work is done. However, the node's thread cannot terminate until
    // we return from here.
    std::unique_lock<std::mutex> ulock(futuresMutex);
    completionOrder.push(nodeId);
    ulock.unlock();
    electionEvent.notify_all();
  };

  for(auto& it : *nodes) {
    // Launch all async operations, one for each node.
    futures.emplace(it.second->getId(), std::async(std::launch::async, &nodeThreadFunc,
      it.second.get(), onEvent));
  }

  std::cout << "Waiting for " << futures.size()
    << " async operations to finish." << std::endl;

  try {
    std::unique_lock<std::mutex> ulock(futuresMutex);
    while(!futures.empty()) {
      // Wait for an incoming event so we know a result will be ready soon.
      while(completionOrder.empty()) {
        electionEvent.wait(ulock, [&]() {
          return !completionOrder.empty();
        });
      }

      // Find the node which will have it's result ready (soon).
      auto nodeId = completionOrder.front();
      auto it = futures.find(nodeId);

      // Get the node's result. This call could still block, because the notification
      // we got happens before the node's thread can terminate.
      auto result = it->second.get();

      // We are still under the lock, so the result messages won't interleave.
      auto& node = *nodes->at(nodeId);
      std::cout << "Node " << node.getId() << " says: sqrt("
        << node.getValue() << ") = " << result << std::endl;

      // Remove the node from the open lists so we know it has been processed.
      completionOrder.pop();
      futures.erase(it);
    }

    auto end = std::chrono::high_resolution_clock::now();
    return std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
  } catch(...) {
    // At least one node's computation failed. Before we can rethrow the exception to
    // notify our parent thread we have to make sure all node threads terminate. We do
    // this by calling Node::terminate() which cancels all waiting operations.
    for(auto& it : *nodes) {
      it.second->terminate();
    }

    // Also for this function to be ablt to return all futures must be "closed" by
    // calling future::get(). Otherwise the futures' scope cannot be left and the
    // main thread never notices that an error occurred.
    for(auto& it : futures) {
      auto& future = it.second;
      if(!future.valid()) {
        // This is the case for completed futures. Here: the future of the node which
        // caused the exception.
        continue;
      }
      std::cout << "Joining async operation for node " << it.first << " ... ";
      try {
        it.second.get();
        std::cout << "terminated normally" << std::endl;
      } catch(std::exception& ex) {
        std::cout << "terminated with exception: " << ex.what() << std::endl;
      } catch(...) {
        std::cout << "terminated with unexpected exception" << std::endl;
      }
    }

    // Propagate first exception to parent thread.
    throw;
  }
}

static void simulate() {
  // Prepare the random numbers.
  auto now = std::chrono::high_resolution_clock::now();
  std::mt19937 rng(now.time_since_epoch().count());
  std::uniform_real_distribution<double> values(-1,100);
  std::uniform_int_distribution<int> delays(100,1000);

  // Create some nodes.
  std::map<int, std::unique_ptr<Node>> nodes;
  for(int i = 0; i < 30; ++i) {
    nodes.emplace(i, std::unique_ptr<Node>(new Node(i, values(rng),
      std::chrono::milliseconds(delays(rng)))));
  }

  // Start async operations through the watcher thread.
  auto compute = std::async(std::launch::async, &watchThreadFunc, &nodes);
  while(true) {
    auto asyncState = compute.wait_for(std::chrono::milliseconds(1));
    if(asyncState == std::future_status::ready) {
      auto result = compute.get();
      std::cout << "Computation using " << nodes.size() << " nodes took "
        << result.count() << " ms" << std::endl;
      break;
    }

    // Do some other work.
  }
}

int main() {
  try {
    simulate();
  } catch(std::exception& ex) {
    std::cout << "Simulation failed: " << ex.what() << std::endl;
  }

  return 0;
}

I have the following questions:

  1. Is there a less complex way of handling multiple asynchronous operations using builtin C++11 functionality with early detection of exceptions? Using an extra task, a synchronized queue and notification callbacks seems a lot of work, but I couldn't think of a better way.
  2. Is it OK that Node::_terminate is not synchronized when it is written in Node::terminate()?
  3. Is there some other way than try-catching a call to std::future::get() to check if an exception occurred while executing the operation?
  4. Why do I have to call std::future::get() to terminate the task even if it has long finished? If I don't the scope where the futures are created cannot be left.

Remarks and tips regarding overall code quality are welcome as well. In production code I would split watchThreadFunc into smaller parts resp. encapsulate it's functionality in a class.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.