Group by async_apply

include/albatross/Indexing

@@ -13,12 +13,14 @@
 #ifndef ALBATROSS_INDEXING_H
 #define ALBATROSS_INDEXING_H
 
+#include <future>
 #include "Dataset"
 
 #include <albatross/src/indexing/traits.hpp>
 #include <albatross/src/indexing/subset.hpp>
 #include <albatross/src/indexing/filter.hpp>
 #include <albatross/src/indexing/apply.hpp>
+#include <albatross/src/indexing/async_apply.hpp>
 #include <albatross/src/indexing/group_by.hpp>
 
 #endif

include/albatross/src/indexing/async_apply.hpp

@@ -0,0 +1,153 @@
+/*
+ * Copyright (C) 2020 Swift Navigation Inc.
+ * Contact: Swift Navigation <[email protected]>
+ *
+ * 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>
+inline void async_apply(const std::vector<ValueType> &xs, ApplyFunction &&f) {
+  std::vector<std::future<void>> futures;
+  for (const ValueType &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>
+inline auto async_apply(const std::vector<ValueType> &xs, ApplyFunction &&f) {
+  std::vector<std::future<ApplyType>> futures;
+  for (const ValueType &x : xs) {
+    futures.emplace_back(async_safe(f, x));
+  }
+
+  std::vector<ApplyType> output;
+  for (auto &f : futures) {
+    output.emplace_back(f.get());
+  }
+  return output;
+}
+
+// Map
+
+template <
+    template <typename...> class Map, typename KeyType, typename ValueType,
+    typename ApplyFunction,
+    typename ApplyType = typename details::key_value_apply_result<
+        ApplyFunction, KeyType, ValueType>::type,
+    typename std::enable_if<details::is_valid_key_value_apply_function<
+                                ApplyFunction, KeyType, ValueType>::value &&
+                                std::is_same<void, ApplyType>::value,
+                            int>::type = 0>
+inline void async_apply(const Map<KeyType, ValueType> &map, ApplyFunction &&f) {
+  std::vector<std::future<void>> futures;
+  for (const auto &pair : map) {
+    futures.emplace_back(async_safe(f, pair.first, pair.second));
+  }
+  for (auto &f : futures) {
+    f.get();
+  }
+}
+
+template <
+    template <typename...> class Map, typename KeyType, typename ValueType,
+    typename ApplyFunction,
+    typename ApplyType = typename details::key_value_apply_result<
+        ApplyFunction, KeyType, ValueType>::type,
+    typename std::enable_if<details::is_valid_key_value_apply_function<
+                                ApplyFunction, KeyType, ValueType>::value &&
+                                !std::is_same<void, ApplyType>::value,
+                            int>::type = 0>
+inline Grouped<KeyType, ApplyType>
+async_apply(const Map<KeyType, ValueType> &map, ApplyFunction &&f) {
+
+  std::map<KeyType, std::future<ApplyType>> futures;
+  for (const auto &pair : map) {
+    futures[pair.first] = async_safe(f, pair.first, pair.second);
+  }
+
+  Grouped<KeyType, ApplyType> output;
+  for (auto &pair : futures) {
+    output.emplace(pair.first, pair.second.get());
+  }
+  return output;
+}
+
+template <template <typename...> class Map, typename KeyType,
+          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>
+inline Grouped<KeyType, ApplyType>
+async_apply(const Map<KeyType, ValueType> &map, ApplyFunction &&f) {
+
+  std::map<KeyType, std::future<ApplyType>> futures;
+  for (const auto &pair : map) {
+    futures[pair.first] = async_safe(f, pair.second);
+  }
+
+  Grouped<KeyType, ApplyType> output;
+  for (auto &pair : futures) {
+    output.emplace(pair.first, pair.second.get());
+  }
+  return output;
+}
+
+template <template <typename...> class Map, typename KeyType,
+          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>
+inline void async_apply(const Map<KeyType, ValueType> &map, ApplyFunction &&f) {
+  std::vector<std::future<void>> futures;
+  for (const auto &pair : map) {
+    futures.emplace_back(async_safe(f, pair.second));
+  }
+  for (auto &f : futures) {
+    f.get();
+  }
+}
+
+} // namespace albatross
+
+#endif /* INCLUDE_ALBATROSS_SRC_UTILS_ASYNC_UTILS_HPP_ */

include/albatross/src/indexing/group_by.hpp

@@ -115,6 +115,10 @@ template <typename KeyType, typename ValueType> class GroupedBase {
     return albatross::apply(map_, std::forward<ApplyFunction>(f));
   }
 
