flow api parallel step

[czajah]

How to do some transformation in parallel?
Can I use somehow actor_pool to configure how many threads I want to use?

sys.spawn([n = cfg.n](caf::event_based_actor* self) {
        self
            ->make_observable()
            .iota(1)

            .map( ) <---- here some operation executed in parallel 

            .take(n)
            .for_each([](int x) { std::cout << x << '\n'; });
    });

[Neverlord]

You can send requests in the flow and then merge the responses back, e.g.:

.concat_map([self{self}, worker](int32_t x) {
            return self->request(worker, infinite, x).as_observable<int32_t>();
          })

If you don't care about the order, you can also use flat_map. The worker actor could then internally dispatch. Right now we don't have an API in place to manage the worker set directly at the actor that runs the flow.

Since you've already mentioned actor_pool:

  auto pool = actor_pool::make(self->context(), 5, spawn_worker,
                               actor_pool::round_robin());

This would give you a pool with five workers, whereas spawn_worker is a function objet that returns an actor. But the actor pools are quit clumsy to use, I gotta say. For now, they probably do the job, though.

Also, just to be clear, the actor pool dispatches to actors, not threads. They are cooperatively scheduled, event-based actors. If you want to force a single thread for each worker, you can spawn them with spawn<detached>(...). I would recommend to try without detaching first, though. CAF already has a thread pool running for actors, so detaching shouldn't be necessary (expect for actors that might starve others by doing long-blocking function calls).

[czajah]

@Neverlord and some more examples:

this:

class my_transform {
public:
    using input_type = int32_t;
    using output_type = flow::observable<int32_t>;

    explicit my_transform(event_based_actor* async_act,actor &buddy): async_act(async_act), buddy(buddy) {

    }

    template <class Next, class... Steps>
    bool on_next(const input_type& item, Next& next, Steps&... steps) {
        next.on_next(async_act->request(buddy, infinite, item).as_observable<int32_t>(), steps...);
        return true;
    }

    template <class Next, class... Steps>
    void on_complete(Next& next, Steps&... steps) {
        next.on_complete(steps...);
    }

    template <class Next, class... Steps>
    void on_error(const error& what, Next& next, Steps&... steps) {
        next.on_error(what, steps...);
    }
private:
    actor buddy;
    event_based_actor* async_act;
};

src->make_observable()
        .iota(1)
        .observe_on(sink)
        .transform(my_transform{sink,pool})
        .take(10)
        .merge()
        .for_each([](auto s) { std::cout <<":: " << s << '\n'; });

returns:

:: -8
:: -1
:: -9
:: -2
:: -10
:: -7
:: -6
:: -5
:: -4
:: -3

and it could be what I want but unfortunately moving merge up as in:

    src->make_observable()
            .iota(1)
            .observe_on(sink)
            .transform(my_transform{sink,pool})
            .merge()
            .flat_map([](int32_t x) -> std::optional<int32_t> {
                if(x%2==0){
                    return std::optional<int32_t>{x};
                } else {
                    return std::nullopt;
                }
            })
            .take(10)
            .for_each([](auto s) { std::cout <<":: " << s << '\n'; });

results in:

:: -2
:: -10
:: -18
:: -26
:: -34
:: -42
:: -50
:: -58
:: -66

Process finished with exit code 139 (interrupted by signal 11: SIGSEGV)

[Neverlord]

I've tracked down why the flat_map explodes. Let me fix that first and then look into the other examples. 🙂

[Neverlord]

@czajah the fix just landed in master. So feel free to give it another spin.

[czajah]

@Neverlord It's better, but this code takes much longer to execute than the expected four seconds:

    actor_pool::factory spawn_worker = [&] {
        return sys.spawn([]{
            return behavior{
                [](int32_t i) {
                    std::cout<<"worker sleep: "<<i<<std::endl;
                    std::this_thread::sleep_for(std::chrono::seconds(2));
                    std::cout<<"worker wake: "<<i<<std::endl;

                    return -i;
                }
            };
        });
    };

    auto pool = actor_pool::make(new scoped_execution_unit(&sys), 5, spawn_worker,
                                 actor_pool::round_robin());

    sys.spawn([n = cfg.n,pool](caf::event_based_actor* self) {
        self
                ->make_observable()
                .iota(1)

                .flat_map([self{self}, pool](int32_t x) {
                    return self->request(pool, infinite, x).as_observable<int32_t>();
                })
                .take(10)

            .for_each([](int x) { std::cout <<":: " << x << '\n'; });
    });

Also, in the output, I can see some high numbers... Is there some kind of backpressure implemented in CAF? Can I control it?

[Neverlord]

There is flow-control between all stages of a flow. The flat_map operator also has a max-concurrent setting now, but this wont't allow you to cut off the requests immediately. The operator tries to "fill" the subscribed observables from downstream. By default, the max-concurrent setting is 8 now. This means the operator tries to have 8 observables from flat_map at any time (until the upstream can no longer deliver).

If you need the take(10) downstream to cut off the inputs from flat_map immediately, I'm not sure how that could work in a pipeline. We currently don't have back-pressure operators implemented, but I'm not even sure they would help here, since you'd have to somehow "wormhole" demand from the take operator and inject it between iota and flat_map. Does ReactiveX specify operators that would give you your desired behavior?