Concurrency primitimes and thread safety

[tomjaguarpaw]

We don't have any Bluefin-specific concurrency primitives and just rely on MonadUnliftIO to give us access to IO currency primitives. This seems dangerous, firstly in light of #29, but also because we could actually use Bluefin's type system to forbid thread-unsafe access to resources.

By way of comparison, effectful seems fairly lax. See haskell-effectful/effectful#292.

#!/usr/bin/env cabal
{- cabal:
  build-depends: base, effectful==2.5.1.0, async
-}
{-# LANGUAGE GHC2021 #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE LambdaCase #-}

import Control.Concurrent
import Control.Concurrent.Async
import Data.IORef
import Effectful
import Effectful.Dispatch.Dynamic
import Effectful.State.Dynamic

evalState ::
  (IOE :> es) =>
  s ->
  Eff (State s : es) a ->
  Eff es a
evalState s0 m = do
  v <- liftIO (newIORef s0)
  reinterpret id (ioState v) m

ioState ::
  (IOE :> es) =>
  IORef s ->
  LocalEnv localEs es ->
  State s (Eff localEs) a ->
  Eff es a
ioState v env = \case
  Get -> liftIO (readIORef v)
  Put s -> liftIO (writeIORef v s)
  State f -> liftIO $ do
    s <- readIORef v
    let (r, s') = f s
    writeIORef v s
    pure r
  StateM _ -> error "Dunno"

useStateConcurrently ::
  (State Int :> es, IOE :> es) => Eff es ()
useStateConcurrently = do
    withEffToIO (ConcUnlift Persistent Unlimited) $
      \effToIO -> do
        concurrently
          ( effToIO $ do
              liftIO (threadDelay 500)
              s <- get @Int
              put (s + 1)
          )
          ( effToIO $ do
              s <- get @Int
              liftIO (threadDelay 1000)
              put (s * 2)
          )

    (liftIO . print) =<< get @Int

-- We "want" the result to be either
--
-- - 12 (== (5 + 1) * 2), or
--
-- - 11 (== (5 * 2) + 1)
--
-- but we get
--
-- % cabal run test-effectful-thread-unsafe.hs
-- 10
main :: IO ()
main = runEff $ do
  evalState @_ @Int 5 $ do
    useStateConcurrently

[jeukshi]

Another problem with the effectful API is that threads can outlive effectful operations. I'm cheating here with IOE, but long computation would suffice.

greatEscape :: IO ()
greatEscape = do
    hSetBuffering stdout LineBuffering
    res <- runEff . runConcurrent $ do
        _ <- async $ do
            liftIO $ forever do
                print "still alive"
                threadDelay 2_000
        return "eff done"
    print res
    threadDelay 10_000

This is mentioned in the documentation, but I don't think it's necessary as a primitive, it can be done with IOE anyway, if needed. I'd rather go with structured concurrency from the start.

Speaking of which, I did some experiments with Ki. The Naive version just exposes that this is IORefs all the way down, which is yikes.

This is my current best idea, with explicit copy/share through a typeclass. I did not think this through and did not use it for anything, but maybe it can serve as an inspiration.

[tomjaguarpaw]

Thanks! I have some ideas forthcoming about structured concurrency in Bluefin which give thread safety by construction, so watch this space. I have some concerns about the ki API, but think I have worked out how to make it type safe with Bluefin.

[jeukshi]

Oh, for the record, this doesn't with my wrapper.

useFile = runEff \io -> do
    runScope io \scope -> do
        _ <- BIO.withFile io "/dev/null" WriteMode \h -> do
            _ <- forkWithNewEff scope do
                BIO.hPutStr h "foo"
                -- • Ambiguous type variable ‘forkEs0’ arising from a use of ‘BIO.hPutStr’
               --    prevents the constraint ‘(e2 :> forkEs0)’ from being solved.
            pure ()
        pure ()

But to be fair, nothing works.

Anyway looking forward to what is cooking!

[tomjaguarpaw]

You can have a look at branch ki for a rough sketch of what I have so far. I think what's exported from Bluefin.Ki is solid, but it needs some documentation and some examples. (There is a long example but it's not really clear what's going on.)

The thread safety story is this:

• Bluefin.Ki is thread safe
• You can only directly use resources within your own Scope
• You can access resources from higher Scopes using exclusively
• exclusively locks everything on the path between you and the Scope you're accessing
• This is a very blunt hammer, but thread safe and deadlock free by construction
• For "lock-free" communication between threads use STM
• Still to come: a way of cloning those resources for which it's fine to pass copies of them down to lower scopes (for example an "HTTP file downloader" -- there's no problem with running multiple of them at a time) and a way of unsafely cloning resources, as an escape hatch

If you're feeling adventurous you might want to check it out. I envisage writing up a better explanation over the coming weeks.

[tomjaguarpaw]

By the way, with the inclusion of the "well-scoped Scopes" ( 😁 ) I think scope and fork are no more powerful than withAsync, but the allow you to avoid nesting and keep everything in the same do block, so I guess that's nice.

[jeukshi]

Thanks! I did some shallow dive. Double scope ("scope" is becoming a bit overloaded term, I blame Ki tho, nursery might be a better term) tag is a nice trick.

I'll check a few things over time. I don't know if I should comment on the WIP code, but two things immediately came to mind.

• scoped needs IOE and maybe we need pureScoped.

impure :: Bool
impure = runPureEff $ do
  evalState (0 :: Integer) $ \s -> do
    scoped $ \scope -> do
      t1 <- fork scope $ \excl1 -> do
        let x = sum [1 .. 9_999_999] -- long op
        exclusively excl1 $ do
          put s x
          put s 1
      t2 <- fork scope $ \excl2 -> do
        let x = sum [1 .. 9_999_999] -- long op
        exclusively excl2 $ do
          put s x
          put s 2
      _ <- awaitEff t1
      _ <- awaitEff t2
      pure ()
    res <- get s
    return $ res > 1

λ> filter (==True) . map (\_ -> impure) $ [1..10]
[True,True,True,True,True,True,True,True,True,True]
λ> :r
Ok, two modules loaded.
λ> filter (==True) . map (\_ -> impure) $ [1..10]
[]

• Scope has that extra tag, which might be a little annoying to carry around. One might want to pass around Scope, as this is the way of saying "you can create threads that outlive you". I'll have to play with it a bit more.

I think scope and fork are no more powerful than withAsync

I like it, because it is a handle, which goes with the spirit of the library.

[tomjaguarpaw]

scoped needs IOE and maybe we need pureScoped

Oh, that's really interesting. I hadn't spotted you could get impurity arising from non-determinism of operation lengths. Thanks for pointing that out!

I don't know if I should comment on the WIP code

Probably best to leave more thoughts here, as the branch will be changing a lot, and be force-rebased, etc..

[tomjaguarpaw]

scoped needs IOE and maybe we need pureScoped

OK, scoped now requires NonDet (which itself can only be obtained from IOE):

bluefin/bluefin-internal/src/Bluefin/Internal/Concurrent.hs

Lines 55 to 71 in 409f8e3

	scoped ::
	(e1 :> es) =>
	NonDet e1 ->
	(forall e. Scope es e -> Eff e r) ->
	-- | ͘
	Eff es r
	scoped nonDet@(UnsafeMkNonDet io) k =
	withEffToIO_ io $ \effToIO -> Ki.scoped $ \scope -> do
	-- Unlocked when it's empty
	lock <- newEmptyMVar
	effToIO
	( k
	( UnsafeMkScope
	scope
	(UnsafeMkExclusiveAccess lock UnsafeMkSTME (mapHandle nonDet))
	)
	)

What were you imagining for pureScoped?

[jeukshi]

What were you imagining for pureScoped?

While we cannot guarantee purity with sharing arbitrary effects across threads boundaries, it is not all that useless to try to make a pure interface for threading. There is some, ehm, prior art. It was all the rage back in the day :) We could even fix some deficiencies.

