Re: [Haskell-cafe] broadcasting stateful computations

newer
Type synonym involing quantified...

older
New package: lazify

Olaf Klinke

2 Sep 2021 2 Sep '21

11:25 a.m.

Thanks everyone for the helpful suggestions! Bryan Richer wrote:

...

what if the TVar is part of the state s? (StateT IO) can't give any guarantees about when or how the state is broadcast, but using STM within StateT actions still can.

I don't understand this, but it may go into the same direction as: Chris Smith wrote:

...

I had the same suggestion. However, you then no longer need StateT, as ReaderT is enough. Using a `ReaderT (TVar s) IO a` will allow atomic changes to the state s, along with interleaved IO when it's done safely rather than in the middle of a transaction.

ReaderT (TVar s) IO is interesting! It does not solve the problem of atomic modifications with side-effects, though. Currently I am implicity using ReaderT (TVar s), by explicitly passing the TVar around. In principle, for atomic modifications I have to get hold of a pure version (s -> s) of the modification. Maybe within a monadic context. But if determining this modification function is side-effectful, then it can not depend on the initial contents of the TVar! This leads me to the conclusion that either (a) atomic modification by a Kleisli map is impossible and another mechanism needs to ensure thread safety, or (b) we need to keep two versions of the state: one private, pure s and one public TVar s which we synchronize from time to time. The latter can be accomplished with atomic writes. More on that below. Isaac Elliott wrote:

...

What about creating a MonadState instance for the TVar? Very neat, but it would not solve the atomicity problem, since the update I have is not expressible as stateTVar.

YueCompl wrote:

...

I believe this is the use case for TMVar, use `takeTMVar` / `putTMVar` instead of `readTVar` / `writeTVar` will do.

TMVars appear non-empty for one thread only, as far as I understand. So any observer thread would be blocked while the worker thread is updating the state? Not good for my use case. The observer thread will be a webserver. I'd rather have the webserver report old states than block. I have neglected to mention one aspect of my problem that might be essential: All my state updates are monoidal. By that I mean that s is a monoid with s' <> s <> s = s' <> s and any effectful modification will be of the form f = \s -> (k s) >>= (\s' -> s' <> s) for some k :: s -> m s. This entails that instead of emitting modifications as functions I can emit modifications as values. emitModification :: (Monad m, Monoid s) => StateT s m a -> WriterT s (StateT s m) a emitModification (StateT f) = WriterT $ StateT $ \s -> do (a,s') <- f s return ((a,s'),s') broadcastModification :: (MonadIO m, Semigroup s) => TVar s -> (WriterT s m) a -> m a broadcastModification ref (WriterT m) = do (a,s') <- m (liftIO.atomically) (modifyTVar ref (\s -> s' <> s)) return a Then \ref -> broadcastModification ref . emitModification :: (MonadIO m, Monoid s) => TVar s -> StateT s m a -> StateT s m a This has the additional charm that I can explicitly control when writes to the TVar happen, by choosing which (StateT s m) blocks to enclose in this wrapper. I might even set up my communication as a TChan, if the observing thread maintains an own copy of state itself to merge the updates s' into. Olaf

Show replies by date

YueCompl

2 Sep 2 Sep

noon

New subject: broadcasting stateful computations

Um, I'm not sure I understand your case right, but if the "mutation" instead of the "mutated result" can be (might non-trivially) computed from a possibly outdated state, and the "mutation" can be trivially applied, I think `modifyTVar'` is the way to go. `readTVar` can be used to obtain an almost up-to-date state on demand, at low frequency.

...

On 2021-09-02, at 23:25, Olaf Klinke wrote:

Thanks everyone for the helpful suggestions!

Bryan Richer wrote:

...
what if the TVar is part of the state s? (StateT IO) can't give any guarantees about when or how the state is broadcast, but using STM within StateT actions still can.

I don't understand this, but it may go into the same direction as:

Chris Smith wrote:

...
I had the same suggestion. However, you then no longer need StateT, as ReaderT is enough. Using a `ReaderT (TVar s) IO a` will allow atomic changes to the state s, along with interleaved IO when it's done safely rather than in the middle of a transaction.

ReaderT (TVar s) IO is interesting! It does not solve the problem of atomic modifications with side-effects, though. Currently I am implicity using ReaderT (TVar s), by explicitly passing the TVar around.

In principle, for atomic modifications I have to get hold of a pure version (s -> s) of the modification. Maybe within a monadic context. But if determining this modification function is side-effectful, then it can not depend on the initial contents of the TVar!

This leads me to the conclusion that either (a) atomic modification by a Kleisli map is impossible and another mechanism needs to ensure thread safety, or (b) we need to keep two versions of the state: one private, pure s and one public TVar s which we synchronize from time to time. The latter can be accomplished with atomic writes. More on that below.

Isaac Elliott wrote:

...
What about creating a MonadState instance for the TVar? Very neat, but it would not solve the atomicity problem, since the update I have is not expressible as stateTVar.

YueCompl wrote:

...
I believe this is the use case for TMVar, use `takeTMVar` / `putTMVar` instead of `readTVar` / `writeTVar` will do.

TMVars appear non-empty for one thread only, as far as I understand. So any observer thread would be blocked while the worker thread is updating the state? Not good for my use case. The observer thread will be a webserver. I'd rather have the webserver report old states than block.

I have neglected to mention one aspect of my problem that might be essential: All my state updates are monoidal. By that I mean that s is a monoid with s' <> s <> s = s' <> s and any effectful modification will be of the form f = \s -> (k s) >>= (\s' -> s' <> s) for some k :: s -> m s. This entails that instead of emitting modifications as functions I can emit modifications as values.

emitModification :: (Monad m, Monoid s) => StateT s m a -> WriterT s (StateT s m) a emitModification (StateT f) = WriterT $ StateT $ \s -> do (a,s') <- f s return ((a,s'),s')

broadcastModification :: (MonadIO m, Semigroup s) => TVar s -> (WriterT s m) a -> m a broadcastModification ref (WriterT m) = do (a,s') <- m (liftIO.atomically) (modifyTVar ref (\s -> s' <> s)) return a

Then \ref -> broadcastModification ref . emitModification :: (MonadIO m, Monoid s) => TVar s -> StateT s m a -> StateT s m a

This has the additional charm that I can explicitly control when writes to the TVar happen, by choosing which (StateT s m) blocks to enclose in this wrapper. I might even set up my communication as a TChan, if the observing thread maintains an own copy of state itself to merge the updates s' into.

Olaf

_______________________________________________ Haskell-Cafe mailing list To (un)subscribe, modify options or view archives go to: http://mail.haskell.org/cgi-bin/mailman/listinfo/haskell-cafe Only members subscribed via the mailman list are allowed to post.

Olaf Klinke

1:42 p.m.

New subject: broadcasting stateful computations

On Fri, 2021-09-03 at 00:00 +0800, YueCompl wrote:

...

Um, I'm not sure I understand your case right, but if the "mutation" instead of the "mutated result" can be (might non-trivially) computed from a possibly outdated state, and the "mutation" can be trivially applied, I think `modifyTVar'` is the way to go. `readTVar` can be used to obtain an almost up-to-date state on demand, at low frequency.

