How to perform parallel evaluation of streams ?

[jerbaroo]

#108 was helpful for me to understand how to write effects that take or return streams, for example an effect to read a file that returns a stream of bytes.

However, now I am struggling with applying the parEval function to Streams from the Streamly library. Here is the definition of parEval and MonadAsync:

-- https://hackage.haskell.org/package/streamly-0.10.1/docs/Streamly-Data-Stream-Prelude.html#v:parEval
parEval :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a

-- https://hackage.haskell.org/package/streamly-0.10.1/docs/Streamly-Data-Stream-Prelude.html#t:MonadAsync
type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

When in plain IO it's straightforward to use parEval, however within a context constrained by effects (where IO may or may not be possible), I am not sure how to achieve this.

Thank you

[arybczak]

To apply parEval to Stream (Eff es) a you need IOE :> es in scope, there's no way around this 🤔

You might also need to use withUnliftStrategy and set ConcUnlift with it if streamly utilizes MonadBaseControl in multiple threads, but you'll get appropriate error at runtime if default SeqUnlift is not enough.

[jerbaroo]

Thanks for the quick response. As a workaround, might it be possible to define an effect StreamParEvalE for parallel evaluation, and then use the parEval from above in the IO-based implementation of the effect.. something along the lines of:

data StreamParEvalE :: Effect where
  StreamParEval
    :: (Stream.Config -> Stream.Config)
    -> Stream m a
    -> StreamParEvalE m (Stream m a)

type instance DispatchOf StreamParEvalE = Dynamic

streamParEval
  :: StreamParEvalE :> es
  => (Stream.Config -> Stream.Config)
  -> Stream (Eff es) a
  -> Eff es (Stream (Eff es) a)
streamParEval f = Eff.send . StreamParEval f

runStreamParEvalIO :: forall es a. (IOE :> es) => Eff (StreamParEvalE : es) a -> Eff es a
runStreamParEvalIO = Eff.interpret \env ->
  \(StreamParEval f s) -> Eff.localSeqUnlift env \unlift -> do
    let x :: Stream (Eff es) _ = Stream.morphInner unlift s
    let y :: Stream (Eff es) _ = Stream.parEval f x
    pure y -- Could not deduce ‘localEs ~ es’

Or does this not make any sense at all? 😄

[arybczak]

Interesting idea. This compiles (didn't test whether it works, probably unlift strategy needs some update):

import Effectful
import Effectful.Dispatch.Dynamic
import Streamly.Data.Stream (Stream)
import Streamly.Data.Stream.Prelude qualified as Stream

data StreamParEvalE :: Effect where
  StreamParEval
    :: (Stream.Config -> Stream.Config)
    -> Stream m a
    -> StreamParEvalE m (Stream m a)

type instance DispatchOf StreamParEvalE = Dynamic

streamParEval
  :: StreamParEvalE :> es
  => (Stream.Config -> Stream.Config)
  -> Stream (Eff es) a
  -> Eff es (Stream (Eff es) a)
streamParEval f = send . StreamParEval f

runStreamParEvalIO :: (IOE :> es) => Eff (StreamParEvalE : es) a -> Eff es a
runStreamParEvalIO = interpret $ \env -> \case
  StreamParEval f (s :: Stream (Eff localEs) a) -> do
    localSeqLend @'[IOE] env $ \runIOE -> do
      let x :: Stream (Eff (IOE : localEs)) a = Stream.morphInner inject s
      let y :: Stream (Eff (IOE : localEs)) a = Stream.parEval f x
      let z :: Stream (Eff localEs) a = Stream.morphInner runIOE y
      pure z

I added type annotations for readability, you need only to annotate the type of x to let inject know we're running in a stack with IOE.

[jerbaroo]

For anyone who may happen on this discussion. The following code results in the run-time error below:

runStreamParEvalIO :: (IOE :> es) => Eff (StreamParEvalE : es) x -> Eff es x
runStreamParEvalIO = Eff.interpret \env -> \case
  (StreamParEval f (s :: Stream (Eff localEs) a)) -> do
    let strat = Eff.ConcUnlift Eff.Ephemeral Eff.Unlimited
    Eff.localLend @'[IOE] env strat \runIOE -> do
      let x :: Stream (Eff (IOE : localEs)) a = Stream.morphInner Eff.inject s
      let y :: Stream (Eff (IOE : localEs)) a = Stream.parEval f x
      let z :: Stream (Eff localEs) a = Stream.morphInner runIOE y
      pure z

exe: If you want to use the unlifting function to run Eff computations in multiple threads, have a look at UnliftStrategy (ConcUnlift).
CallStack (from HasCallStack):
  error, called at src/Effectful/Internal/Monad.hs:234:12 in effectful-core-2.5.1.0-KeXwreUwGx0HOvqXDasFMg:Effectful.Internal.Monad
  seqUnliftIO, called at src/Effectful/Internal/Monad.hs:201:40 in effectful-core-2.5.1.0-KeXwreUwGx0HOvqXDasFMg:Effectful.Internal.Monad
  withEffToIO, called at src/Effectful/Internal/Monad.hs:393:40 in effectful-core-2.5.1.0-KeXwreUwGx0HOvqXDasFMg:Effectful.Internal.Monad

[arybczak]

You need to also set ConcUnlift with withUnliftStrategy. The slight problem here is the scope - the stream needs to be evaluated within the withUnliftStrategy block.