[CDEC-464] Switch uses of pipes to conduit

I did not replace uses of `pipes` in the `post-mortem` executable from
`cardano-sl-tools`, since it does not build at present.

Author: Michael Hueschen

db/cardano-sl-db.cabal

@@ -143,7 +143,6 @@ library
                      , memory
                      , mmorph
                      , mtl
-                     , pipes
                      , reflection
                      , resourcet
                      , rocksdb-haskell-ng

db/src/Pos/DB/Block/GState/BlockExtra.hs

@@ -20,10 +20,10 @@ module Pos.DB.Block.GState.BlockExtra
 
 import           Universum hiding (init)
 
+import           Data.Conduit (ConduitT, yield)
 import qualified Database.RocksDB as Rocks
 import           Formatting (bprint, build, (%))
 import qualified Formatting.Buildable
-import           Pipes (Producer, yield)
 import           Serokell.Util.Text (listJson)
 
 import           Pos.Binary.Class (serialize')
@@ -122,7 +122,7 @@ streamBlocks
     => (HeaderHash -> m (Maybe SerializedBlock))
     -> (HeaderHash -> m (Maybe HeaderHash))
     -> HeaderHash
-    -> Producer SerializedBlock m ()
+    -> ConduitT () SerializedBlock m ()
 streamBlocks loadBlock forwardLink base = do
     mFirst <- lift $ forwardLink base
     maybe (pure ()) loop mFirst

infra/src/Pos/Infra/Diffusion/Subscription/Subscriber.hs

@@ -12,7 +12,7 @@ module Pos.Infra.Diffusion.Subscription.Subscriber
 import           Universum
 
 -- | Generate subscription targets in some monad.
--- TBD any value in using a streaming solution like pipes?
+-- TBD any value in using a streaming solution like conduit?
 newtype SubscriptionTarget m target = SubscriptionTarget
     { getSubscriptionTarget :: m (Maybe (target, SubscriptionTarget m target))
     }

lib/bench/Bench/Pos/Diffusion/BlockDownload.hs

@@ -19,22 +19,23 @@ import qualified Criterion
 import qualified Criterion.Main as Criterion
 import qualified Criterion.Main.Options as Criterion
 import qualified Data.ByteString.Lazy as LBS
+import           Data.Conduit.Combinators (yieldMany)
+import           Data.List.NonEmpty (NonEmpty ((:|)))
 import           Data.Semigroup ((<>))
+import           Data.Time.Units (Microsecond)
 import qualified Options.Applicative as Opt (execParser)
 
-import           Data.List.NonEmpty (NonEmpty ((:|)))
-import           Data.Time.Units (Microsecond)
 import qualified Network.Broadcast.OutboundQueue as OQ
 import qualified Network.Broadcast.OutboundQueue.Types as OQ
 import           Network.Transport (Transport)
 import qualified Network.Transport.TCP as TCP
 import           Node (NodeId)
 import qualified Node
-import           Pipes (each)
 
 import           Pos.Binary (serialize, serialize')
 import           Pos.Core.Block (Block, BlockHeader, HeaderHash)
 import qualified Pos.Core.Block as Block (getBlockHeader)
+import           Pos.Core.Chrono (NewestFirst (..), OldestFirst (..))
 import           Pos.Core.ProtocolConstants (ProtocolConstants (..))
 import           Pos.Core.Update (BlockVersion (..))
 import           Pos.Crypto (ProtocolMagic (..))
@@ -52,9 +53,8 @@ import           Pos.Infra.Network.Types (Bucket (..))
 import           Pos.Infra.Reporting.Health.Types (HealthStatus (..))
 import           Pos.Logic.Pure (pureLogic)
 import           Pos.Logic.Types as Logic (Logic (..))
-
-import           Pos.Core.Chrono (NewestFirst (..), OldestFirst (..))
 import           Pos.Util.Trace (noTrace, wlogTrace)
+
 import           Test.Pos.Chain.Block.Arbitrary.Generate (generateMainBlock)
 
 -- TODO
@@ -121,7 +121,7 @@ serverLogic streamIORef arbitraryBlock arbitraryHashes arbitraryHeaders = pureLo
     , getTipHeader = pure (Block.getBlockHeader arbitraryBlock)
     , Logic.streamBlocks = \_ -> do
           bs <-  readIORef streamIORef
-          each $ map serializedBlock bs
+          yieldMany $ map serializedBlock bs
     }
 
 serializedBlock :: Block -> SerializedBlock

lib/cardano-sl.cabal

@@ -183,7 +183,6 @@ library
                       , network-transport
                       , optparse-applicative
                       , parsec
-                      , pipes
                       , pvss
                       , random
                       , reflection
@@ -313,6 +312,7 @@ test-suite cardano-test
                      , cardano-sl-networking
                      , cardano-sl-util
                      , cardano-sl-util-test
+                     , conduit
                      , containers
                      , cryptonite
                      , data-default
@@ -326,7 +326,6 @@ test-suite cardano-test
                      , log-warper
                      , network-transport
                      , network-transport-inmemory
-                     , pipes
                      , pvss
                      , random
                      , reflection
@@ -400,13 +399,13 @@ benchmark cardano-bench-criterion
                      , cardano-sl-networking
                      , cardano-sl-util
                      , cardano-sl-util-test
+                     , conduit
                      , criterion
                      , deepseq
                      , formatting
                      , network-transport
                      , network-transport-tcp
                      , optparse-applicative
-                     , pipes
                      , time-units
                      , universum >= 0.1.11
   default-language:    Haskell2010

lib/src/Pos/Diffusion/Full/Block.hs

@@ -19,6 +19,7 @@ import           Control.Exception (Exception (..), throwIO)
 import           Control.Lens (to)
 import           Control.Monad.Except (ExceptT, runExceptT, throwError)
 import qualified Data.ByteString.Lazy as BSL
+import           Data.Conduit (await, runConduit, (.|))
 import           Data.List.NonEmpty (NonEmpty ((:|)))
 import qualified Data.List.NonEmpty as NE
 import qualified Data.Map as M
@@ -27,7 +28,6 @@ import           Formatting (bprint, build, int, sformat, shown, stext, (%))
 import qualified Formatting.Buildable as B
 import qualified Network.Broadcast.OutboundQueue as OQ
 import           Node.Conversation (sendRaw)
-import           Pipes (await, runEffect, (>->))
 import           Serokell.Util.Text (listJson)
 import qualified System.Metrics.Gauge as Gauge
 
@@ -689,7 +689,7 @@ handleStreamStart logTrace logic oq = listenerConv logTrace oq $ \__ourVerInfo n
                         Logic.streamBlocks logic lca
                         lift $ send conv MsgStreamEnd
                     consumer = loop nodeId conv window
-                runEffect $ producer >-> consumer
+                runConduit $ producer .| consumer
 
     loop nodeId conv 0 = do
         lift $ traceWith logTrace (Debug, "handleStreamStart:loop waiting on window update")
@@ -703,10 +703,10 @@ handleStreamStart logTrace logic oq = listenerConv logTrace oq $ \__ourVerInfo n
                   MsgUpdate u -> do
                       lift $ traceWith logTrace (Debug, sformat ("handleStreamStart:loop new window "%shown%" from "%build) u nodeId)
                       loop nodeId conv (msuWindow u)
-    loop nodeId conv window = do
-        b <- await
-        lift $ sendRaw conv $ serializeMsgStreamBlock $ MsgSerializedBlock b
-        loop nodeId conv (window - 1)
+    loop nodeId conv window =
+        whenJustM await $ \b -> do
+            lift $ sendRaw conv $ serializeMsgStreamBlock $ MsgSerializedBlock b
+            loop nodeId conv (window - 1)
 
 ----------------------------------------------------------------------------
 -- Header propagation

lib/src/Pos/Logic/Full.hs

@@ -9,10 +9,10 @@ module Pos.Logic.Full
 import           Universum hiding (id)
 
 import           Control.Lens (at, to)
+import           Data.Conduit (ConduitT)
 import qualified Data.HashMap.Strict as HM
 import           Data.Tagged (Tagged (..), tagWith)
 import           Formatting (build, sformat, (%))
-import           Pipes (Producer)
 import           System.Wlog (WithLogger, logDebug)
 
 import           Pos.Chain.Block (HasBlockConfiguration)
@@ -106,7 +106,7 @@ logicFull pm txpConfig ourStakeholderId securityParams jsonLogTx =
         getSerializedBlock :: HeaderHash -> m (Maybe SerializedBlock)
         getSerializedBlock = DB.dbGetSerBlock
 
-        streamBlocks :: HeaderHash -> Producer SerializedBlock m ()
+        streamBlocks :: HeaderHash -> ConduitT () SerializedBlock m ()
         streamBlocks = Block.streamBlocks DB.dbGetSerBlock Block.resolveForwardLink
 
         getTip :: m Block

lib/src/Pos/Logic/Types.hs

@@ -12,10 +12,9 @@ module Pos.Logic.Types
 
 import           Universum
 
+import           Data.Conduit (ConduitT, transPipe)
 import           Data.Default (def)
 import           Data.Tagged (Tagged)
-import           Pipes (Producer)
-import           Pipes.Internal (unsafeHoist)
 
 import           Pos.Chain.Security (SecurityParams (..))
 import           Pos.Chain.Ssc (MCCommitment, MCOpening, MCShares,
@@ -38,7 +37,7 @@ data Logic m = Logic
       ourStakeholderId   :: StakeholderId
       -- | Get serialized block, perhaps from a database.
     , getSerializedBlock :: HeaderHash -> m (Maybe SerializedBlock)
-    , streamBlocks       :: HeaderHash -> Producer SerializedBlock m ()
+    , streamBlocks       :: HeaderHash -> ConduitT () SerializedBlock m ()
       -- | Get a block header.
     , getBlockHeader     :: HeaderHash -> m (Maybe BlockHeader)
       -- TODO CSL-2089 use conduits in this and the following methods
@@ -102,11 +101,15 @@ data Logic m = Logic
     , securityParams     :: SecurityParams
     }
 
--- | We have to hoist a pipes producer, so the Monad constraint arises.
+-- | The Monad constraint arises due to `transPipe` from Conduit.
+--   The transformation function `foo :: (forall x. m x -> n x)` must be a
+--   *monad morphism* and not just any natural transformation. This means,
+--   roughly, that `foo a >> foo b` should behave the same as `foo (a >> b)`.
+--   `foo = flip evalState 1`, for example, does not satisfy this requirement.
 hoistLogic :: Monad m => (forall x . m x -> n x) -> Logic m -> Logic n
 hoistLogic nat logic = logic
     { getSerializedBlock = nat . getSerializedBlock logic
-    , streamBlocks = unsafeHoist nat . streamBlocks logic
+    , streamBlocks = transPipe nat . streamBlocks logic
     , getBlockHeader = nat . getBlockHeader logic
     , getHashesRange = \a b c -> nat (getHashesRange logic a b c)
     , getBlockHeaders = \a b c -> nat (getBlockHeaders logic a b c)

lib/test/Test/Pos/Diffusion/BlockSpec.hs

@@ -10,25 +10,26 @@ import           Universum
 
 import           Control.DeepSeq (force)
 import           Control.Monad.IO.Class (liftIO)
+import           Data.Bits
 import qualified Data.ByteString.Lazy as LBS
+import           Data.Conduit.Combinators (yieldMany)
+import           Data.List.NonEmpty (NonEmpty ((:|)))
+import qualified Data.List.NonEmpty as NE
 import           Data.Semigroup ((<>))
 import           Test.Hspec (Spec, describe, it, shouldBe)
 
-import           Data.Bits
-import           Data.List.NonEmpty (NonEmpty ((:|)))
-import qualified Data.List.NonEmpty as NE
 import qualified Network.Broadcast.OutboundQueue as OQ
 import qualified Network.Broadcast.OutboundQueue.Types as OQ
 import           Network.Transport (Transport, closeTransport)
 import qualified Network.Transport.InMemory as InMemory
 import           Node (NodeId)
 import qualified Node
-import           Pipes (each)
 
 import           Pos.Binary.Class (serialize')
 import           Pos.Core.Block (Block, BlockHeader, HeaderHash,
                      blockHeaderHash)
 import qualified Pos.Core.Block as Block (getBlockHeader)
+import           Pos.Core.Chrono (NewestFirst (..), OldestFirst (..))
 import           Pos.Core.ProtocolConstants (ProtocolConstants (..))
 import           Pos.Core.Update (BlockVersion (..))
 import           Pos.Crypto (ProtocolMagic (..))
@@ -44,9 +45,8 @@ import           Pos.Infra.Network.Types (Bucket (..))
 import           Pos.Infra.Reporting.Health.Types (HealthStatus (..))
 import           Pos.Logic.Pure (pureLogic)
 import           Pos.Logic.Types as Logic (Logic (..))
-
-import           Pos.Core.Chrono (NewestFirst (..), OldestFirst (..))
 import           Pos.Util.Trace (wlogTrace)
+
 import           Test.Pos.Chain.Block.Arbitrary.Generate (generateMainBlock)
 
 -- HLint warning disabled since I ran into https://ghc.haskell.org/trac/ghc/ticket/13106
@@ -104,7 +104,7 @@ serverLogic streamIORef arbitraryBlock arbitraryHashes arbitraryHeaders = pureLo
     , getTipHeader = pure (Block.getBlockHeader arbitraryBlock)
     , Logic.streamBlocks = \_ -> do
           bs <-  readIORef streamIORef
-          each $ map serializedBlock bs
+          yieldMany $ map serializedBlock bs
     }
 
 serializedBlock :: Block -> SerializedBlock

pkgs/default.nix

@@ -14759,7 +14759,6 @@ license = stdenv.lib.licenses.bsd3;
 , network-transport-tcp
 , optparse-applicative
 , parsec
-, pipes
 , pvss
 , QuickCheck
 , random
@@ -14854,7 +14853,6 @@ network
 network-transport
 optparse-applicative
 parsec
-pipes
 pvss
 QuickCheck
 random
@@ -14907,6 +14905,7 @@ cardano-sl-infra-test
 cardano-sl-networking
 cardano-sl-util
 cardano-sl-util-test
+conduit
 containers
 cryptonite
 data-default
@@ -14920,7 +14919,6 @@ lens
 log-warper
 network-transport
 network-transport-inmemory
-pipes
 pvss
 QuickCheck
 random
@@ -14947,13 +14945,13 @@ cardano-sl-infra
 cardano-sl-networking
 cardano-sl-util
 cardano-sl-util-test
+conduit
 criterion
 deepseq
 formatting
 network-transport
 network-transport-tcp
 optparse-applicative
-pipes
 QuickCheck
 time-units
 universum
@@ -16106,7 +16104,6 @@ license = stdenv.lib.licenses.mit;
 , memory
 , mmorph
 , mtl
-, pipes
 , reflection
 , resourcet
 , rocksdb-haskell-ng
@@ -16157,7 +16154,6 @@ lrucache
 memory
 mmorph
 mtl
-pipes
 reflection
 resourcet
 rocksdb-haskell-ng