From 0ebf498c42a422d96aac6438dd6904d9cdc96604 Mon Sep 17 00:00:00 2001
From: kleeman <akleeman@gmail.com>
Date: Fri, 27 Mar 2020 15:00:33 -0700
Subject: [PATCH 1/4] Add async_apply for map-like objects

---
 include/albatross/Indexing                  |  1 +
 include/albatross/src/indexing/group_by.hpp | 68 ++++++---------
 include/albatross/src/utils/async_utils.hpp | 92 ++++++++++++++++++++-
 tests/test_group_by.cc                      | 19 ++++-
 4 files changed, 132 insertions(+), 48 deletions(-)

diff --git a/include/albatross/Indexing b/include/albatross/Indexing
index 9563f424..f37c94a3 100644
--- a/include/albatross/Indexing
+++ b/include/albatross/Indexing
@@ -19,6 +19,7 @@
 #include <albatross/src/indexing/subset.hpp>
 #include <albatross/src/indexing/filter.hpp>
 #include <albatross/src/indexing/apply.hpp>
+#include "./utils/AsyncUtils"
 #include <albatross/src/indexing/group_by.hpp>
 
 #endif
diff --git a/include/albatross/src/indexing/group_by.hpp b/include/albatross/src/indexing/group_by.hpp
index 313b87c0..123345f0 100644
--- a/include/albatross/src/indexing/group_by.hpp
+++ b/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 {
diff --git a/include/albatross/src/utils/async_utils.hpp b/include/albatross/src/utils/async_utils.hpp
index 9f2fc40c..4c70b48b 100644
--- a/include/albatross/src/utils/async_utils.hpp
+++ b/include/albatross/src/utils/async_utils.hpp
@@ -32,7 +32,8 @@ template <typename ValueType, typename ApplyFunction,
                                       ApplyFunction, ValueType>::value &&
                                       std::is_same<void, ApplyType>::value,
                                   int>::type = 0>
-void async_apply(const std::vector<ValueType> &xs, const ApplyFunction &f) {
+inline void async_apply(const std::vector<ValueType> &xs,
+                        ApplyFunction &&f) {
   std::vector<std::future<void>> futures;
   for (const auto &x : xs) {
     futures.emplace_back(async_safe(f, x));
@@ -49,7 +50,8 @@ template <typename ValueType, typename ApplyFunction,
                                       ApplyFunction, ValueType>::value &&
                                       !std::is_same<void, ApplyType>::value,
                                   int>::type = 0>
-auto async_apply(const std::vector<ValueType> &xs, const ApplyFunction &f) {
+inline auto async_apply(const std::vector<ValueType> &xs,
+                        ApplyFunction &&f) {
   std::vector<std::future<ApplyType>> futures;
   for (const auto &x : xs) {
     futures.emplace_back(async_safe(f, x));
@@ -62,6 +64,92 @@ auto async_apply(const std::vector<ValueType> &xs, const ApplyFunction &f) {
   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_ */
diff --git a/tests/test_group_by.cc b/tests/test_group_by.cc
index a950f4ef..1b2e304e 100644
--- a/tests/test_group_by.cc
+++ b/tests/test_group_by.cc
@@ -283,6 +283,11 @@ TYPED_TEST_P(GroupByTester, test_groupby_apply_combine) {
 
   const auto combined = grouped.apply(only_keep_one).combine();
 
+  const auto async_combined = grouped.async_apply(only_keep_one).combine();
+
+  // Async an normal apply should match
+  EXPECT_EQ(combined, async_combined);
+
   // Same number of final combined elements as there are groups.
   EXPECT_EQ(grouped.size(), combined.size());
 }
@@ -296,7 +301,11 @@ TYPED_TEST_P(GroupByTester, test_groupby_apply_void) {
   const auto increment_count = [&](const auto &, const auto &) { ++count; };
 
   grouped.apply(increment_count);
+  EXPECT_EQ(grouped.size(), count);
 
+  // Again but with async
+  count = 0;
+  grouped.async_apply(increment_count);
   EXPECT_EQ(grouped.size(), count);
 }
 
@@ -305,11 +314,14 @@ TYPED_TEST_P(GroupByTester, test_groupby_apply_value_only) {
   const auto grouped = group_by(parent, this->test_case.get_grouper());
 
   std::size_t count = 0;
-
   const auto increment_count = [&](const auto &) { ++count; };
 
   grouped.apply(increment_count);
+  EXPECT_EQ(grouped.size(), count);
 
+  // Again but with async
+  count = 0;
+  grouped.async_apply(increment_count);
   EXPECT_EQ(grouped.size(), count);
 }
 
@@ -318,13 +330,16 @@ TYPED_TEST_P(GroupByTester, test_groupby_index_apply) {
   const auto grouped = group_by(parent, this->test_case.get_grouper());
 
   std::size_t count = 0;
-
   const auto increment_count = [&](const auto &, const GroupIndices &) {
     ++count;
   };
 
   grouped.index_apply(increment_count);
+  EXPECT_EQ(grouped.size(), count);
 
+  // Again but with async
+  count = 0;
+  grouped.async_index_apply(increment_count);
   EXPECT_EQ(grouped.size(), count);
 }
 

From 20a2366224e0ce9b0bae579c996f2afd13fc7e79 Mon Sep 17 00:00:00 2001
From: kleeman <akleeman@gmail.com>
Date: Fri, 27 Mar 2020 15:02:36 -0700
Subject: [PATCH 2/4] Move async utils into indexing

---
 include/albatross/Indexing                    |  3 ++-
 .../async_apply.hpp}                          |  0
 include/albatross/utils/AsyncUtils            | 20 -------------------
 3 files changed, 2 insertions(+), 21 deletions(-)
 rename include/albatross/src/{utils/async_utils.hpp => indexing/async_apply.hpp} (100%)
 delete mode 100644 include/albatross/utils/AsyncUtils

diff --git a/include/albatross/Indexing b/include/albatross/Indexing
index f37c94a3..dbd3f0f9 100644
--- a/include/albatross/Indexing
+++ b/include/albatross/Indexing
@@ -13,13 +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 "./utils/AsyncUtils"
+#include <albatross/src/indexing/async_apply.hpp>
 #include <albatross/src/indexing/group_by.hpp>
 
 #endif
diff --git a/include/albatross/src/utils/async_utils.hpp b/include/albatross/src/indexing/async_apply.hpp
similarity index 100%
rename from include/albatross/src/utils/async_utils.hpp
rename to include/albatross/src/indexing/async_apply.hpp
diff --git a/include/albatross/utils/AsyncUtils b/include/albatross/utils/AsyncUtils
deleted file mode 100644
index 436fdb63..00000000
--- a/include/albatross/utils/AsyncUtils
+++ /dev/null
@@ -1,20 +0,0 @@
-/*
- * 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

From 8ec8b0da634ad845d092fec03ee4e5362bae97f8 Mon Sep 17 00:00:00 2001
From: kleeman <akleeman@gmail.com>
Date: Fri, 27 Mar 2020 15:06:46 -0700
Subject: [PATCH 3/4] Clang

---
 include/albatross/src/indexing/async_apply.hpp | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/include/albatross/src/indexing/async_apply.hpp b/include/albatross/src/indexing/async_apply.hpp
index 4c70b48b..740e6dfd 100644
--- a/include/albatross/src/indexing/async_apply.hpp
+++ b/include/albatross/src/indexing/async_apply.hpp
@@ -32,8 +32,7 @@ template <typename ValueType, typename ApplyFunction,
                                       ApplyFunction, ValueType>::value &&
                                       std::is_same<void, ApplyType>::value,
                                   int>::type = 0>
-inline void async_apply(const std::vector<ValueType> &xs,
-                        ApplyFunction &&f) {
+inline void async_apply(const std::vector<ValueType> &xs, ApplyFunction &&f) {
   std::vector<std::future<void>> futures;
   for (const auto &x : xs) {
     futures.emplace_back(async_safe(f, x));
@@ -50,8 +49,7 @@ template <typename ValueType, typename ApplyFunction,
                                       ApplyFunction, ValueType>::value &&
                                       !std::is_same<void, ApplyType>::value,
                                   int>::type = 0>
-inline auto async_apply(const std::vector<ValueType> &xs,
-                        ApplyFunction &&f) {
+inline auto async_apply(const std::vector<ValueType> &xs, ApplyFunction &&f) {
   std::vector<std::future<ApplyType>> futures;
   for (const auto &x : xs) {
     futures.emplace_back(async_safe(f, x));

From 978fcf776503ff1b30d532ded9839dfd0fa2ec4a Mon Sep 17 00:00:00 2001
From: kleeman <akleeman@gmail.com>
Date: Fri, 27 Mar 2020 17:40:11 -0700
Subject: [PATCH 4/4] Explicitly use value type in apply

---
 include/albatross/src/indexing/async_apply.hpp     | 4 ++--
 tests/CMakeLists.txt                               | 2 +-
 tests/{test_async_utils.cc => test_async_apply.cc} | 1 -
 3 files changed, 3 insertions(+), 4 deletions(-)
 rename tests/{test_async_utils.cc => test_async_apply.cc} (98%)

diff --git a/include/albatross/src/indexing/async_apply.hpp b/include/albatross/src/indexing/async_apply.hpp
index 740e6dfd..f3df10b9 100644
--- a/include/albatross/src/indexing/async_apply.hpp
+++ b/include/albatross/src/indexing/async_apply.hpp
@@ -34,7 +34,7 @@ template <typename ValueType, typename ApplyFunction,
                                   int>::type = 0>
 inline void async_apply(const std::vector<ValueType> &xs, ApplyFunction &&f) {
   std::vector<std::future<void>> futures;
-  for (const auto &x : xs) {
+  for (const ValueType &x : xs) {
     futures.emplace_back(async_safe(f, x));
   }
   for (auto &f : futures) {
@@ -51,7 +51,7 @@ template <typename ValueType, typename ApplyFunction,
                                   int>::type = 0>
 inline auto async_apply(const std::vector<ValueType> &xs, ApplyFunction &&f) {
   std::vector<std::future<ApplyType>> futures;
-  for (const auto &x : xs) {
+  for (const ValueType &x : xs) {
     futures.emplace_back(async_safe(f, x));
   }
 
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 53e9fade..5247cc8a 100644
--- a/tests/CMakeLists.txt
+++ b/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
diff --git a/tests/test_async_utils.cc b/tests/test_async_apply.cc
similarity index 98%
rename from tests/test_async_utils.cc
rename to tests/test_async_apply.cc
index 0ea2843a..a8f9adf3 100644
--- a/tests/test_async_utils.cc
+++ b/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>