To be concrete, my state is a collection of time stamped values, where the monoid operation overwrites old values with new ones. But I need to know the current state (x,t) to determine the "mutation", because I'll be asking questions like "server, tell me if there is a value of x newer than t." Any observer whose initial state is synchronized with the worker thread can in principle re-construct the worker's internal state by observing the stream of emitted "mutations". The most general abstraction would be that of a monoid action on a type, but in my case the monoid (mutations) and the mutated type are identical. act :: m -> a -> a act memtpy = id act (x <> y) = act x . act y -- monoid homomorphism act (x <> x) = act x -- idempotent Olaf

Isaac Elliott

6:29 p.m.

New subject: broadcasting stateful computations

How about using a read-write lock? data Locked a = Locked { lockVar :: TVar Bool, valueVar :: TVar a } new :: a -> IO (Locked a) new val = Locked <$> newTVarIO False <*> newTVarIO val -- lock blocks until unlocked lock, unlock :: Locked a -> STM () -- never blocks read :: Locked a -> STM a -- blocks until unlocked write :: Locked a -> a -> STM () When you want to apply your effectful state trasition `f :: s -> m s`: l <- ask s <- liftIO . atomically $ lock l *> read l s' <- f s liftIO . atomically $ write l s' *> unlock l On Fri, 3 Sep 2021, 3:46 am Olaf Klinke, wrote:

...

...
Um, I'm not sure I understand your case right, but if the "mutation" instead of the "mutated result" can be (might non-trivially) computed from a possibly outdated state, and the "mutation" can be trivially applied, I

On Fri, 2021-09-03 at 00:00 +0800, YueCompl wrote: think `modifyTVar'` is the way to go. `readTVar` can be used to obtain an almost up-to-date state on demand, at low frequency.

To be concrete, my state is a collection of time stamped values, where the monoid operation overwrites old values with new ones. But I need to know the current state (x,t) to determine the "mutation", because I'll be asking questions like "server, tell me if there is a value of x newer than t." Any observer whose initial state is synchronized with the worker thread can in principle re-construct the worker's internal state by observing the stream of emitted "mutations".

The most general abstraction would be that of a monoid action on a type, but in my case the monoid (mutations) and the mutated type are identical.

act :: m -> a -> a act memtpy = id act (x <> y) = act x . act y -- monoid homomorphism act (x <> x) = act x -- idempotent

Olaf

_______________________________________________ Haskell-Cafe mailing list To (un)subscribe, modify options or view archives go to: http://mail.haskell.org/cgi-bin/mailman/listinfo/haskell-cafe Only members subscribed via the mailman list are allowed to post.

YueCompl

3 Sep 3 Sep

4:26 a.m.

New subject: broadcasting stateful computations

It's a bit sad that I'm not so mathematically minded to understand you in that abstract level. But I have a more imperative solution in my mind, wrt the question:

...

"server, tell me if there is a value of x newer than t."

