基于C++实现带延续性的Haskell风格zip函数并行执行问题

Great question! The core issue here is that you can't pre-declare the types of a and b in your F_zip::operator() because their types are only known when the upstream/other callbacks fire. But since your F-style types produce a single, predictable value type (like Haskell's IO monad), we can solve this with type traits to extract those value types upfront, then use a shared state to coordinate between the two parallel callbacks.

Here's a step-by-step solution:

1. Add Value Type Traits #

First, define a trait to extract the value type produced by any F-derived struct. This lets us know the type of values upstream and other will emit before their callbacks run.

// Trait to get the value type produced by an F-derived type
template <typename F>
struct value_type_traits;

// Specialization for F_id
template <typename X>
struct value_type_traits<F_id<X>> {
    using type = X;
};

// Specialization for F_zip (its value type is the result of the zip function)
template <typename Upstream, typename Other, typename Zip>
struct value_type_traits<F_zip<Upstream, Other, Zip>> {
    using upstream_value = typename value_type_traits<Upstream>::type;
    using other_value = typename value_type_traits<Other>::type;
    using type = decltype(std::declval<Zip>()(std::declval<upstream_value>(), std::declval<other_value>()));
};

// Helper alias for cleaner syntax
template <typename F>
using value_type_t = typename value_type_traits<F>::type;

2. Rewrite F_zip's operator() with Shared State #

We'll use a shared state (wrapped in std::shared_ptr) to hold the values from upstream and other, track their readiness, and coordinate calling the final callback once both values are available. We also add thread safety with a mutex in case your "parallel" execution uses multiple threads.

template <typename Upstream, typename Other, typename Zip> struct F_zip : F<F_zip<Upstream, Other, Zip>> {
    Upstream upstream;
    Other other;
    Zip zip;

    F_zip(const Upstream& upstream, const Other& other, const Zip& zip) 
        : upstream(upstream), other(other), zip(zip) {}

    // Forwarding constructor for movable types (optional optimization)
    F_zip(Upstream&& upstream, Other&& other, Zip&& zip) 
        : upstream(std::move(upstream)), other(std::move(other)), zip(std::move(zip)) {}

    template <typename Callback>
    void operator()(Callback&& callback) const {
        using AType = value_type_t<Upstream>;
        using BType = value_type_t<Other>;

        // Shared state to coordinate between callbacks
        struct State {
            AType a;
            BType b;
            bool has_a = false;
            bool has_b = false;
            Zip zip;
            std::decay_t<Callback> cb;
            std::mutex mtx; // For thread safety if callbacks run in parallel

            State(Zip z, Callback&& c) 
                : zip(std::move(z)), cb(std::forward<Callback>(c)) {}

            // Try to invoke the final callback if both values are ready
            void try_invoke() {
                std::lock_guard<std::mutex> lock(mtx);
                if (has_a && has_b) {
                    cb(zip(a, b));
                }
            }
        };

        auto state = std::make_shared<State>(std::move(zip), std::forward<Callback>(callback));

        // Trigger upstream execution
        upstream([state](AType&& val_a) {
            std::lock_guard<std::mutex> lock(state->mtx);
            state->a = std::forward<AType>(val_a);
            state->has_a = true;
            state->try_invoke();
        });

        // Trigger other execution (parallel to upstream)
        other([state](BType&& val_b) {
            std::lock_guard<std::mutex> lock(state->mtx);
            state->b = std::forward<BType>(val_b);
            state->has_b = true;
            state->try_invoke();
        });
    }
};

3. Extend Traits for Other F Types #

If you have other F-derived types (like your fmap implementation), add specializations to value_type_traits to keep the system consistent. For example, for a F_fmap struct:

template <typename Upstream, typename Func>
struct F_fmap : F<F_fmap<Upstream, Func>> {
    Upstream upstream;
    Func func;

    // ... constructor ...

    template <typename Callback>
    void operator()(Callback&& callback) const {
        upstream([=](auto val) {
            callback(func(val));
        });
    }
};

// Trait specialization for F_fmap
template <typename U, typename F>
struct value_type_traits<F_fmap<U, F>> {
    using upstream_val = typename value_type_traits<U>::type;
    using type = decltype(std::declval<F>()(std::declval<upstream_val>()));
};

How This Works #

• The value_type_traits lets us know exactly what types a and b will be before the callbacks run, so we can declare them in the shared State struct.
• We trigger both upstream and other executions immediately (in parallel).
• Each callback updates the shared state and checks if both values are ready. If so, it invokes the final callback with the zipped result.
• The mutex ensures thread safety if your callbacks run on different threads (remove it if you're using single-threaded async execution).

Your existing test_zip() function will work exactly as before, but now the two F_id executions run in parallel instead of sequentially.

内容的提问来源于stack exchange，提问作者Artiano