I imagine Bluefin.Parallel as a separate module. I have created PoC here and coming across this I couldn't stop myself from adding streaming too. There are some examples.

Now, my name is not Simon, so it is probably broken in a few unparalleled ways, but it is a start. I did some simple benchmarks too, and we are on par with monad-par (pardon!).

benchmarking map/pure
time                 630.4 ms   (626.7 ms .. 634.4 ms)
                     1.000 R²   (1.000 R² .. 1.000 R²)
mean                 633.5 ms   (631.5 ms .. 634.6 ms)
std dev              1.543 ms   (126.5 μs .. 1.920 ms)
variance introduced by outliers: 19% (moderately inflated)

benchmarking map/parMap'book
time                 299.4 ms   (290.7 ms .. 306.5 ms)
                     1.000 R²   (0.999 R² .. 1.000 R²)
mean                 302.7 ms   (300.8 ms .. 304.9 ms)
std dev              2.726 ms   (2.292 ms .. 3.196 ms)
variance introduced by outliers: 16% (moderately inflated)

benchmarking map/parMapBf
time                 295.4 ms   (292.4 ms .. 298.6 ms)
                     1.000 R²   (1.000 R² .. 1.000 R²)
mean                 301.5 ms   (298.5 ms .. 306.1 ms)
std dev              4.990 ms   (33.80 μs .. 6.312 ms)
variance introduced by outliers: 16% (moderately inflated)