and do further mutate-or-giveup, like this: data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a } type ValueSink a = TMVar (ValueNode a) type Timestamp = Int seekTail :: ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail sink = go sink Nothing where go ref ancestor = tryReadTMVar ref >>= \case Nothing -> return (ref, ancestor) Just self@(ValueNode _ _ nxt) -> go nxt $ Just self updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m () updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt Just (ValueNode seenVal seenTs seenNxt) -> do (myVal, myTs) <- f $ Just (seenVal, seenTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt liftIO $ atomically $ putAsNewTailOrDiscard seenNxt -- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...

On 2021-09-03, at 01:42, Olaf Klinke wrote:

On Fri, 2021-09-03 at 00:00 +0800, YueCompl wrote:

...
Um, I'm not sure I understand your case right, but if the "mutation" instead of the "mutated result" can be (might non-trivially) computed from a possibly outdated state, and the "mutation" can be trivially applied, I think `modifyTVar'` is the way to go. `readTVar` can be used to obtain an almost up-to-date state on demand, at low frequency.

To be concrete, my state is a collection of time stamped values, where the monoid operation overwrites old values with new ones. But I need to know the current state (x,t) to determine the "mutation", because I'll be asking questions like "server, tell me if there is a value of x newer than t." Any observer whose initial state is synchronized with the worker thread can in principle re-construct the worker's internal state by observing the stream of emitted "mutations".

The most general abstraction would be that of a monoid action on a type, but in my case the monoid (mutations) and the mutated type are identical.

act :: m -> a -> a act memtpy = id act (x <> y) = act x . act y -- monoid homomorphism act (x <> x) = act x -- idempotent

Olaf

YueCompl

4 Sep 4 Sep

6:42 a.m.

New subject: broadcasting stateful computations

I'd like to add a new feature that you can fold the historic value stream in deriving the new state value, then it becomes: data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a } type ValueSink a = TMVar (ValueNode a) type Timestamp = Int seekTail :: forall a. (a -> Timestamp -> a -> Timestamp -> a) -> ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail f sink = go sink Nothing where go :: ValueSink a -> Maybe (ValueNode a) -> STM (ValueSink a, Maybe (ValueNode a)) go ref prevNode = tryReadTMVar ref >>= \case Nothing -> return (ref, prevNode) Just self@(ValueNode spotVal spotTs nxt) -> go nxt $ Just self { node'value = case prevNode of Nothing -> spotVal Just (ValueNode prevVal prevTs _prevNxt) -> f prevVal prevTs spotVal spotTs } updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m () updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail justLatest sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt Just (ValueNode spotVal spotTs spotNxt) -> do (myVal, myTs) <- f $ Just (spotVal, spotTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt liftIO $ atomically $ putAsNewTailOrDiscard spotNxt where justLatest :: (a -> Timestamp -> a -> Timestamp -> a) justLatest _prevVal _prevTs spotVal _spotTs = spotVal -- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...

On 2021-09-03, at 16:26, YueCompl wrote:

It's a bit sad that I'm not so mathematically minded to understand you in that abstract level. But I have a more imperative solution in my mind, wrt the question:

...
"server, tell me if there is a value of x newer than t."

and do further mutate-or-giveup, like this:

data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a }

type ValueSink a = TMVar (ValueNode a)

type Timestamp = Int

seekTail :: ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail sink = go sink Nothing where go ref ancestor = tryReadTMVar ref >>= \case Nothing -> return (ref, ancestor) Just self@(ValueNode _ _ nxt) -> go nxt $ Just self

updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m () updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt Just (ValueNode seenVal seenTs seenNxt) -> do (myVal, myTs) <- f $ Just (seenVal, seenTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt

putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt

liftIO $ atomically $ putAsNewTailOrDiscard seenNxt

-- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...
On 2021-09-03, at 01:42, Olaf Klinke mailto:olf@aatal-apotheke.de> wrote:

On Fri, 2021-09-03 at 00:00 +0800, YueCompl wrote:

...
Um, I'm not sure I understand your case right, but if the "mutation" instead of the "mutated result" can be (might non-trivially) computed from a possibly outdated state, and the "mutation" can be trivially applied, I think `modifyTVar'` is the way to go. `readTVar` can be used to obtain an almost up-to-date state on demand, at low frequency.

To be concrete, my state is a collection of time stamped values, where the monoid operation overwrites old values with new ones. But I need to know the current state (x,t) to determine the "mutation", because I'll be asking questions like "server, tell me if there is a value of x newer than t." Any observer whose initial state is synchronized with the worker thread can in principle re-construct the worker's internal state by observing the stream of emitted "mutations".

The most general abstraction would be that of a monoid action on a type, but in my case the monoid (mutations) and the mutated type are identical.

act :: m -> a -> a act memtpy = id act (x <> y) = act x . act y -- monoid homomorphism act (x <> x) = act x -- idempotent

Olaf

YueCompl

6:50 a.m.

New subject: broadcasting stateful computations

Oh, a bugfix: the new tail reference should be returned after update, so a thread local stream reference can technically be cached. And I realize this is more than you originally need, never mind if it's not so useful to you. data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a } type ValueSink a = TMVar (ValueNode a) type Timestamp = Int seekTail :: forall a. (a -> Timestamp -> a -> Timestamp -> a) -> ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail f sink = go sink Nothing where go :: ValueSink a -> Maybe (ValueNode a) -> STM (ValueSink a, Maybe (ValueNode a)) go ref prevNode = tryReadTMVar ref >>= \case Nothing -> return (ref, prevNode) Just self@(ValueNode spotVal spotTs nxt) -> go nxt $ Just self { node'value = case prevNode of Nothing -> spotVal Just (ValueNode prevVal prevTs _prevNxt) -> f prevVal prevTs spotVal spotTs } updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m (ValueSink a) updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail justLatest sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt return nxt Just (ValueNode spotVal spotTs spotNxt) -> do (myVal, myTs) <- f $ Just (spotVal, spotTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt liftIO $ atomically $ putAsNewTailOrDiscard spotNxt return spotNxt where justLatest :: (a -> Timestamp -> a -> Timestamp -> a) justLatest _prevVal _prevTs spotVal _spotTs = spotVal -- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...

On 2021-09-04, at 18:42, YueCompl wrote:

I'd like to add a new feature that you can fold the historic value stream in deriving the new state value, then it becomes:

data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a }

type ValueSink a = TMVar (ValueNode a)

type Timestamp = Int

