Async Utils (#212)

* Add async_utils for easier multithreading

* Joseph's comments

* Consolidate async tests

Author: akleeman

include/albatross/src/utils/async_utils.hpp

@@ -0,0 +1,67 @@
+/*
+ * Copyright (C) 2020 Swift Navigation Inc.
+ * Contact: Swift Navigation <dev@swiftnav.com>
+ *
+ * This source is subject to the license found in the file 'LICENSE' which must
+ * be distributed together with this source. All other rights reserved.
+ *
+ * THIS CODE AND INFORMATION IS PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND,
+ * EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE IMPLIED
+ * WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A PARTICULAR PURPOSE.
+ */
+
+#ifndef INCLUDE_ALBATROSS_SRC_UTILS_ASYNC_UTILS_HPP_
+#define INCLUDE_ALBATROSS_SRC_UTILS_ASYNC_UTILS_HPP_
+
+namespace albatross {
+
+// This method makes sure we don't accidentally call async with the
+// default mode which has some flaws:
+//
+// https://eli.thegreenplace.net/2016/the-promises-and-challenges-of-stdasync-task-based-parallelism-in-c11/
+template <typename F, typename... Ts>
+inline auto async_safe(F &&f, Ts &&... params) {
+  return std::async(std::launch::async, std::forward<F>(f),
+                    std::forward<Ts>(params)...);
+}
+
+template <typename ValueType, typename ApplyFunction,
+          typename ApplyType = typename details::value_only_apply_result<
+              ApplyFunction, ValueType>::type,
+          typename std::enable_if<details::is_valid_value_only_apply_function<
+                                      ApplyFunction, ValueType>::value &&
+                                      std::is_same<void, ApplyType>::value,
+                                  int>::type = 0>
+void async_apply(const std::vector<ValueType> &xs, const ApplyFunction &f) {
+  std::vector<std::future<void>> futures;
+  for (const auto &x : xs) {
+    futures.emplace_back(async_safe(f, x));
+  }
+  for (auto &f : futures) {
+    f.get();
+  }
+}
+
+template <typename ValueType, typename ApplyFunction,
+          typename ApplyType = typename details::value_only_apply_result<
+              ApplyFunction, ValueType>::type,
+          typename std::enable_if<details::is_valid_value_only_apply_function<
+                                      ApplyFunction, ValueType>::value &&
+                                      !std::is_same<void, ApplyType>::value,
+                                  int>::type = 0>
+auto async_apply(const std::vector<ValueType> &xs, const ApplyFunction &f) {
+  std::vector<std::future<ApplyType>> futures;
+  for (const auto &x : xs) {
+    futures.emplace_back(async_safe(f, x));
+  }
+
+  std::vector<ApplyType> output;
+  for (auto &f : futures) {
+    output.emplace_back(f.get());
+  }
+  return output;
+}
+
+} // namespace albatross
+
+#endif /* INCLUDE_ALBATROSS_SRC_UTILS_ASYNC_UTILS_HPP_ */

include/albatross/utils/AsyncUtils

@@ -0,0 +1,20 @@
+/*
+ * Copyright (C) 2020 Swift Navigation Inc.
+ * Contact: Swift Navigation <dev@swiftnav.com>
+ *
+ * This source is subject to the license found in the file 'LICENSE' which must
+ * be distributed together with this source. All other rights reserved.
+ *
+ * THIS CODE AND INFORMATION IS PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND,
+ * EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE IMPLIED
+ * WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A PARTICULAR PURPOSE.
+ */
+
+#ifndef ALBATROSS_UTILS_ASYNC_UTILS_H
+#define ALBATROSS_UTILS_ASYNC_UTILS_H
+
+#include <future>
+
+#include "../src/utils/async_utils.hpp"
+
+#endif

tests/CMakeLists.txt

@@ -1,5 +1,6 @@
 add_executable(albatross_unit_tests 
   test_apply.cc
+  test_async_utils.cc
   test_block_utils.cc
   test_call_trace.cc
   test_callers.cc

tests/test_async_utils.cc

@@ -0,0 +1,78 @@
+/*
+ * Copyright (C) 2020 Swift Navigation Inc.
+ * Contact: Swift Navigation <dev@swiftnav.com>
+ *
+ * This source is subject to the license found in the file 'LICENSE' which must
+ * be distributed together with this source. All other rights reserved.
+ *
+ * THIS CODE AND INFORMATION IS PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND,
+ * EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE IMPLIED
+ * WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A PARTICULAR PURPOSE.
+ */
+
+#include <gtest/gtest.h>
+
+#include <albatross/Indexing>
+#include <albatross/utils/AsyncUtils>
+#include <albatross/utils/RandomUtils>
+
+#include <chrono>
+#include <mutex>
+#include <numeric>
+
+namespace albatross {
+
+TEST(test_async_utils, test_async_apply_with_capture) {
+  std::vector<int> xs = {0, 1, 2, 3, 4, 5};
+
+  std::mutex mu;
+  int sum = 0.;
+
+  std::vector<int> order_processed;
+
+  auto add_to_sum = [&](const int x) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(abs(x - 2)));
+
+    std::lock_guard<std::mutex> lock(mu);
+    sum += x;
+    order_processed.push_back(x);
+  };
+
+  async_apply(xs, add_to_sum);
+
+  EXPECT_EQ(sum, std::accumulate(xs.begin(), xs.end(), 0));
+  // Make sure the async apply was indeed processed out of order.
+  EXPECT_NE(order_processed, xs);
+}
+
+TEST(test_async_utils, test_async_is_faster) {
+
+  auto slow_process = [&](const int i) {
+    const auto start = std::chrono::system_clock::now();
+    std::chrono::seconds delay(1);
+    while (std::chrono::system_clock::now() - start < delay) {
+    };
+    return i;
+  };
+
+  std::vector<int> inds;
+  for (std::size_t i = 0; i < 4; ++i) {
+    inds.push_back(i);
+  }
+
+  const auto start = std::chrono::system_clock::now();
+  const auto actual = async_apply(inds, slow_process);
+  const auto end = std::chrono::system_clock::now();
+
+  EXPECT_LT(end - start, std::chrono::seconds(inds.size() - 1));
+
+  const auto start_direct = std::chrono::system_clock::now();
+  const auto expected = apply(inds, slow_process);
+  const auto end_direct = std::chrono::system_clock::now();
+
+  EXPECT_EQ(actual, expected);
+
+  EXPECT_GT(end_direct - start_direct, std::chrono::seconds(inds.size() - 1));
+}
+
+} // namespace albatross