Adding Combination Operators Zip, CombineLatest, WithLatestFrom

cpp/mrc/include/mrc/channel/status.hpp

@@ -17,6 +17,8 @@
 
 #pragma once
 
+#include <ostream>
+
 namespace mrc::channel {
 
 enum class Status
@@ -29,4 +31,25 @@ enum class Status
     error
 };
 
+static inline std::ostream& operator<<(std::ostream& os, const Status& s)
+{
+    switch (s)
+    {
+    case Status::success:
+        return os << "success";
+    case Status::empty:
+        return os << "empty";
+    case Status::full:
+        return os << "full";
+    case Status::closed:
+        return os << "closed";
+    case Status::timeout:
+        return os << "timeout";
+    case Status::error:
+        return os << "error";
+    default:
+        throw std::logic_error("Unsupported channel::Status enum. Was a new value added recently?");
+    }
 }
+
+}  // namespace mrc::channel

cpp/mrc/include/mrc/core/utils.hpp

@@ -19,6 +19,8 @@
 
 #include "mrc/utils/macros.hpp"
 
+#include <glog/logging.h>
+
 #include <algorithm>
 #include <exception>
 #include <map>
@@ -54,47 +56,47 @@ std::set<KeyT> extract_keys(const std::map<KeyT, ValT>& stdmap)
 }
 
 // RAII will execute a function when destroyed.
-template <typename FunctionT>
 class Unwinder
 {
   public:
+    explicit Unwinder(std::function<void()> unwind_fn) : m_unwind_fn(std::move(unwind_fn)) {}
+
     ~Unwinder()
     {
-        if (!!m_function)
+        if (!!m_unwind_fn)
         {
             try
             {
-                (*m_function)();
+                m_unwind_fn();
             } catch (...)
             {
+                LOG(ERROR) << "Fatal error during unwinder function";
                 std::terminate();
             }
         }
     }
 
-    explicit Unwinder(FunctionT* function_arg) : m_function(function_arg) {}
-
     void detach()
     {
-        m_function = nullptr;
+        m_unwind_fn = nullptr;
     }
 
     Unwinder()                           = delete;
     Unwinder(const Unwinder&)            = delete;
     Unwinder& operator=(const Unwinder&) = delete;
 
+    static Unwinder create(std::function<void()> unwind_fn)
+    {
+        return Unwinder(std::move(unwind_fn));
+    }
+
   private:
-    FunctionT* m_function;
+    std::function<void()> m_unwind_fn;
 };
 
-#define MRC_UNWIND(var_name, function) MRC_UNWIND_EXPLICIT(uw_func_##var_name, var_name, function)
-
-#define MRC_UNWIND_AUTO(function) \
-    MRC_UNWIND_EXPLICIT(MRC_UNIQUE_VAR_NAME(uw_func_), MRC_UNIQUE_VAR_NAME(un_obj_), function)
+#define MRC_UNWIND(unwinder_name, function) mrc::Unwinder unwinder_name(function);
 
-#define MRC_UNWIND_EXPLICIT(function_name, unwinder_name, function) \
-    auto function_name = (function);                                \
-    mrc::Unwinder<decltype(function_name)> unwinder_name(std::addressof(function_name))
+#define MRC_UNWIND_AUTO(function) MRC_UNWIND(MRC_UNIQUE_VAR_NAME(__un_obj_), function)
 
 template <typename T>
 std::pair<std::set<T>, std::set<T>> set_compare(const std::set<T>& cur_set, const std::set<T>& new_set)

cpp/mrc/include/mrc/edge/edge_channel.hpp

@@ -20,6 +20,7 @@
 #include "mrc/edge/edge_readable.hpp"
 #include "mrc/edge/edge_writable.hpp"
 #include "mrc/edge/forward.hpp"
+#include "mrc/utils/macros.hpp"
 
 #include <memory>
 
@@ -89,6 +90,24 @@ class EdgeChannel
     {
         CHECK(m_channel) << "Cannot create an EdgeChannel from an empty pointer";
     }
+
+    EdgeChannel(EdgeChannel&& other) : m_channel(std::move(other.m_channel)) {}
+
+    EdgeChannel& operator=(EdgeChannel&& other)
+    {
+        if (this == &other)
+        {
+            return *this;
+        }
+
+        m_channel = std::move(other.m_channel);
+
+        return *this;
+    }
+
+    // This should not be copyable because it requires passing in a unique_ptr
+    DELETE_COPYABILITY(EdgeChannel);
+
     virtual ~EdgeChannel() = default;
 
     [[nodiscard]] std::shared_ptr<EdgeChannelReader<T>> get_reader() const

cpp/mrc/include/mrc/node/operators/combine_latest.hpp