seekTail :: forall a. (a -> Timestamp -> a -> Timestamp -> a) -> ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail f sink = go sink Nothing where go :: ValueSink a -> Maybe (ValueNode a) -> STM (ValueSink a, Maybe (ValueNode a)) go ref prevNode = tryReadTMVar ref >>= \case Nothing -> return (ref, prevNode) Just self@(ValueNode spotVal spotTs nxt) -> go nxt $ Just self { node'value = case prevNode of Nothing -> spotVal Just (ValueNode prevVal prevTs _prevNxt) -> f prevVal prevTs spotVal spotTs }

updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m () updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail justLatest sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt Just (ValueNode spotVal spotTs spotNxt) -> do (myVal, myTs) <- f $ Just (spotVal, spotTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt

putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt

liftIO $ atomically $ putAsNewTailOrDiscard spotNxt where justLatest :: (a -> Timestamp -> a -> Timestamp -> a) justLatest _prevVal _prevTs spotVal _spotTs = spotVal

-- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...
On 2021-09-03, at 16:26, YueCompl mailto:compl.yue@icloud.com> wrote:

It's a bit sad that I'm not so mathematically minded to understand you in that abstract level. But I have a more imperative solution in my mind, wrt the question:

...
"server, tell me if there is a value of x newer than t."

and do further mutate-or-giveup, like this:

data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a }

type ValueSink a = TMVar (ValueNode a)

type Timestamp = Int

seekTail :: ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail sink = go sink Nothing where go ref ancestor = tryReadTMVar ref >>= \case Nothing -> return (ref, ancestor) Just self@(ValueNode _ _ nxt) -> go nxt $ Just self

updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m () updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt Just (ValueNode seenVal seenTs seenNxt) -> do (myVal, myTs) <- f $ Just (seenVal, seenTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt

putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt

liftIO $ atomically $ putAsNewTailOrDiscard seenNxt

-- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...
On 2021-09-03, at 01:42, Olaf Klinke mailto:olf@aatal-apotheke.de> wrote:

On Fri, 2021-09-03 at 00:00 +0800, YueCompl wrote:

...
Um, I'm not sure I understand your case right, but if the "mutation" instead of the "mutated result" can be (might non-trivially) computed from a possibly outdated state, and the "mutation" can be trivially applied, I think `modifyTVar'` is the way to go. `readTVar` can be used to obtain an almost up-to-date state on demand, at low frequency.

To be concrete, my state is a collection of time stamped values, where the monoid operation overwrites old values with new ones. But I need to know the current state (x,t) to determine the "mutation", because I'll be asking questions like "server, tell me if there is a value of x newer than t." Any observer whose initial state is synchronized with the worker thread can in principle re-construct the worker's internal state by observing the stream of emitted "mutations".

The most general abstraction would be that of a monoid action on a type, but in my case the monoid (mutations) and the mutated type are identical.

act :: m -> a -> a act memtpy = id act (x <> y) = act x . act y -- monoid homomorphism act (x <> x) = act x -- idempotent

Olaf

Olaf Klinke

6 Sep 6 Sep

5:04 a.m.

New subject: broadcasting stateful computations

Dear Compl, thanks for putting so much thought into this. Regarding your suggestion, I'm afraid I misled you somehow or I don't understand the purpose of your code. I don't need the entire history of each value, only the most recent one. Remark: The ValueNode looks like a ListT (like in list-t package) over the STM monad. We're developing a gateway between two data protocols. We have a Python implementation, but it is not parallel enough and message parsing is too slow. Therefore we hired MLabs to implement the source protocol in Haskell [1]. The gateway does the following. 1. Parent thread reads a config and forks several concurrent child threads, each talking to a different server. The state of the children is read through TVars. 2. The child thread polls data from its assigned server, decodes the message and executes pre-defined callbacks [2], which are functions type Callback m = Value -> UTCTime -> m (). 3. Each data packet can contain several data items and I want these callbacks to be executed in parallel, too. This is because we expect the network IO to be the slowest part in the entire process. I use the parallel-io package for that. import Control.Monad.Reader import Control.Monad.IO.Unlift (MonadUnliftIO(..)) import Control.Concurrent.ParallelIO.Local (Pool,parallel_) type CallbackStrategy m n = forall f. Foldable f => f (m ()) -> n () sequentialStrategy :: Applicative m => CallbackStrategy m m sequentialStrategy = Data.Foldable.sequenceA_ -- ^ just sequence the callbacks concurrentStrategy :: MonadUnliftIO m => CallbackStrategy m (ReaderT Pool m) concurrentStrategy = execConcurrently . Data.Foldable.toList where execConcurrently actionlist = ReaderT (\pool -> withRunInIO (\asIO -> parallel_ pool (map asIO actionlist))) -- ^ execute the callbacks concurrently on a pool of worker threads I'd be quite happy with m = StateT s IO where s holds the most recent time stamp and value for each data address. The simplest way to broadcast state in this monad would be: broadcast :: Monoid s => TVar s -> StateT s IO () -> StateT s IO () broadcast var (StateT f) = StateT (\s -> do s' <- f s atomically (modifyTVar ref (\s -> s <> s'))) That is, we run the StateT action and send the changes to the TVar afterwards. But there are two more problems with this: (1) The concurrent callbacks all want to atomically modify the same TVar. This is a concurrency bottleneck and effectively implements Isaac Elliott's Locks. Hence it seems that giving each data address its own value TVar could allow more parallelism. (2) StateT is inherently sequential, which defeats concurrentStrategy. In fact the unliftio-core package states:

...

Note that, in order to meet the laws given below, the intuition is that a monad must have no monadic state, but may have monadic context. This essentially limits MonadUnliftIO to ReaderT and IdentityT transformers on top of IO. Intuitively, there is no way to safely and concurrently modify the same state. Nice example of how the type system guides you towards doing the right thing.

Therefore I will go back to what Chris Smith and Bryan Richter suggested, and use ReaderT (TVar s) individually for each value. And your suggestion of TMVars is also good in finer granularity, I think. In the following, m can have a MonadUnliftIO instance. unStateT :: MonadIO m => TMVar s -> Callback (StateT s m) -> Callback m unStateT var cb = \value time -> do before <- (liftIO.atomically) (takeTMVar var) -- TMVar is now empty, no other thread can access it after <- execStateT (cb value time) before (liftIO.atomically) (putTMVar var after) -- TMVar now full again Thanks everyone for helping! Olaf [1] https://github.com/mlabs-haskell/opc-xml-da-client [2] I suppose in FP-land we'd name these continuations, not callbacks. Not sure whether the continuation monad abstraction buys me any advantage here, though. On Sat, 2021-09-04 at 18:50 +0800, YueCompl wrote:

...

Oh, a bugfix: the new tail reference should be returned after update, so a thread local stream reference can technically be cached.

And I realize this is more than you originally need, never mind if it's not so useful to you.

data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a }

type ValueSink a = TMVar (ValueNode a)

type Timestamp = Int

seekTail :: forall a. (a -> Timestamp -> a -> Timestamp -> a) -> ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail f sink = go sink Nothing where go :: ValueSink a -> Maybe (ValueNode a) -> STM (ValueSink a, Maybe (ValueNode a)) go ref prevNode = tryReadTMVar ref >>= \case Nothing -> return (ref, prevNode) Just self@(ValueNode spotVal spotTs nxt) -> go nxt $ Just self { node'value = case prevNode of Nothing -> spotVal Just (ValueNode prevVal prevTs _prevNxt) -> f prevVal prevTs spotVal spotTs }

updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m (ValueSink a) updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail justLatest sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt return nxt Just (ValueNode spotVal spotTs spotNxt) -> do (myVal, myTs) <- f $ Just (spotVal, spotTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt

putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt

liftIO $ atomically $ putAsNewTailOrDiscard spotNxt return spotNxt where justLatest :: (a -> Timestamp -> a -> Timestamp -> a) justLatest _prevVal _prevTs spotVal _spotTs = spotVal

-- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...
On 2021-09-04, at 18:42, YueCompl wrote:

I'd like to add a new feature that you can fold the historic value stream in deriving the new state value, then it becomes:

data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a }

type ValueSink a = TMVar (ValueNode a)

type Timestamp = Int

seekTail :: forall a. (a -> Timestamp -> a -> Timestamp -> a) -> ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail f sink = go sink Nothing where go :: ValueSink a -> Maybe (ValueNode a) -> STM (ValueSink a, Maybe (ValueNode a)) go ref prevNode = tryReadTMVar ref >>= \case Nothing -> return (ref, prevNode) Just self@(ValueNode spotVal spotTs nxt) -> go nxt $ Just self { node'value = case prevNode of Nothing -> spotVal Just (ValueNode prevVal prevTs _prevNxt) -> f prevVal prevTs spotVal spotTs }

updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m () updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail justLatest sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt Just (ValueNode spotVal spotTs spotNxt) -> do (myVal, myTs) <- f $ Just (spotVal, spotTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt

putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt

liftIO $ atomically $ putAsNewTailOrDiscard spotNxt where justLatest :: (a -> Timestamp -> a -> Timestamp -> a) justLatest _prevVal _prevTs spotVal _spotTs = spotVal

-- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...
On 2021-09-03, at 16:26, YueCompl mailto:compl.yue@icloud.com> wrote:

It's a bit sad that I'm not so mathematically minded to understand you in that abstract level. But I have a more imperative solution in my mind, wrt the question:

...
"server, tell me if there is a value of x newer than t."

and do further mutate-or-giveup, like this:

data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a }

type ValueSink a = TMVar (ValueNode a)

type Timestamp = Int

seekTail :: ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail sink = go sink Nothing where go ref ancestor = tryReadTMVar ref >>= \case Nothing -> return (ref, ancestor) Just self@(ValueNode _ _ nxt) -> go nxt $ Just self

updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m () updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt Just (ValueNode seenVal seenTs seenNxt) -> do (myVal, myTs) <- f $ Just (seenVal, seenTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt

putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt

liftIO $ atomically $ putAsNewTailOrDiscard seenNxt

-- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...
On 2021-09-03, at 01:42, Olaf Klinke mailto:olf@aatal-apotheke.de> wrote:

On Fri, 2021-09-03 at 00:00 +0800, YueCompl wrote:

...
Um, I'm not sure I understand your case right, but if the "mutation" instead of the "mutated result" can be (might non-trivially) computed from a possibly outdated state, and the "mutation" can be trivially applied, I think `modifyTVar'` is the way to go. `readTVar` can be used to obtain an almost up-to-date state on demand, at low frequency.

To be concrete, my state is a collection of time stamped values, where the monoid operation overwrites old values with new ones. But I need to know the current state (x,t) to determine the "mutation", because I'll be asking questions like "server, tell me if there is a value of x newer than t." Any observer whose initial state is synchronized with the worker thread can in principle re-construct the worker's internal state by observing the stream of emitted "mutations".

The most general abstraction would be that of a monoid action on a type, but in my case the monoid (mutations) and the mutated type are identical.

act :: m -> a -> a act memtpy = id act (x <> y) = act x . act y -- monoid homomorphism act (x <> x) = act x -- idempotent

Olaf

YueCompl

6:27 a.m.

New subject: broadcasting stateful computations

I happen to be pondering with ideas about mutable yet shared data/state, unlike the actor model, a Turing machine has no notation of external change notification, I wonder this limitation propagates to lambda calculus, so there seem be no approved ways in pure functional paradigm, to handle concurrent mutation on shared state, since it's not modeled in the first place. From your updated description, I guess `atomicModifyIORef` (or maybe the strict version `atomicModifyIORef'` more desirable?) might work for you as well, and it's further more performant than 'TVar' based. Btw, just come to my mind about the timestamps, if they are of high precision and come from different server nodes, you would not trust their total ordering, since small drifts are allowed/inevitable even the clocks are actively synchronized.

...

On 2021-09-06, at 17:04, Olaf Klinke wrote:

Dear Compl,

thanks for putting so much thought into this. Regarding your suggestion, I'm afraid I misled you somehow or I don't understand the purpose of your code. I don't need the entire history of each value, only the most recent one. Remark: The ValueNode looks like a ListT (like in list-t package) over the STM monad.

We're developing a gateway between two data protocols. We have a Python implementation, but it is not parallel enough and message parsing is too slow. Therefore we hired MLabs to implement the source protocol in Haskell [1]. The gateway does the following.

1. Parent thread reads a config and forks several concurrent child threads, each talking to a different server. The state of the children is read through TVars. 2. The child thread polls data from its assigned server, decodes the message and executes pre-defined callbacks [2], which are functions type Callback m = Value -> UTCTime -> m (). 3. Each data packet can contain several data items and I want these callbacks to be executed in parallel, too. This is because we expect the network IO to be the slowest part in the entire process. I use the parallel-io package for that.

import Control.Monad.Reader import Control.Monad.IO.Unlift (MonadUnliftIO(..)) import Control.Concurrent.ParallelIO.Local (Pool,parallel_)

type CallbackStrategy m n = forall f. Foldable f => f (m ()) -> n () sequentialStrategy :: Applicative m => CallbackStrategy m m sequentialStrategy = Data.Foldable.sequenceA_ -- ^ just sequence the callbacks

concurrentStrategy :: MonadUnliftIO m => CallbackStrategy m (ReaderT Pool m) concurrentStrategy = execConcurrently . Data.Foldable.toList where execConcurrently actionlist = ReaderT (\pool -> withRunInIO (\asIO -> parallel_ pool (map asIO actionlist))) -- ^ execute the callbacks concurrently on a pool of worker threads

I'd be quite happy with m = StateT s IO where s holds the most recent time stamp and value for each data address. The simplest way to broadcast state in this monad would be:

broadcast :: Monoid s => TVar s -> StateT s IO () -> StateT s IO () broadcast var (StateT f) = StateT (\s -> do s' <- f s atomically (modifyTVar ref (\s -> s <> s')))

That is, we run the StateT action and send the changes to the TVar afterwards. But there are two more problems with this:

(1) The concurrent callbacks all want to atomically modify the same TVar. This is a concurrency bottleneck and effectively implements Isaac Elliott's Locks. Hence it seems that giving each data address its own value TVar could allow more parallelism.

(2) StateT is inherently sequential, which defeats concurrentStrategy. In fact the unliftio-core package states:

...
Note that, in order to meet the laws given below, the intuition is that a monad must have no monadic state, but may have monadic context. This essentially limits MonadUnliftIO to ReaderT and IdentityT transformers on top of IO. Intuitively, there is no way to safely and concurrently modify the same state. Nice example of how the type system guides you towards doing the right thing.

Therefore I will go back to what Chris Smith and Bryan Richter suggested, and use ReaderT (TVar s) individually for each value. And your suggestion of TMVars is also good in finer granularity, I think. In the following, m can have a MonadUnliftIO instance.

unStateT :: MonadIO m => TMVar s -> Callback (StateT s m) -> Callback m unStateT var cb = \value time -> do before <- (liftIO.atomically) (takeTMVar var) -- TMVar is now empty, no other thread can access it after <- execStateT (cb value time) before (liftIO.atomically) (putTMVar var after) -- TMVar now full again

Thanks everyone for helping! Olaf

[1] https://github.com/mlabs-haskell/opc-xml-da-client https://github.com/mlabs-haskell/opc-xml-da-client [2] I suppose in FP-land we'd name these continuations, not callbacks. Not sure whether the continuation monad abstraction buys me any advantage here, though.

On Sat, 2021-09-04 at 18:50 +0800, YueCompl wrote:

...
Oh, a bugfix: the new tail reference should be returned after update, so a thread local stream reference can technically be cached.

And I realize this is more than you originally need, never mind if it's not so useful to you.

data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a }

type ValueSink a = TMVar (ValueNode a)

type Timestamp = Int

seekTail :: forall a. (a -> Timestamp -> a -> Timestamp -> a) -> ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail f sink = go sink Nothing where go :: ValueSink a -> Maybe (ValueNode a) -> STM (ValueSink a, Maybe (ValueNode a)) go ref prevNode = tryReadTMVar ref >>= \case Nothing -> return (ref, prevNode) Just self@(ValueNode spotVal spotTs nxt) -> go nxt $ Just self { node'value = case prevNode of Nothing -> spotVal Just (ValueNode prevVal prevTs _prevNxt) -> f prevVal prevTs spotVal spotTs }

updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m (ValueSink a) updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail justLatest sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt return nxt Just (ValueNode spotVal spotTs spotNxt) -> do (myVal, myTs) <- f $ Just (spotVal, spotTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt

putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt

liftIO $ atomically $ putAsNewTailOrDiscard spotNxt return spotNxt where justLatest :: (a -> Timestamp -> a -> Timestamp -> a) justLatest _prevVal _prevTs spotVal _spotTs = spotVal

-- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...
On 2021-09-04, at 18:42, YueCompl wrote:

I'd like to add a new feature that you can fold the historic value stream in deriving the new state value, then it becomes:

data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a }

type ValueSink a = TMVar (ValueNode a)

type Timestamp = Int

seekTail :: forall a. (a -> Timestamp -> a -> Timestamp -> a) -> ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail f sink = go sink Nothing where go :: ValueSink a -> Maybe (ValueNode a) -> STM (ValueSink a, Maybe (ValueNode a)) go ref prevNode = tryReadTMVar ref >>= \case Nothing -> return (ref, prevNode) Just self@(ValueNode spotVal spotTs nxt) -> go nxt $ Just self { node'value = case prevNode of Nothing -> spotVal Just (ValueNode prevVal prevTs _prevNxt) -> f prevVal prevTs spotVal spotTs }

updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m () updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail justLatest sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt Just (ValueNode spotVal spotTs spotNxt) -> do (myVal, myTs) <- f $ Just (spotVal, spotTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt

putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt

liftIO $ atomically $ putAsNewTailOrDiscard spotNxt where justLatest :: (a -> Timestamp -> a -> Timestamp -> a) justLatest _prevVal _prevTs spotVal _spotTs = spotVal

-- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...
On 2021-09-03, at 16:26, YueCompl mailto:compl.yue@icloud.com mailto:compl.yue@icloud.com>> wrote:

It's a bit sad that I'm not so mathematically minded to understand you in that abstract level. But I have a more imperative solution in my mind, wrt the question:

...
"server, tell me if there is a value of x newer than t."

and do further mutate-or-giveup, like this:

data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a }

type ValueSink a = TMVar (ValueNode a)

type Timestamp = Int

seekTail :: ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail sink = go sink Nothing where go ref ancestor = tryReadTMVar ref >>= \case Nothing -> return (ref, ancestor) Just self@(ValueNode _ _ nxt) -> go nxt $ Just self

updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m () updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt Just (ValueNode seenVal seenTs seenNxt) -> do (myVal, myTs) <- f $ Just (seenVal, seenTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt

putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt

liftIO $ atomically $ putAsNewTailOrDiscard seenNxt

-- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...
On 2021-09-03, at 01:42, Olaf Klinke mailto:olf@aatal-apotheke.de mailto:olf@aatal-apotheke.de>> wrote:

On Fri, 2021-09-03 at 00:00 +0800, YueCompl wrote:

...
Um, I'm not sure I understand your case right, but if the "mutation" instead of the "mutated result" can be (might non-trivially) computed from a possibly outdated state, and the "mutation" can be trivially applied, I think `modifyTVar'` is the way to go. `readTVar` can be used to obtain an almost up-to-date state on demand, at low frequency.

To be concrete, my state is a collection of time stamped values, where the monoid operation overwrites old values with new ones. But I need to know the current state (x,t) to determine the "mutation", because I'll be asking questions like "server, tell me if there is a value of x newer than t." Any observer whose initial state is synchronized with the worker thread can in principle re-construct the worker's internal state by observing the stream of emitted "mutations".

The most general abstraction would be that of a monoid action on a type, but in my case the monoid (mutations) and the mutated type are identical.

act :: m -> a -> a act memtpy = id act (x <> y) = act x . act y -- monoid homomorphism act (x <> x) = act x -- idempotent

Olaf

Olaf Klinke

12:10 p.m.

New subject: broadcasting stateful computations

On Mon, 2021-09-06 at 18:27 +0800, YueCompl wrote:

...

I happen to be pondering with ideas about mutable yet shared data/state, unlike the actor model, a Turing machine has no notation of external change notification, I wonder this limitation propagates to lambda calculus, so there seem be no approved ways in pure functional paradigm, to handle concurrent mutation on shared state, since it's not modeled in the first place.

Concurrency is hard. All we can do is provide shorthand notation for commonly used idioms. Consider another widely used shared state modifier: Version control systems like SVN or git. For true concurrency you need a way of conflict resolution, which your state may or may not support. Databases (and STM) have rollbacks in case of conflicts as a simple kind of conflict resolution. My conflict resolution now comprises fragmentation of the state into small data address components, akin to row locks in a database.

...

From your updated description, I guess `atomicModifyIORef` (or maybe the strict version `atomicModifyIORef'` more desirable?) might work for you as well, and it's further more performant than 'TVar' based.

I am a newbie to STM. I was hoping that STM provides some of the aforementioned commonly used idioms and is therefore safer to use than naked IORef. I do not want to worry about race conditions and deadlocks, the library should to that for me.

...

Btw, just come to my mind about the timestamps, if they are of high precision and come from different server nodes, you would not trust their total ordering, since small drifts are allowed/inevitable even the clocks are actively synchronized.

Indeed I do not trust the timestamps and I never compare timestamps obatained from different servers. However, I do trust that if I obtain a timestamp t0 from server A and later ask server A for a timestamp t1, then t1 >= t0 holds. Olaf

...

...
On 2021-09-06, at 17:04, Olaf Klinke wrote:

Dear Compl,

thanks for putting so much thought into this. Regarding your suggestion, I'm afraid I misled you somehow or I don't understand the purpose of your code. I don't need the entire history of each value, only the most recent one. Remark: The ValueNode looks like a ListT (like in list-t package) over the STM monad.

We're developing a gateway between two data protocols. We have a Python implementation, but it is not parallel enough and message parsing is too slow. Therefore we hired MLabs to implement the source protocol in Haskell [1]. The gateway does the following.

1. Parent thread reads a config and forks several concurrent child threads, each talking to a different server. The state of the children is read through TVars. 2. The child thread polls data from its assigned server, decodes the message and executes pre-defined callbacks [2], which are functions type Callback m = Value -> UTCTime -> m (). 3. Each data packet can contain several data items and I want these callbacks to be executed in parallel, too. This is because we expect the network IO to be the slowest part in the entire process. I use the parallel-io package for that.

import Control.Monad.Reader import Control.Monad.IO.Unlift (MonadUnliftIO(..)) import Control.Concurrent.ParallelIO.Local (Pool,parallel_)

type CallbackStrategy m n = forall f. Foldable f => f (m ()) -> n () sequentialStrategy :: Applicative m => CallbackStrategy m m sequentialStrategy = Data.Foldable.sequenceA_ -- ^ just sequence the callbacks

concurrentStrategy :: MonadUnliftIO m => CallbackStrategy m (ReaderT Pool m) concurrentStrategy = execConcurrently . Data.Foldable.toList where execConcurrently actionlist = ReaderT (\pool -> withRunInIO (\asIO -> parallel_ pool (map asIO actionlist))) -- ^ execute the callbacks concurrently on a pool of worker threads

I'd be quite happy with m = StateT s IO where s holds the most recent time stamp and value for each data address. The simplest way to broadcast state in this monad would be:

broadcast :: Monoid s => TVar s -> StateT s IO () -> StateT s IO () broadcast var (StateT f) = StateT (\s -> do s' <- f s atomically (modifyTVar ref (\s -> s <> s')))