+  template <typename ApplyFunction> auto async_apply(ApplyFunction &&f) const {
+    return albatross::async_apply(map_, std::forward<ApplyFunction>(f));
+  }
+
 protected:
   std::map<KeyType, ValueType> map_;
 };
@@ -188,20 +192,13 @@ class Grouped<KeyType, GroupIndices>
   using Base = GroupedBase<KeyType, GroupIndices>;
   using Base::Base;
 
-  template <typename ApplyFunction,
-            typename ApplyType = typename details::key_value_apply_result<
-                ApplyFunction, KeyType, GroupIndices>::type,
-            typename std::enable_if<
-                details::is_valid_index_apply_function<ApplyFunction, KeyType,
-                                                       GroupIndices>::value &&
-                    !std::is_same<void, ApplyType>::value,
-                int>::type = 0>
-  auto index_apply(const ApplyFunction &f) const {
-    Grouped<KeyType, ApplyType> output;
-    for (const auto &pair : this->map_) {
-      output.emplace(pair.first, f(pair.first, pair.second));
-    }
-    return output;
+  template <typename ApplyFunction> auto index_apply(ApplyFunction &&f) const {
+    return apply(this->map_, std::forward<ApplyFunction>(f));
+  }
+
+  template <typename ApplyFunction>
+  auto async_index_apply(ApplyFunction &&f) const {
+    return async_apply(this->map_, std::forward<ApplyFunction>(f));
   }
 };
 
@@ -401,8 +398,12 @@ template <typename Derived> class GroupByBase {
 
   std::size_t size() const { return indexers().size(); }
 
-  template <typename ApplyFunction> auto apply(const ApplyFunction &f) const {
-    return groups().apply(f);
+  template <typename ApplyFunction> auto apply(ApplyFunction &&f) const {
+    return groups().apply(std::forward<ApplyFunction>(f));
+  }
+
+  template <typename ApplyFunction> auto async_apply(ApplyFunction &&f) const {
+    return groups().async_apply(std::forward<ApplyFunction>(f));
   }
 
   ValueType get_group(const KeyType &key) const {
@@ -415,38 +416,17 @@ template <typename Derived> class GroupByBase {
                           albatross::subset(parent_, first_indexer.second));
   }
 
-  template <typename ApplyFunction,
-            typename ApplyType = typename details::key_value_apply_result<
-                ApplyFunction, KeyType, GroupIndices>::type,
-            typename std::enable_if<
-                details::is_valid_index_apply_function<ApplyFunction, KeyType,
-                                                       GroupIndices>::value &&
-                    !std::is_same<void, ApplyType>::value,
-                int>::type = 0>
-  auto index_apply(const ApplyFunction &f) const {
-    Grouped<KeyType, ApplyType> output;
-    for (const auto &pair : indexers()) {
-      output.emplace(pair.first, f(pair.first, pair.second));
-    }
-    return output;
+  template <typename ApplyFunction> auto index_apply(ApplyFunction &&f) const {
+    return albatross::apply(indexers(), std::forward<ApplyFunction>(f));
   }
 
-  template <typename ApplyFunction,
-            typename ApplyType = typename details::key_value_apply_result<
-                ApplyFunction, KeyType, GroupIndices>::type,
-            typename std::enable_if<
-                details::is_valid_index_apply_function<ApplyFunction, KeyType,
-                                                       GroupIndices>::value &&
-                    std::is_same<void, ApplyType>::value,
-                int>::type = 0>
-  void index_apply(const ApplyFunction &f) const {
-    for (const auto &pair : indexers()) {
-      f(pair.first, pair.second);
-    }
+  template <typename ApplyFunction>
+  auto async_index_apply(ApplyFunction &&f) const {
+    return albatross::async_apply(indexers(), std::forward<ApplyFunction>(f));
   }
 
-  template <typename FilterFunction> auto filter(FilterFunction f) const {
-    return groups().filter(f);
+  template <typename FilterFunction> auto filter(FilterFunction &&f) const {
+    return groups().filter(std::forward<FilterFunction>(f));
   }
 
   Grouped<KeyType, std::size_t> counts() const {

tests/CMakeLists.txt

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

tests/test_async_utils.cc → tests/test_async_apply.cc

@@ -13,7 +13,6 @@
 #include <gtest/gtest.h>
 
 #include <albatross/Indexing>
-#include <albatross/utils/AsyncUtils>
 #include <albatross/utils/RandomUtils>
 
 #include <chrono>