IntersectMBO / IntersectMBO/typed-protocols
Experiment with a bearer which allows to avoid extra threads in a Driver
- Dominant language
- Haskell
- Stars
- 16
- Forks
- 7
- Avg merge
- 2d 23h
- Merged PRs (30d)
- 1
Description
We need a new `Channel` type:
```hs
data Channel m = Channel {
send :: LBS.ByteString -> m (),
recv :: STM m (Maybe LBS.ByteString)
}
```
The [Driver](https://github.com/input-output-hk/typed-protocols/blob/coot/typed-protocols-rewrite/typed-protocols/src/Network/TypedProtocol/Driver.hs?plain=1#L84-L152=) type can stay as is. However codec type is not general enough:
```hs
data Codec ps failure m bytes = Codec {
encode :: forall (st :: ps) (st' :: ps).
SingI st
=> ActiveState st
=> Message ps st st'
-> bytes,
decode :: forall (st :: ps).
ActiveState st
=> Sing st
-> m (DecodeStep bytes failure m (SomeMessage st))
}
```
We will need a codec which works in both `m` and `STM m`. `cborg` requires access to the `ST` operations, e.g. [mkCodecCborStrictST](https://github.com/input-output-hk/typed-protocols/blob/coot/typed-protocols-rewrite/typed-protocols-cborg/src/Network/TypedProtocol/Codec/CBOR.hs?plain=1#L50-L63=) but `STM` monad has no `MonadST` instance.
In `coot/typed-protocols-rewrite` branch we have:
```hs
runDecoderWithChannel :: MonadSTM m
=> Channel m bytes
-> Maybe bytes
-> DecodeStep bytes failure m a
-> m (Either failure (a, Maybe bytes))
tryRunDecoderWithChannel :: Monad m
=> Channel m bytes
-> Maybe bytes
-> DecodeStep bytes failure m (SomeMessage st)
-> m (Either failure
(Either (DriverState ps pr st bytes failure (Maybe bytes) m)
(SomeMessage st, Maybe bytes)))
```
because of the above constraint we cannot change its signature to `STM m`, but we can guarantee that all `recv`s are non blocking (e.g. ``atomically $ Just <$> recv `orElse` pure Nothing``).
The `tryRunDecoderWithChannel` one is used to implement the `tryRecvMessage` record field of `Driver`. And it is plausible to implement it with `recv :: STM m (Maybe ByteString)`)
```haskell
data Driver ps (pr :: PeerRole) bytes failure dstate m =
Driver {
...
tryRecvMessage :: forall (st :: ps).
SingI st
=> ActiveState st
=> ReflRelativeAgency (StateAgency st)
TheyHaveAgency
(Relative pr (StateAgency st))
-> DriverState ps pr st bytes failure dstate m
-> m (Either (DriverState ps pr st bytes failure dstate m)
( SomeMessage st
, dstate
))
, -- | Construct a non-blocking stm action which awaits for the
-- message.
--
recvMessageSTM :: forall (st :: ps).
SingI st
=> ActiveState st
=> ReflRelativeAgency (StateAgency st)
TheyHaveAgency
(Relative pr (StateAgency st))
-> DriverState ps pr st bytes failure dstate m
-> m (STM m (SomeMessage st, dstate))
, startDState :: dstate
}
```
The question is how we can implement `recvMessageSTM`. For that it seems that being able to run a decoder in the `STM` monad (without forking a thread) is indispensable.
`GHC` exposes [unsafeIOToSTM](https://hackage.haskell.org/package/base-4.16.1.0/docs/GHC-Conc.html#v:unsafeIOToSTM) which could be used to lift `ST` to `STM` (via `IO`), but this is rather dodgy way, so a different solution is needed. On the other hand, a rudimentary inspection of `cborg` library shows that `ST` is deeply grained, e.g.
* [Partial](https://hackage.haskell.org/package/cborg-0.2.7.0/docs/Codec-CBOR-Read.html#v:Partial)
* [DecodeAction](https://hackage.haskell.org/package/cborg-0.2.7.0/docs/Codec-CBOR-Decoding.html#t:DecodeAction)
Contributor guide
Assessment
This issue has not been assessed yet.