That is, we run the StateT action and send the changes to the TVar afterwards. But there are two more problems with this:

(1) The concurrent callbacks all want to atomically modify the same TVar. This is a concurrency bottleneck and effectively implements Isaac Elliott's Locks. Hence it seems that giving each data address its own value TVar could allow more parallelism.

(2) StateT is inherently sequential, which defeats concurrentStrategy. In fact the unliftio-core package states:

...
Note that, in order to meet the laws given below, the intuition is that a monad must have no monadic state, but may have monadic context. This essentially limits MonadUnliftIO to ReaderT and IdentityT transformers on top of IO. Intuitively, there is no way to safely and concurrently modify the same state. Nice example of how the type system guides you towards doing the right thing.

Therefore I will go back to what Chris Smith and Bryan Richter suggested, and use ReaderT (TVar s) individually for each value. And your suggestion of TMVars is also good in finer granularity, I think. In the following, m can have a MonadUnliftIO instance.

unStateT :: MonadIO m => TMVar s -> Callback (StateT s m) -> Callback m unStateT var cb = \value time -> do before <- (liftIO.atomically) (takeTMVar var) -- TMVar is now empty, no other thread can access it after <- execStateT (cb value time) before (liftIO.atomically) (putTMVar var after) -- TMVar now full again

Thanks everyone for helping! Olaf