The main benefit of adding Bluefin.Parallel would be that users could still use pure effects when writing parallel code, just not for inter-thread communication. Plus, we would offer some useful functions out of the box.

I propose this: if you are interested, I can open a new issue about this, and we will leave this thread for concurrent primitives. Parallel stuff could be added later (there is an easy parallel pun awaiting here, but I have resisted).

[tomjaguarpaw]

I propose this: if you are interested, I can open a new issue about this

Oh nice, yes, I'm interested, so please start that new issue!

[jeukshi]

After trying it for a bit, here are some thoughts.

exclusively is a footgun.

runApp :: IO ()
runApp = runEff \io -> do
    BC.runSTM io \stmMain -> do
        BC.withNonDet io \nonDet -> do
            BC.scoped nonDet \scope -> do
                t <- BC.fork scope \acc -> do
                    -- background process
                    forever do
                        BC.exclusively acc do
                            effIO io do print "ping"
                t2 <- BC.fork scope \acc -> do
                    -- setup servant server
                    let app :: Application = genericServeT (nt effEnvProxy) record
                    BC.exclusively acc do
                        effIO io do Warp.run 3000 app -- OOPS, `warp` just sucked IO from my whole program!
                BC.awaitEff t2

exclusively is unergonomic.

logging :: IO ()
logging = runEff \io -> do
    runReader "global" \re -> do
        BC.withNonDet io \nonDet -> do
            BC.scoped nonDet \scope -> do
                t1 <- BC.fork scope $ \acc -> do
                    BC.exclusively acc do
                        -- WELP, I block _everything_ to just do some logging!
                        r <- ask re
                        local re (const "local") do
                            rl <- ask re
                            effIO io do print rl
                BC.awaitEff t1

And I'm pretty sure I can write an effect with TChan where the reader will just block the writer from ever accessing it, and there will be no good way to prevent it from happening.

So I went back to my typeclass approach, but with some ideas from the branch.

class ThreadSafe a where
    accessConcurrently :: (e' :> es', e :> es) => BC.ExclusiveAccess es' e -> a e' -> Eff es (a e)
-- `accessConcurrently` is basically `accessSTME`, but for everything that has an instance

instance ThreadSafe IOE where
    accessConcurrently _ _ = pure Bluefin.Internal.MkIOE

data BroadcastClient a e = UnsafeMkBroadcastClient
    { ioe :: IOE e
    , inChan :: InChan a
    }

instance ThreadSafe (BroadcastClient a) where
    accessConcurrently access (UnsafeMkBroadcastClient io inChan) =
       -- if I was using State I coudn't even write it!
        UnsafeMkBroadcastClient <$> accessConcurrently access io <*> pure inChan

