Interpreting effectful callbacks in effects in the same environment

[osa1]

I'm trying to write an interpretable (i.e. dynamic) concurrent effect.

The effect type should be parametric on synchronization types, like MVar, Async etc.

So far I have this:

data Concurrent' threadHandle m a where
    Async :: m a -> Concurrent' threadHandle m (threadHandle a)
    Await :: threadHandle a -> Concurrent' threadHandle m a

type instance DispatchOf (Concurrent' threadHandle) = Dynamic

async' :: Concurrent' threadHandle :> es => Eff es a -> Eff es (threadHandle a)
async' = send . Async

await' :: Concurrent' threadHandle :> es => threadHandle a -> Eff es a
await' = send . Await

To handle this, I think I need reinterpret, and since the Async callback effects need to be interpreted by the same handler, I probably need something like:

runConcurrent'IO :: IOE :> es => Eff (Concurrent' Async : es) a -> Eff es a
runConcurrent'IO = reinterpret runConcurrent'IO $ \_ -> \case
    Async action -> undefined
    Await handle -> liftIO (A.wait handle)

But I can't figure out what to put in the undefined part. Any tips?

Thanks.

[arybczak]

You need to use localUnliftIO with ConcUnlift strategy (Ephemeral with Limited 1 will work).

Have a look at haddock if these for more info. There's also an example of a higher order effect in the introduction section of Effectful.Dispatch.Dynamic.

[osa1]

Thanks. Here's the full code:

data Concurrent' threadHandle m a where
    Async :: m a -> Concurrent' threadHandle m (threadHandle a)
    Await :: threadHandle a -> Concurrent' threadHandle m a

type instance DispatchOf (Concurrent' threadHandle) = Dynamic

async' :: Concurrent' threadHandle :> es => Eff es a -> Eff es (threadHandle a)
async' = send . Async

await' :: Concurrent' threadHandle :> es => threadHandle a -> Eff es a
await' = send . Await

runConcurrent'IO :: IOE :> es => Eff (Concurrent' Async : es) a -> Eff es a
runConcurrent'IO = reinterpret runConcurrent'IO $ \env -> \case
    Async action -> localUnliftIO env (ConcUnlift Ephemeral Unlimited) (\unlift -> liftIO (A.async (unlift action)))
    Await handle -> liftIO (A.wait handle)

I haven't tried to run it yet, but it type checks.

[osa1]

Unfortunately this seems to be looping and leaking memory...

The other issue is that passing a function that uses this effect type to runConcurrent'IO is not enough to specify threadHandle type parameter, I still get various type errors unless I explicitly apply the threadHandle type, e.g. runConcurrent'IO (myFunction @A.Async). For that I also need to move the threadHandle type parameter to the first position. It's not really usable.

[ocharles]

I don't see why you need that recursive reinterpret - interpret should be fine here. Does it work if you just use interpret? For the type parameters, you should be able to use an explicit forall and reorder the type variables accordingly

[osa1]

I don't see why you need that recursive reinterpret - interpret should be fine here.

I was thinking that because the same handler needs to handle effects in the Async action I need a recursive definition here, but you are right, interpreter works and it fixes the looping.

For the type parameters, you should be able to use an explicit forall and reorder the type variables accordingly

I'm already doing that it doesn't seem to be a scalable approach, if I have multiple effects like this. Currently my signatures are already out of control:

main' :: forall threadHandle es . (Fail :> es, Environment :> es, Console :> es, FileSystem :> es, Concurrent' threadHandle :> es) => Eff es ()