[1] https://github.com/mlabs-haskell/opc-xml-da-client https://github.com/mlabs-haskell/opc-xml-da-client [2] I suppose in FP-land we'd name these continuations, not callbacks. Not sure whether the continuation monad abstraction buys me any advantage here, though.

On Sat, 2021-09-04 at 18:50 +0800, YueCompl wrote:

...
Oh, a bugfix: the new tail reference should be returned after update, so a thread local stream reference can technically be cached.

And I realize this is more than you originally need, never mind if it's not so useful to you.

data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a }

type ValueSink a = TMVar (ValueNode a)

type Timestamp = Int

seekTail :: forall a. (a -> Timestamp -> a -> Timestamp -> a) -> ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail f sink = go sink Nothing where go :: ValueSink a -> Maybe (ValueNode a) -> STM (ValueSink a, Maybe (ValueNode a)) go ref prevNode = tryReadTMVar ref >>= \case Nothing -> return (ref, prevNode) Just self@(ValueNode spotVal spotTs nxt) -> go nxt $ Just self { node'value = case prevNode of Nothing -> spotVal Just (ValueNode prevVal prevTs _prevNxt) -> f prevVal prevTs spotVal spotTs }

updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m (ValueSink a) updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail justLatest sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt return nxt Just (ValueNode spotVal spotTs spotNxt) -> do (myVal, myTs) <- f $ Just (spotVal, spotTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt

putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt

liftIO $ atomically $ putAsNewTailOrDiscard spotNxt return spotNxt where justLatest :: (a -> Timestamp -> a -> Timestamp -> a) justLatest _prevVal _prevTs spotVal _spotTs = spotVal

-- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...
On 2021-09-04, at 18:42, YueCompl wrote:

I'd like to add a new feature that you can fold the historic value stream in deriving the new state value, then it becomes:

data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a }

type ValueSink a = TMVar (ValueNode a)

type Timestamp = Int

seekTail :: forall a. (a -> Timestamp -> a -> Timestamp -> a) -> ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail f sink = go sink Nothing where go :: ValueSink a -> Maybe (ValueNode a) -> STM (ValueSink a, Maybe (ValueNode a)) go ref prevNode = tryReadTMVar ref >>= \case Nothing -> return (ref, prevNode) Just self@(ValueNode spotVal spotTs nxt) -> go nxt $ Just self { node'value = case prevNode of Nothing -> spotVal Just (ValueNode prevVal prevTs _prevNxt) -> f prevVal prevTs spotVal spotTs }

updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m () updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail justLatest sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt Just (ValueNode spotVal spotTs spotNxt) -> do (myVal, myTs) <- f $ Just (spotVal, spotTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt

putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt

liftIO $ atomically $ putAsNewTailOrDiscard spotNxt where justLatest :: (a -> Timestamp -> a -> Timestamp -> a) justLatest _prevVal _prevTs spotVal _spotTs = spotVal

-- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...
On 2021-09-03, at 16:26, YueCompl mailto:compl.yue@icloud.com mailto:compl.yue@icloud.com>> wrote:

It's a bit sad that I'm not so mathematically minded to understand you in that abstract level. But I have a more imperative solution in my mind, wrt the question:

...
"server, tell me if there is a value of x newer than t."

and do further mutate-or-giveup, like this:

data ValueNode a = ValueNode { node'value :: a, node'timestamp :: Timestamp, node'next :: ValueSink a }

type ValueSink a = TMVar (ValueNode a)

type Timestamp = Int

seekTail :: ValueSink a -> STM (ValueSink a, Maybe (ValueNode a)) seekTail sink = go sink Nothing where go ref ancestor = tryReadTMVar ref >>= \case Nothing -> return (ref, ancestor) Just self@(ValueNode _ _ nxt) -> go nxt $ Just self

updateValue :: forall a m. MonadIO m => (Maybe (a, Timestamp) -> m (a, Timestamp)) -> ValueSink a -> m () updateValue f sink = do (tailRef, tailNode) <- liftIO $ atomically $ seekTail sink case tailNode of Nothing -> do (myVal, myTs) <- f Nothing liftIO $ atomically $ do nxt <- newEmptyTMVar void $ tryPutTMVar tailRef $ ValueNode myVal myTs nxt Just (ValueNode seenVal seenTs seenNxt) -> do (myVal, myTs) <- f $ Just (seenVal, seenTs) newNxt <- liftIO newEmptyTMVarIO let newTail = ValueNode myVal myTs newNxt

putAsNewTailOrDiscard :: ValueSink a -> STM () putAsNewTailOrDiscard nodeRef = putTMVar nodeRef newTail `orElse` yetOther'sTail where yetOther'sTail = do (ValueNode _other'sVal other'sTs other'sNxt) <- readTMVar nodeRef if other'sTs >= myTs then return () else putAsNewTailOrDiscard other'sNxt

liftIO $ atomically $ putAsNewTailOrDiscard seenNxt

-- Each concurrent thread is supposed to have its local 'ValueSink' reference -- "cached" over time, but keep in mind that for any such thread who is slow -- in unfolding the value stream, the historical values will pile up in heap.

...
On 2021-09-03, at 01:42, Olaf Klinke mailto:olf@aatal-apotheke.de mailto:olf@aatal-apotheke.de>> wrote:

On Fri, 2021-09-03 at 00:00 +0800, YueCompl wrote: > Um, I'm not sure I understand your case right, but if the "mutation" instead of the "mutated result" can be (might non-trivially) computed from a possibly outdated state, and the "mutation" can be trivially applied, I think `modifyTVar'` is the way to go. `readTVar` can be used to obtain an almost up-to-date state on demand, at low frequency.

To be concrete, my state is a collection of time stamped values, where the monoid operation overwrites old values with new ones. But I need to know the current state (x,t) to determine the "mutation", because I'll be asking questions like "server, tell me if there is a value of x newer than t." Any observer whose initial state is synchronized with the worker thread can in principle re-construct the worker's internal state by observing the stream of emitted "mutations".

The most general abstraction would be that of a monoid action on a type, but in my case the monoid (mutations) and the mutated type are identical.

act :: m -> a -> a act memtpy = id act (x <> y) = act x . act y -- monoid homomorphism act (x <> x) = act x -- idempotent

Olaf

1518

Age (days ago)

1522

Last active (days ago)

List overview

Download

9 comments

3 participants

participants (3)

Isaac Elliott
Olaf Klinke
YueCompl

Re: [Haskell-cafe] broadcasting stateful computations

tags

participants (3)