data BroadcastServer a e = UnsafeMkBroadcastServer
    { ioe :: IOE e
    , inChan :: InChan a
    }

instance ThreadSafe (BroadcastServer a) where
    accessConcurrently access (UnsafeMkBroadcastServer io inChan) =
        UnsafeMkBroadcastServer <$> accessConcurrently access io <*> pure inChan

runBroadcast
    :: (e1 :> es)
    => IOE e1
    -> (forall e. BroadcastClient a e -> BroadcastServer a e -> Eff (e :& es) r)
    -> Eff es r
runBroadcast io action = do
    (iChan, _) <- effIO io newChan
    let broadcastOut = UnsafeMkBroadcastServer (mapHandle io) iChan
    let broadcast = UnsafeMkBroadcastClient (mapHandle io) iChan
    useImplIn2 action broadcast broadcastOut

writeBroadcast
    :: (e :> es)
    => BroadcastServer a e
    -> a
    -> Eff es ()
writeBroadcast (UnsafeMkBroadcastServer io inChan) x =
    effIO io do writeChan inChan x

likeAndSubscribe
    :: (e :> es, e1 :> es)
    => BroadcastClient a e
    -> Stream a e1
    -> Eff es ()
likeAndSubscribe (UnsafeMkBroadcastClient io inChan) s = do
    outChan <- effIO io do dupChan inChan
    forever do
        e <- effIO io do readChan outChan
        yield s e

example :: IO ()
example = runEff \io -> do
    runBroadcast io \broadcastMain broadcastInMain -> do
        BC.withNonDet io \nonDet -> do
            BC.scoped nonDet \scope -> do
                t1 <- BC.fork scope $ \acc -> do
                    broadcast <- accessConcurrently acc broadcastMain
                    io' <- accessConcurrently acc io
                    forEach (likeAndSubscribe broadcast) \i -> do
                        effIO io' $ print $ "t1: " <> show i

                t2 <- BC.fork scope $ \acc -> do
                    broadcast <- accessConcurrently acc broadcastMain
                    io' <- accessConcurrently acc io
                    forEach (likeAndSubscribe broadcast) \i -> do
                        effIO io' $ print $ "t2: " <> show i

                tw <- BC.fork scope $ \acc -> do
                    broadcastIn <- accessConcurrently acc broadcastInMain
                    for_ [0 .. 9] \i -> do
                        writeBroadcast broadcastIn i
                BC.awaitEff t1
                pure ()

I like it more, because it is up to the author of an effect to decide whether their effect can be used concurrently.

Now maybe there is a need for other typeclass indicating that we can work against a copy of an effect, accessCopyConcurrently or something. But I'm not sure.

[tomjaguarpaw]

exclusively is a footgun.

Yes, absolutely. It is indeed a very blunt hammer.

exclusively is unergonomic.

In this particular case I think exclusively is actually doing its job. If you don't block print like this then you end up getting interleaved output. It really is better if only one thread can write to the terminal! Now, that thread shouldn't block other threads, but that falls under "footgun".

---

In any case, I like ThreadSafe! It seems like it provides what we need. I think we could just provide instance ThreadSafe STME and avoid packaging STME in ExclusiveAccess. I'll have to give some more thought to the type of accessConcurrently, firstly to convince my self it's really what we want (as opposed to a "handler style") and secondly to make sure the use of constraints doesn't get us stuck with ambiguous type variables. I think maybe es' and e' should be the same, for ergonomic purposes. At least that's what I used in exclusively.

accessConcurrently :: (e' :> es', e :> es) => BC.ExclusiveAccess es' e -> a e' -> Eff es (a e)

We should definitely not have instance ThreadSafe IOE! It defeats the whole point of this endeavour. If IOE is ThreadSafe then anything could be given a ThreadSafe instance without any hint that it may be dangerous. We'll have to provide some well-labelled escape hatch, like unsafeAccessConcurrentlyIOE.