run :: forall threadHandle es . (Concurrent' threadHandle :> es, FileSystem :> es) => String -> [FilePath] -> Eff es [(FilePath, [Match])]

It might help if I could specify only some of the type parameters, for the ones I want to specify in the call sites.

I also still don't understand why I need to explicitly pass the threadHandle argument. Full example:

#!/usr/bin/env stack
{-
stack script
    --resolver lts-22.38
    --package async
    --package effectful
    --package mtl
    --package random
    --package transformers
    --package text
    --package bytestring
    --ghc-options -Wall
-}

{-# LANGUAGE DataKinds, LambdaCase, TypeFamilies, OverloadedStrings, AllowAmbiguousTypes #-}

import Prelude hiding (putStrLn, readFile)
import qualified Prelude as P
import qualified System.IO as P

import qualified Control.Concurrent.Async as A
import Data.Kind
import Data.Traversable
import Control.Monad
import Effectful
import Effectful.Dispatch.Dynamic
import Effectful.Dispatch.Static (unEff, unsafeEff)
import Effectful.Environment
import Effectful.Fail
import Effectful.FileSystem.IO hiding (stdout, stderr, stdin)
import Effectful.FileSystem.IO.ByteString
import Effectful.State.Static.Local
import qualified Data.Text as T
import qualified Data.Text.Encoding as T

main :: IO ()
main = runEff $ runFailIO $ runEnvironment $ runConsoleIO $ runFileSystem $ runConcurrent'IO $ main' @A.Async

main' :: forall threadHandle es . (Fail :> es, Environment :> es, Console :> es, FileSystem :> es, Concurrent' threadHandle :> es) => Eff es ()
main' =
    getArgs >>= \case
        pattern : (files@(_ : _)) -> do
            results <- run @threadHandle pattern files
            forM_ results $ \(file, matches) -> do
                putStrLn (file ++ ":")
                forM_ matches $ \match -> do
                    putStrLn (show (matchLine match + 1) ++ ":" ++ show (matchCharacter match + 1))

        _ -> putStrLnErr "Usage: ./Grep.hs pattern file1 file2 ..."

run :: forall threadHandle es . (Concurrent' threadHandle :> es, FileSystem :> es) => String -> [FilePath] -> Eff es [(FilePath, [Match])]
run str files = do
    asyncJobs :: [threadHandle (FilePath, [Match])] <- forM files (\file -> async' (fmap (file,) (searchFile file str)))
    forM asyncJobs await'

data Match = Match
    { matchLine :: Int
    , matchCharacter :: Int
    }

searchFile :: FileSystem :> es => FilePath -> String -> Eff es [Match]
searchFile filePath str = do
    fileContents <- readFile filePath
    let fileContentsText = T.decodeUtf8 fileContents
    let occurrences = init (T.splitOn (T.pack str) fileContentsText)
    pure (snd (mapAccumL occurrenceToMatch (0, 0) occurrences))
  where
    occurrenceToMatch :: (Int, Int) -> T.Text -> ((Int, Int), Match)
    occurrenceToMatch currentPos part =
        let (matchPosLine, matchPosChar) = incrementPosText currentPos part in
        let nextPos = (matchPosLine, matchPosChar + searchTermSize) in
        (nextPos, Match { matchLine = matchPosLine, matchCharacter = matchPosChar })

    searchTermSize :: Int
    searchTermSize = length str

incrementPosText :: (Int, Int) -> T.Text -> (Int, Int)
incrementPosText = T.foldl' incrementPosChar

incrementPosChar :: (Int, Int) -> Char -> (Int, Int)
incrementPosChar (lineNum, _) '\n' = (lineNum + 1, 0)
incrementPosChar (lineNum, charNum) _ = (lineNum, charNum + 1)

---------------------------------------------------------------------------------

data Console :: Effect where
  PutStrLn :: String -> Console m ()
  PutStrLnErr :: String -> Console m ()

type instance DispatchOf Console = Dynamic

putStrLn :: Console :> es => String -> Eff es ()
putStrLn = send . PutStrLn

putStrLnErr :: Console :> es => String -> Eff es ()
putStrLnErr = send . PutStrLnErr

runConsoleIO :: IOE :> es => Eff (Console : es) a -> Eff es a
runConsoleIO = interpret $ \_ -> \case
    PutStrLn str -> liftIO (P.putStrLn str)
    PutStrLnErr str -> liftIO (P.hPutStrLn P.stderr str)

data PureConsoleState = PureConsoleState {
  stdout :: [String],
  stderr :: [String]
}

runConsolePure :: Fail :> es => Eff (Console : es) a -> Eff es (a, PureConsoleState)
runConsolePure = reinterpret (runState (PureConsoleState [] [])) $ \_ -> \case
    PutStrLn str ->
      modify @PureConsoleState $ \(PureConsoleState stdout stderr) ->
        PureConsoleState (stdout ++ [str]) stderr

    PutStrLnErr str ->
        modify @PureConsoleState $ \(PureConsoleState stdout stderr) ->
        PureConsoleState stdout (stderr ++ [str])

---------------------------------------------------------------------------------

data Concurrent' threadHandle m a where
    Async :: m a -> Concurrent' threadHandle m (threadHandle a)
    Await :: threadHandle a -> Concurrent' threadHandle m a

type instance DispatchOf (Concurrent' threadHandle) = Dynamic

async' :: Concurrent' threadHandle :> es => Eff es a -> Eff es (threadHandle a)
async' = send . Async

await' :: Concurrent' threadHandle :> es => threadHandle a -> Eff es a
await' = send . Await

runConcurrent'IO :: IOE :> es => Eff (Concurrent' A.Async : es) a -> Eff es a
runConcurrent'IO = interpret $ \env -> \case
    Async action -> localUnliftIO env (ConcUnlift Ephemeral Unlimited) (\unlift -> liftIO (A.async (unlift action)))
    Await handle -> liftIO (A.wait handle)

I would expect the type of runConcurrent'IO to specify the threadHandle type parameter of main', but it's not happening for some reason.