2008/10/9 Claus Reinke

I was wondering if it was possible to implement synchronous channels

...

within STM. In particular, I'd like to have CSP-like send and recv primitives on a channel that each block until the other side arrives to complete the transaction.

Assuming that retry blocks until something changes, you could associate a channel with a thread that encapsulates the transaction. Somewhat like this?

You don't need an additional channel thread: module SyncChan (SyncChan, send, recv, newSyncChan) where import Control.Concurrent.STM import Control.Monad import Control.Concurrent newtype SyncChan a = SC { unSC :: TVar (State a) } data State a = Ready | Sent a | Received newSyncChan :: STM (SyncChan a) newSyncChan = SC `fmap` newTVar Ready send :: SyncChan a -> a -> IO () send (SC chan) x = do atomically $ unsafeSend chan x atomically $ waitReceiver chan recv :: SyncChan a -> STM a recv (SC chan) = do s <- readTVar chan case s of Sent s -> writeTVar chan Received >> return s _ -> retry unsafeSend chan x = do s <- readTVar chan case s of Ready -> writeTVar chan (Sent x) _ -> retry waitReceiver chan = do s <- readTVar chan case s of Received -> writeTVar chan Ready _ -> retry x |> f = fmap f x test b = do (x,y) <- atomically $ liftM2 (,) newSyncChan newSyncChan forkIO $ join $ atomically $ -- since recv is in STM you can wait on multiple channels at the same time (recv x |> print) `mplus` (recv y |> print) if b then send x 'a' else send y 1 as a bonus you can also try to send to the first available among multiple channels: (this formulation uses ExistentialQuantification but it's just a convenience) data Sending a = forall b. Sending (SyncChan b) b a sendMulti :: [Sending a] -> IO a sendMulti [] = fail "empty" sendMulti xs = do (m,r) <- atomically $ msum $ map sending xs atomically m return r sending :: Sending t -> STM (STM (), t) sending (Sending (SC chan) x k) = do unsafeSend chan x return (waitReceiver chan,k)

On Thu, Oct 9, 2008 at 9:15 AM, Ryan Ingram wrote:

I don't think what you want is possible if both sides are in STM. Other authors have posted solutions where one side or the other of the transaction is in I/O, but wholly inside STM it's not possible.

Thanks, that's what I thought, although I wasn't sure of it, being new to both Haskell and STM. Presumably this result means that it's not possible to implement any bounded-buffer-type interface within (rather than on top of) STM. Isn't that a rather serious restriction? cheers, rog.

On Thu, Oct 9, 2008 at 10:12 AM, Arnar Birgisson wrote:

Sorry, I come into this discussion late. One-place buffers, or MVars, are indeed implemented over STM in the orignal paper [1].

Yes, I should have remembered that. It's ok just as long as there's a buffer there because then there's a consistent *memory* state that can be awaited. But using an unbuffered channel, we need to await the state of another *process*, so we can rendezvous with it, something that is not possible in principle with STM.

Is that what you seek?

It's useful, thanks, but not really what I was originally looking for. Synchronous channels are generally easier to reason about (less states to deal with). On Thu, Oct 9, 2008 at 10:19 AM, Andrea Vezzosi wrote:

I'd rather say that STM is intended to be used just for building up transactions

It seems to me then that, despite the existence of the very elegant STM, there's still room for a commonly used set of transactions outside of it, with as many as possible within STM, along the lines of the SyncChan snippet you posted earlier. By the way, where does FRP (which I haven't got my head around yet) sit with respect to STM? cheers, rog.

roger peppe wrote:

By the way, where does FRP (which I haven't got my head around yet) sit with respect to STM?

Entirely orthogonal. FRP is not generally thought of as (explicitly) threaded at all. It's more declarative than that. It's also supposed to be deterministic (up to the determinism of input events) which STM is not. However an elegant, efficient, implementation of FRP in pure haskell evades us, so people discuss different ways to implement it which use, possibly, concurrency such as STM or otherwise, under the hood. So STM may or may not be a good tool to implement FRP, but at the level that you *use* FRP, any threading should be entirely implicit. Jules

On Thu, Oct 9, 2008 at 1:50 AM, roger peppe wrote:

On Thu, Oct 9, 2008 at 9:15 AM, Ryan Ingram wrote:

...

I don't think what you want is possible if both sides are in STM. Other authors have posted solutions where one side or the other of the transaction is in I/O, but wholly inside STM it's not possible.

Thanks, that's what I thought, although I wasn't sure of it, being new to both Haskell and STM.

Presumably this result means that it's not possible to implement any bounded-buffer-type interface within (rather than on top of) STM.

Isn't that a rather serious restriction?

I don't know that it's practically-speaking that serious. One can write it in IO, using STM. I think of CSP as I/O anyway, but perhaps my thinking is flawed and dirty from MPI and Erlang "message passing" :-). Then again, I'm not sure why keeping it in STM is even valuable really. IO gets the job done right?

cheers, rog. _______________________________________________ Haskell-Cafe mailing list Haskell-Cafe@haskell.org http://www.haskell.org/mailman/listinfo/haskell-cafe

On Thu, Oct 9, 2008 at 18:10, Arnar Birgisson wrote:

...

But I did write a concurrent prime sieve with it:

I did the same, with the one-place-buffers (the MVars implemented over STM). Be warned that there is no stop condition, this just keeps printing primes forever.

Please forgive me for reposting, but the last one exited quite prematurely :) module Main where import Control.Concurrent (forkIO) import Control.Concurrent.STM import System (getArgs) -- MVars from the STM paper type MVar a = TVar (Maybe a) newEmptyMVar :: STM (MVar a) newEmptyMVar = newTVar Nothing takeMVar :: MVar a -> STM a takeMVar mv = do v <- readTVar mv case v of Nothing -> retry Just val -> do writeTVar mv Nothing return val putMVar :: MVar a -> a -> STM () putMVar mv val = do v <- readTVar mv case v of Nothing -> writeTVar mv (Just val) Just _ -> retry -- Sieve forever a = do a; forever a pfilter :: Int -> MVar Int -> MVar Int -> IO () pfilter p in_ out = forever $ do atomically $ do v <- takeMVar in_ if v `mod` p /= 0 then putMVar out v else return () sieve :: MVar Int -> MVar Int -> IO () sieve in_ out = do p <- atomically $ takeMVar in_ atomically $ putMVar out p ch <- atomically $ newEmptyMVar forkIO $ pfilter p in_ ch sieve ch out feeder :: MVar Int -> IO () feeder out = feed' 2 where feed' i = do atomically $ putMVar out i feed' (i+1) printer :: MVar () -> MVar Int -> Int -> IO () printer stop in_ max = do v <- atomically $ takeMVar in_ putStrLn $ show v if v > max then atomically $ putMVar stop () else printer stop in_ max main :: IO () main = do max:_ <- getArgs in_ <- atomically newEmptyMVar out <- atomically newEmptyMVar stop <- atomically newEmptyMVar forkIO $ feeder in_ forkIO $ printer stop out (read max) forkIO $ sieve in_ out atomically $ takeMVar stop return ()