{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE ScopedTypeVariables #-}
module Ouroboros.Network.Channel
( Channel (..)
, toChannel
, fromChannel
, createPipeConnectedChannels
, hoistChannel
, isoKleisliChannel
, fixedInputChannel
, mvarsAsChannel
, handlesAsChannel
, createConnectedChannels
, createConnectedBufferedChannels
, createConnectedBufferedChannelsSTM
, createPipelineTestChannels
, channelEffect
, delayChannel
, loggingChannel
) where
import Control.Monad ((>=>))
import Control.Monad.Class.MonadSay
import Control.Monad.Class.MonadTimer.SI
import Data.ByteString qualified as BS
import Data.ByteString.Lazy qualified as LBS
import Data.ByteString.Lazy.Internal (smallChunkSize)
import Numeric.Natural
import System.IO qualified as IO (Handle, hFlush, hIsEOF)
import Control.Concurrent.Class.MonadSTM.Strict
import Network.Mux.Channel qualified as Mx
data Channel m a = Channel {
forall (m :: * -> *) a. Channel m a -> a -> m ()
send :: a -> m (),
forall (m :: * -> *) a. Channel m a -> m (Maybe a)
recv :: m (Maybe a)
}
fromChannel :: Mx.Channel m
-> Channel m LBS.ByteString
fromChannel :: forall (m :: * -> *). Channel m -> Channel m ByteString
fromChannel Mx.Channel { ByteString -> m ()
send :: ByteString -> m ()
send :: forall (m :: * -> *). Channel m -> ByteString -> m ()
Mx.send, m (Maybe ByteString)
recv :: m (Maybe ByteString)
recv :: forall (m :: * -> *). Channel m -> m (Maybe ByteString)
Mx.recv } = Channel {
send :: ByteString -> m ()
send = ByteString -> m ()
send,
recv :: m (Maybe ByteString)
recv = m (Maybe ByteString)
recv
}
toChannel :: Channel m LBS.ByteString
-> Mx.Channel m
toChannel :: forall (m :: * -> *). Channel m ByteString -> Channel m
toChannel Channel { ByteString -> m ()
send :: forall (m :: * -> *) a. Channel m a -> a -> m ()
send :: ByteString -> m ()
send, m (Maybe ByteString)
recv :: forall (m :: * -> *) a. Channel m a -> m (Maybe a)
recv :: m (Maybe ByteString)
recv } = Mx.Channel {
send :: ByteString -> m ()
Mx.send = ByteString -> m ()
send,
recv :: m (Maybe ByteString)
Mx.recv = m (Maybe ByteString)
recv
}
createPipeConnectedChannels :: IO (Channel IO LBS.ByteString,
Channel IO LBS.ByteString)
createPipeConnectedChannels :: IO (Channel IO ByteString, Channel IO ByteString)
createPipeConnectedChannels =
(\(Channel IO
a, Channel IO
b) -> (Channel IO -> Channel IO ByteString
forall (m :: * -> *). Channel m -> Channel m ByteString
fromChannel Channel IO
a, Channel IO -> Channel IO ByteString
forall (m :: * -> *). Channel m -> Channel m ByteString
fromChannel Channel IO
b))
((Channel IO, Channel IO)
-> (Channel IO ByteString, Channel IO ByteString))
-> IO (Channel IO, Channel IO)
-> IO (Channel IO ByteString, Channel IO ByteString)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IO (Channel IO, Channel IO)
Mx.createPipeConnectedChannels
isoKleisliChannel
:: forall a b m. Monad m
=> (a -> m b)
-> (b -> m a)
-> Channel m a
-> Channel m b
isoKleisliChannel :: forall a b (m :: * -> *).
Monad m =>
(a -> m b) -> (b -> m a) -> Channel m a -> Channel m b
isoKleisliChannel a -> m b
f b -> m a
finv Channel{a -> m ()
send :: forall (m :: * -> *) a. Channel m a -> a -> m ()
send :: a -> m ()
send, m (Maybe a)
recv :: forall (m :: * -> *) a. Channel m a -> m (Maybe a)
recv :: m (Maybe a)
recv} = Channel {
send :: b -> m ()
send = b -> m a
finv (b -> m a) -> (a -> m ()) -> b -> m ()
forall (m :: * -> *) a b c.
Monad m =>
(a -> m b) -> (b -> m c) -> a -> m c
>=> a -> m ()
send,
recv :: m (Maybe b)
recv = m (Maybe a)
recv m (Maybe a) -> (Maybe a -> m (Maybe b)) -> m (Maybe b)
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (a -> m b) -> Maybe a -> m (Maybe b)
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> Maybe a -> f (Maybe b)
traverse a -> m b
f
}
hoistChannel
:: (forall x . m x -> n x)
-> Channel m a
-> Channel n a
hoistChannel :: forall (m :: * -> *) (n :: * -> *) a.
(forall x. m x -> n x) -> Channel m a -> Channel n a
hoistChannel forall x. m x -> n x
nat Channel m a
channel = Channel
{ send :: a -> n ()
send = m () -> n ()
forall x. m x -> n x
nat (m () -> n ()) -> (a -> m ()) -> a -> n ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Channel m a -> a -> m ()
forall (m :: * -> *) a. Channel m a -> a -> m ()
send Channel m a
channel
, recv :: n (Maybe a)
recv = m (Maybe a) -> n (Maybe a)
forall x. m x -> n x
nat (Channel m a -> m (Maybe a)
forall (m :: * -> *) a. Channel m a -> m (Maybe a)
recv Channel m a
channel)
}
fixedInputChannel :: MonadSTM m => [a] -> m (Channel m a)
fixedInputChannel :: forall (m :: * -> *) a. MonadSTM m => [a] -> m (Channel m a)
fixedInputChannel [a]
xs0 = do
v <- [a] -> m (StrictTVar m [a])
forall (m :: * -> *) a. MonadSTM m => a -> m (StrictTVar m a)
newTVarIO [a]
xs0
return Channel {send, recv = recv v}
where
recv :: StrictTVar m [a] -> m (Maybe a)
recv StrictTVar m [a]
v = STM m (Maybe a) -> m (Maybe a)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (Maybe a) -> m (Maybe a)) -> STM m (Maybe a) -> m (Maybe a)
forall a b. (a -> b) -> a -> b
$ do
xs <- StrictTVar m [a] -> STM m [a]
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m [a]
v
case xs of
[] -> Maybe a -> STM m (Maybe a)
forall a. a -> STM m a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe a
forall a. Maybe a
Nothing
(a
x:[a]
xs') -> StrictTVar m [a] -> [a] -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> a -> STM m ()
writeTVar StrictTVar m [a]
v [a]
xs' STM m () -> STM m (Maybe a) -> STM m (Maybe a)
forall a b. STM m a -> STM m b -> STM m b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Maybe a -> STM m (Maybe a)
forall a. a -> STM m a
forall (m :: * -> *) a. Monad m => a -> m a
return (a -> Maybe a
forall a. a -> Maybe a
Just a
x)
send :: p -> m ()
send p
_ = () -> m ()
forall a. a -> m a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
mvarsAsChannel :: MonadSTM m
=> StrictTMVar m a
-> StrictTMVar m a
-> Channel m a
mvarsAsChannel :: forall (m :: * -> *) a.
MonadSTM m =>
StrictTMVar m a -> StrictTMVar m a -> Channel m a
mvarsAsChannel StrictTMVar m a
bufferRead StrictTMVar m a
bufferWrite =
Channel{a -> m ()
send :: a -> m ()
send :: a -> m ()
send, m (Maybe a)
recv :: m (Maybe a)
recv :: m (Maybe a)
recv}
where
send :: a -> m ()
send a
x = STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (StrictTMVar m a -> a -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTMVar m a -> a -> STM m ()
putTMVar StrictTMVar m a
bufferWrite a
x)
recv :: m (Maybe a)
recv = STM m (Maybe a) -> m (Maybe a)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (a -> Maybe a
forall a. a -> Maybe a
Just (a -> Maybe a) -> STM m a -> STM m (Maybe a)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> StrictTMVar m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => StrictTMVar m a -> STM m a
takeTMVar StrictTMVar m a
bufferRead)
createConnectedChannels :: MonadSTM m => m (Channel m a, Channel m a)
createConnectedChannels :: forall (m :: * -> *) a. MonadSTM m => m (Channel m a, Channel m a)
createConnectedChannels = do
bufferA <- m (StrictTMVar m a)
forall (m :: * -> *) a. MonadSTM m => m (StrictTMVar m a)
newEmptyTMVarIO
bufferB <- newEmptyTMVarIO
return (mvarsAsChannel bufferB bufferA,
mvarsAsChannel bufferA bufferB)
createConnectedBufferedChannels :: forall m a. MonadSTM m
=> Natural -> m (Channel m a, Channel m a)
createConnectedBufferedChannels :: forall (m :: * -> *) a.
MonadSTM m =>
Natural -> m (Channel m a, Channel m a)
createConnectedBufferedChannels Natural
sz = do
(chan1, chan2) <- STM m (Channel (STM m) a, Channel (STM m) a)
-> m (Channel (STM m) a, Channel (STM m) a)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (Channel (STM m) a, Channel (STM m) a)
-> m (Channel (STM m) a, Channel (STM m) a))
-> STM m (Channel (STM m) a, Channel (STM m) a)
-> m (Channel (STM m) a, Channel (STM m) a)
forall a b. (a -> b) -> a -> b
$ Natural -> STM m (Channel (STM m) a, Channel (STM m) a)
forall (m :: * -> *) a.
MonadSTM m =>
Natural -> STM m (Channel (STM m) a, Channel (STM m) a)
createConnectedBufferedChannelsSTM Natural
sz
pure (wrap chan1, wrap chan2)
where
wrap :: Channel (STM m) a -> Channel m a
wrap :: Channel (STM m) a -> Channel m a
wrap Channel{a -> STM m ()
send :: forall (m :: * -> *) a. Channel m a -> a -> m ()
send :: a -> STM m ()
send, STM m (Maybe a)
recv :: forall (m :: * -> *) a. Channel m a -> m (Maybe a)
recv :: STM m (Maybe a)
recv} = Channel
{ send :: a -> m ()
send = STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> (a -> STM m ()) -> a -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> STM m ()
send
, recv :: m (Maybe a)
recv = STM m (Maybe a) -> m (Maybe a)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM m (Maybe a)
recv
}
createConnectedBufferedChannelsSTM :: MonadSTM m
=> Natural -> STM m (Channel (STM m) a, Channel (STM m) a)
createConnectedBufferedChannelsSTM :: forall (m :: * -> *) a.
MonadSTM m =>
Natural -> STM m (Channel (STM m) a, Channel (STM m) a)
createConnectedBufferedChannelsSTM Natural
sz = do
bufferA <- Natural -> STM m (StrictTBQueue m a)
forall (m :: * -> *) a.
MonadSTM m =>
Natural -> STM m (StrictTBQueue m a)
newTBQueue Natural
sz
bufferB <- newTBQueue sz
return (queuesAsChannel bufferB bufferA,
queuesAsChannel bufferA bufferB)
where
queuesAsChannel :: StrictTBQueue m a -> StrictTBQueue m a -> Channel (STM m) a
queuesAsChannel StrictTBQueue m a
bufferRead StrictTBQueue m a
bufferWrite =
Channel{a -> STM m ()
send :: a -> STM m ()
send :: a -> STM m ()
send, STM m (Maybe a)
recv :: STM m (Maybe a)
recv :: STM m (Maybe a)
recv}
where
send :: a -> STM m ()
send a
x = StrictTBQueue m a -> a -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTBQueue m a -> a -> STM m ()
writeTBQueue StrictTBQueue m a
bufferWrite a
x
recv :: STM m (Maybe a)
recv = a -> Maybe a
forall a. a -> Maybe a
Just (a -> Maybe a) -> STM m a -> STM m (Maybe a)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> StrictTBQueue m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => StrictTBQueue m a -> STM m a
readTBQueue StrictTBQueue m a
bufferRead
createPipelineTestChannels :: MonadSTM m
=> Natural -> m (Channel m a, Channel m a)
createPipelineTestChannels :: forall (m :: * -> *) a.
MonadSTM m =>
Natural -> m (Channel m a, Channel m a)
createPipelineTestChannels Natural
sz = do
bufferA <- STM m (StrictTBQueue m a) -> m (StrictTBQueue m a)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (StrictTBQueue m a) -> m (StrictTBQueue m a))
-> STM m (StrictTBQueue m a) -> m (StrictTBQueue m a)
forall a b. (a -> b) -> a -> b
$ Natural -> STM m (StrictTBQueue m a)
forall (m :: * -> *) a.
MonadSTM m =>
Natural -> STM m (StrictTBQueue m a)
newTBQueue Natural
sz
bufferB <- atomically $ newTBQueue sz
return (queuesAsChannel bufferB bufferA,
queuesAsChannel bufferA bufferB)
where
queuesAsChannel :: StrictTBQueue m a -> StrictTBQueue m a -> Channel m a
queuesAsChannel StrictTBQueue m a
bufferRead StrictTBQueue m a
bufferWrite =
Channel{a -> m ()
send :: a -> m ()
send :: a -> m ()
send, m (Maybe a)
recv :: m (Maybe a)
recv :: m (Maybe a)
recv}
where
send :: a -> m ()
send a
x = STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
full <- StrictTBQueue m a -> STM m Bool
forall (m :: * -> *) a.
MonadSTM m =>
StrictTBQueue m a -> STM m Bool
isFullTBQueue StrictTBQueue m a
bufferWrite
if full then error failureMsg
else writeTBQueue bufferWrite x
recv :: m (Maybe a)
recv = STM m (Maybe a) -> m (Maybe a)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (a -> Maybe a
forall a. a -> Maybe a
Just (a -> Maybe a) -> STM m a -> STM m (Maybe a)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> StrictTBQueue m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => StrictTBQueue m a -> STM m a
readTBQueue StrictTBQueue m a
bufferRead)
failureMsg :: [Char]
failureMsg = [Char]
"createPipelineTestChannels: "
[Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ [Char]
"maximum pipeline depth exceeded: " [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ Natural -> [Char]
forall a. Show a => a -> [Char]
show Natural
sz
handlesAsChannel :: IO.Handle
-> IO.Handle
-> Channel IO LBS.ByteString
handlesAsChannel :: Handle -> Handle -> Channel IO ByteString
handlesAsChannel Handle
hndRead Handle
hndWrite =
Channel{ByteString -> IO ()
send :: ByteString -> IO ()
send :: ByteString -> IO ()
send, IO (Maybe ByteString)
recv :: IO (Maybe ByteString)
recv :: IO (Maybe ByteString)
recv}
where
send :: LBS.ByteString -> IO ()
send :: ByteString -> IO ()
send ByteString
chunk = do
Handle -> ByteString -> IO ()
LBS.hPut Handle
hndWrite ByteString
chunk
Handle -> IO ()
IO.hFlush Handle
hndWrite
recv :: IO (Maybe LBS.ByteString)
recv :: IO (Maybe ByteString)
recv = do
eof <- Handle -> IO Bool
IO.hIsEOF Handle
hndRead
if eof
then return Nothing
else Just . LBS.fromStrict <$> BS.hGetSome hndRead smallChunkSize
channelEffect :: forall m a.
Monad m
=> (a -> m ())
-> (Maybe a -> m ())
-> Channel m a
-> Channel m a
channelEffect :: forall (m :: * -> *) a.
Monad m =>
(a -> m ()) -> (Maybe a -> m ()) -> Channel m a -> Channel m a
channelEffect a -> m ()
beforeSend Maybe a -> m ()
afterRecv Channel{a -> m ()
send :: forall (m :: * -> *) a. Channel m a -> a -> m ()
send :: a -> m ()
send, m (Maybe a)
recv :: forall (m :: * -> *) a. Channel m a -> m (Maybe a)
recv :: m (Maybe a)
recv} =
Channel{
send :: a -> m ()
send = \a
x -> do
a -> m ()
beforeSend a
x
a -> m ()
send a
x
, recv :: m (Maybe a)
recv = do
mx <- m (Maybe a)
recv
afterRecv mx
return mx
}
delayChannel :: MonadDelay m
=> DiffTime
-> Channel m a
-> Channel m a
delayChannel :: forall (m :: * -> *) a.
MonadDelay m =>
DiffTime -> Channel m a -> Channel m a
delayChannel DiffTime
delay = (a -> m ()) -> (Maybe a -> m ()) -> Channel m a -> Channel m a
forall (m :: * -> *) a.
Monad m =>
(a -> m ()) -> (Maybe a -> m ()) -> Channel m a -> Channel m a
channelEffect (\a
_ -> () -> m ()
forall a. a -> m a
forall (m :: * -> *) a. Monad m => a -> m a
return ())
(\Maybe a
_ -> DiffTime -> m ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
delay)
loggingChannel :: ( MonadSay m
, Show id
, Show a
)
=> id
-> Channel m a
-> Channel m a
loggingChannel :: forall (m :: * -> *) id a.
(MonadSay m, Show id, Show a) =>
id -> Channel m a -> Channel m a
loggingChannel id
ident Channel{a -> m ()
send :: forall (m :: * -> *) a. Channel m a -> a -> m ()
send :: a -> m ()
send,m (Maybe a)
recv :: forall (m :: * -> *) a. Channel m a -> m (Maybe a)
recv :: m (Maybe a)
recv} =
Channel {
send :: a -> m ()
send = a -> m ()
loggingSend,
recv :: m (Maybe a)
recv = m (Maybe a)
loggingRecv
}
where
loggingSend :: a -> m ()
loggingSend a
a = do
[Char] -> m ()
forall (m :: * -> *). MonadSay m => [Char] -> m ()
say (id -> [Char]
forall a. Show a => a -> [Char]
show id
ident [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ [Char]
":send:" [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ a -> [Char]
forall a. Show a => a -> [Char]
show a
a)
a -> m ()
send a
a
loggingRecv :: m (Maybe a)
loggingRecv = do
msg <- m (Maybe a)
recv
case msg of
Maybe a
Nothing -> () -> m ()
forall a. a -> m a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
Just a
a -> [Char] -> m ()
forall (m :: * -> *). MonadSay m => [Char] -> m ()
say (id -> [Char]
forall a. Show a => a -> [Char]
show id
ident [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ [Char]
":recv:" [Char] -> [Char] -> [Char]
forall a. [a] -> [a] -> [a]
++ a -> [Char]
forall a. Show a => a -> [Char]
show a
a)
return msg