@@ -20,6 +20,7 @@
 #include "mrc/channel/status.hpp"
 #include "mrc/node/sink_properties.hpp"
 #include "mrc/node/source_properties.hpp"
+#include "mrc/utils/tuple_utils.hpp"
 #include "mrc/utils/type_utils.hpp"
 
 #include <boost/fiber/mutex.hpp>
@@ -31,91 +32,6 @@
 
 namespace mrc::node {
 
-// template <typename... TypesT>
-// class ParameterPackIndexer
-// {
-//   public:
-//     ParameterPackIndexer(TypesT... ts) : ParameterPackIndexer(std::make_index_sequence<sizeof...(TypesT)>{}, ts...)
-//     {}
-
-//     std::tuple<std::tuple<TypesT, std::size_t>...> tup;
-
-//   private:
-//     template <std::size_t... Is>
-//     ParameterPackIndexer(std::index_sequence<Is...> const& /*unused*/, TypesT... ts) : tup{std::make_tuple(ts,
-//     Is)...}
-//     {}
-// };
-
-// template <typename TargetT, typename ListHeadT, typename... ListTailsT>
-// constexpr size_t getTypeIndexInTemplateList()
-// {
-//     if constexpr (std::is_same<TargetT, ListHeadT>::value)
-//     {
-//         return 0;
-//     }
-//     else
-//     {
-//         return 1 + getTypeIndexInTemplateList<TargetT, ListTailsT...>();
-//     }
-// }
-
-namespace detail {
-struct Surely
-{
-    template <class... T>
-    auto operator()(const T&... t) const -> decltype(std::make_tuple(t.value()...))
-    {
-        return std::make_tuple(t.value()...);
-    }
-};
-}  // namespace detail
-
-// template <class... T>
-// inline auto surely(const std::tuple<T...>& tpl) -> decltype(rxcpp::util::apply(tpl, detail::surely()))
-// {
-//     return rxcpp::util::apply(tpl, detail::surely());
-// }
-
-template <class... T>
-inline auto surely2(const std::tuple<T...>& tpl)
-{
-    return std::apply([](auto... args) {
-        return std::make_tuple(args.value()...);
-    });
-}
-
-// template <typename... TypesT>
-// static auto surely2(const std::tuple<TypesT...>& tpl, std::index_sequence<Is...>)
-// {
-//     return std::make_tuple(std::make_shared<Upstream<Is>>(*self)...);
-// }
-
-// template <size_t i, typename T>
-// struct IndexTypePair
-// {
-//     static constexpr size_t index{i};
-//     using Type = T;
-// };
-
-// template <typename... T>
-// struct make_index_type_tuple_helper
-// {
-//     template <typename V>
-//     struct idx;
-
-//     template <size_t... Indices>
-//     struct idx<std::index_sequence<Indices...>>
-//     {
-//         using tuple_type = std::tuple<IndexTypePair<Indices, T>...>;
-//     };
-
-//     using tuple_type = typename idx<std::make_index_sequence<sizeof...(T)>>::tuple_type;
-// };
-
-// template <typename... T>
-// using make_index_type_tuple = typename make_index_type_tuple_helper<T...>::tuple_type;
-
 template <typename... TypesT>
 class CombineLatest : public WritableAcceptor<std::tuple<TypesT...>>
 {
@@ -128,9 +44,7 @@ class CombineLatest : public WritableAcceptor<std::tuple<TypesT...>>
   public:
     CombineLatest() :
       m_upstream_holders(build_ingress(const_cast<CombineLatest*>(this), std::index_sequence_for<TypesT...>{}))
-    {
-        // auto a = build_ingress(const_cast<CombineLatest*>(this), std::index_sequence_for<TypesT...>{});
-    }
+    {}
 
     virtual ~CombineLatest() = default;
 
@@ -193,9 +107,9 @@ class CombineLatest : public WritableAcceptor<std::tuple<TypesT...>>
         // Check if we should push the new value
         if (m_values_set == sizeof...(TypesT))
         {
-            // std::tuple<TypesT...> new_val = surely2(m_state);
+            std::tuple<TypesT...> new_val = utils::tuple_surely(m_state);
 
-            // status = this->get_writable_edge()->await_write(std::move(new_val));
+            status = this->get_writable_edge()->await_write(std::move(new_val));
         }
 
         return status;
@@ -209,6 +123,9 @@ class CombineLatest : public WritableAcceptor<std::tuple<TypesT...>>
 
         if (m_completions == sizeof...(TypesT))
         {
+            // Clear the held tuple to remove any dangling values
+            m_state = std::tuple<std::optional<TypesT>...>();
+
             WritableAcceptor<std::tuple<TypesT...>>::release_edge_connection();
         }
     }