hnix-store-remote: use pipes to stream logs..

..instead of collecting them in memory and then returning everything
at once.

I did this by changing the `MonadStore` to:
```haskell
type MonadStore a = ExceptT Error (Producer Logger (ReaderT Socket IO)) a
```
It's a `Producer` of log messages that can fail with an `Error`:
```haskell
data Error =
    LogError Int LBS.ByteString
  | ParseError String
  | ConnError String
```

`runStore` can now accept a `Consumer Logger IO (Either Error a)` parameter to consume
logs incrementally. I added a `runStore_` that simply discards all logs.

Similarly, I changed the `runOpArgs` and `runOp`: now they accept a
`Get a` parser that is used to get the result. `runOp_` and `runOpArgs_`
assume that op does not return anything interesting.

Author: stolyaroleh

hnix-store-remote/src/System/Nix/Store/Remote.hs

@@ -22,13 +22,12 @@ type RepairFlag = Bool
 type CheckFlag = Bool
 
 syncWithGC :: MonadStore ()
-syncWithGC = void $ simpleOp SyncWithGC
+syncWithGC = runOp_ SyncWithGC
 
 optimiseStore :: MonadStore ()
-optimiseStore = void $ simpleOp OptimiseStore
+optimiseStore = runOp_ OptimiseStore
 
--- returns True on errors
-verifyStore :: CheckFlag -> RepairFlag -> MonadStore Bool
-verifyStore check repair = simpleOpArgs VerifyStore $ do
+verifyStore :: CheckFlag -> RepairFlag -> MonadStore ()
+verifyStore check repair = runOpArgs_ VerifyStore $ do
   putBool check
   putBool repair

hnix-store-remote/src/System/Nix/Store/Remote/Logger.hs

@@ -1,14 +1,15 @@
 module System.Nix.Store.Remote.Logger (
     Logger(..)
   , Field(..)
-  , processOutput)
-  where
+  , streamLogs
+  ) where
 
+import           Control.Monad.Except      (throwError)
+import           Control.Monad             (replicateM)
 import           Control.Monad.Reader      (ask, liftIO)
 import           Data.Binary.Get
-
 import           Network.Socket.ByteString (recv)
-
+import           Pipes                     (lift, yield)
 import           System.Nix.Store.Remote.Types
 import           System.Nix.Util
 
@@ -26,30 +27,38 @@ controlParser = do
     0x52534c54 -> Result        <$> getInt <*> getInt <*> getFields
     x          -> fail           $ "Invalid control message received:" ++ show x
 
-processOutput :: MonadStore [Logger]
-processOutput = go decoder
-  where decoder = runGetIncremental controlParser
-        go :: Decoder Logger -> MonadStore [Logger]
-        go (Done _leftover _consumed ctrl) = do
-          case ctrl of
-            e@(Error _ _) -> return [e]
-            Last -> return [Last]
-            -- we should probably handle Read here as well
-            x -> do
-              next <- go decoder
-              return $ x:next
-        go (Partial k) = do
-          soc <- ask
-          chunk <- liftIO (Just <$> recv soc 8)
-          go (k chunk)
+logger :: Logger -> MonadStore ()
+logger = lift . yield
+
+streamLogs :: MonadStore ()
+streamLogs = go decoder
+  where
+    go :: Decoder Logger -> MonadStore ()
+    go (Done _leftover _consumed ctrl) = do
+      case ctrl of
+        e@(Error status err) -> do
+          logger e
+          throwError (LogError status err)
+        Last ->
+          logger Last
+        -- we should probably handle Read here as well
+        x -> do
+          logger x
+          go decoder
+    go (Partial cont) = do
+      soc <- ask
+      chunk <- liftIO (recv soc 8)
+      go (cont (Just chunk))
+    go (Fail _leftover _consumed msg) =
+      throwError (ParseError msg)
 
-        go (Fail _leftover _consumed msg) = do
-          error msg
+    decoder :: Decoder Logger
+    decoder = runGetIncremental controlParser
 
 getFields :: Get [Field]
 getFields = do
-  cnt <- getInt
-  sequence $ replicate cnt getField
+  count <- getInt
+  replicateM count getField
 
 getField :: Get Field
 getField = do

hnix-store-remote/src/System/Nix/Store/Remote/Protocol.hs

@@ -1,23 +1,25 @@
+{-# LANGUAGE TypeApplications #-}
 module System.Nix.Store.Remote.Protocol (
     WorkerOp(..)
-  , simpleOp
-  , simpleOpArgs
   , runOp
+  , runOp_
   , runOpArgs
-  , runStore) where
+  , runOpArgs_
+  , runStore
+  , runStore_
+  ) where
 
-import           Control.Exception         (bracket)
-import           Control.Monad.Except
+import           Control.Exception         (SomeException, bracket, catch, displayException)
+import           Control.Monad.Except      (throwError, runExceptT)
 import           Control.Monad.Reader
-import           Control.Monad.State
 
-import           Data.Binary.Get
 import           Data.Binary.Put
-import qualified Data.ByteString.Char8     as BSC
-import qualified Data.ByteString.Lazy      as LBS
+import           Data.Binary.Get
+
+import           Network.Socket
 
-import           Network.Socket            hiding (send, sendTo, recv, recvFrom)
-import           Network.Socket.ByteString (recv)
+import           Pipes
+import qualified Pipes.Prelude as Pipes
 
 import           System.Nix.Store.Remote.Logger
 import           System.Nix.Store.Remote.Types
@@ -106,65 +108,57 @@ opNum NarFromPath                 = 38
 opNum AddToStoreNar               = 39
 opNum QueryMissing                = 40
 
+runOp :: WorkerOp -> Get a -> MonadStore a
+runOp op result = runOpArgs op mempty result
 
-simpleOp :: WorkerOp -> MonadStore Bool
-simpleOp op = do
-  simpleOpArgs op $ return ()
-
-simpleOpArgs :: WorkerOp -> Put -> MonadStore Bool
-simpleOpArgs op args = do
-  runOpArgs op args
-  err <- gotError
-  case err of
-    True -> do
-      Error _num msg <- head <$> getError
-      throwError $ BSC.unpack $ LBS.toStrict msg
-    False -> do
-      sockGetBool
-
-runOp :: WorkerOp -> MonadStore ()
-runOp op = runOpArgs op $ return ()
-
-runOpArgs :: WorkerOp -> Put -> MonadStore ()
-runOpArgs op args = do
-
-  -- Temporary hack for printing the messages destined for nix-daemon socket
-  when False $
-    liftIO $ LBS.writeFile "mytestfile2" $ runPut $ do
-      putInt $ opNum op
-      args
+runOp_ :: WorkerOp -> MonadStore ()
+runOp_ op = runOp op (skip 8)
 
+runOpArgs :: WorkerOp -> Put -> Get a -> MonadStore a
+runOpArgs op args result = do
   sockPut $ do
-    putInt $ opNum op
+    putInt (opNum op)
     args
+  streamLogs
+  sockGet result
 
-  out <- processOutput
-  modify (++out)
-  err <- gotError
-  when err $ do
-    Error _num msg <- head <$> getError
-    throwError $ BSC.unpack $ LBS.toStrict msg
+runOpArgs_ :: WorkerOp -> Put -> MonadStore ()
+runOpArgs_ op args = runOpArgs op args (skip 8)
 
-runStore :: MonadStore a -> IO (Either String a, [Logger])
-runStore code = do
-  bracket (open sockPath) close run
+runStore :: Consumer Logger IO (Either Error a) -> MonadStore a -> IO (Either Error a)
+runStore sink code =
+  bracket (open sockPath) close run `catch` onException
   where
     open path = do
-      soc <- socket AF_UNIX Stream 0
-      connect soc (SockAddrUnix path)
-      return soc
+      sock <- socket AF_UNIX Stream 0
+      connect sock (SockAddrUnix path)
+      return sock
+
     greet = do
       sockPut $ putInt workerMagic1
-      soc <- ask
-      vermagic <- liftIO $ recv soc 16
-      let (magic2, daemonProtoVersion) = flip runGet (LBS.fromStrict vermagic) $ (,) <$> getInt <*> getInt
-      unless (magic2 == workerMagic2) $ error "Worker magic 2 mismatch"
+
+      magic2 <- sockGetInt
+      _ <- sockGetInt -- daemonVersion
+
+      unless (magic2 == workerMagic2) $
+        throwError (ConnError "Worker magic 2 mismatch")
 
       sockPut $ putInt protoVersion -- clientVersion
       sockPut $ putInt (0 :: Int)   -- affinity
       sockPut $ putInt (0 :: Int)   -- obsolete reserveSpace
 
-      processOutput
+      streamLogs -- receive startup error messages, if any
 
     run sock =
-      flip runReaderT sock $ flip runStateT [] $ runExceptT (greet >> code)
+      let producer =
+            runExceptT $ do
+              greet
+              code
+          effect = producer >-> hoist liftIO sink
+      in  runReaderT (runEffect effect) sock
+
+    onException :: SomeException -> IO (Either Error a)
+    onException = return . Left . ConnError . displayException
+
+runStore_ :: MonadStore a -> IO (Either Error a)
+runStore_ = runStore Pipes.drain

hnix-store-remote/src/System/Nix/Store/Remote/Types.hs

@@ -1,20 +1,23 @@
 module System.Nix.Store.Remote.Types (
     MonadStore
+  , Error(..)
   , Logger(..)
   , Field(..)
-  , getLog
-  , flushLog
-  , gotError
-  , getError) where
+  ) where
 
-
-import qualified Data.ByteString.Lazy      as LBS
-import           Network.Socket            (Socket)
 import           Control.Monad.Except
 import           Control.Monad.Reader
-import           Control.Monad.State
+import qualified Data.ByteString.Lazy      as LBS
+import           Network.Socket            (Socket)
+import           Pipes
+
+data Error =
+    LogError Int LBS.ByteString
+  | ParseError String
+  | ConnError String
+  deriving (Eq, Show)
 
-type MonadStore a = ExceptT String (StateT [Logger] (ReaderT Socket IO)) a
+type MonadStore a = ExceptT Error (Producer Logger (ReaderT Socket IO)) a
 
 type ActivityID = Int
 type ActivityParentID = Int
@@ -35,19 +38,3 @@ data Logger =
   | StopActivity  ActivityID
   | Result        ActivityID ResultType [Field]
   deriving (Eq, Ord, Show)
-
-isError :: Logger -> Bool
-isError (Error _ _) = True
-isError _           = False
-
-gotError :: MonadStore Bool
-gotError = any isError <$> get
-
-getError :: MonadStore [Logger]
-getError = filter isError <$> get
-
-getLog :: MonadStore [Logger]
-getLog = get
-
-flushLog :: MonadStore ()
-flushLog = put []

hnix-store-remote/src/System/Nix/Store/Remote/Util.hs

@@ -1,59 +1,47 @@
 module System.Nix.Store.Remote.Util where
 
-import           Control.Monad.Reader
-
-import           Data.Maybe
+import           Control.Monad.Except (throwError)
+import           Control.Monad.Reader (ask, liftIO)
 import           Data.Binary.Get
 import           Data.Binary.Put
 import           Data.Text                 (Text)
 import qualified Data.Text                 as T
-import qualified Data.ByteString           as B
 import qualified Data.ByteString.Char8     as BSC
 import qualified Data.ByteString.Lazy      as LBS
-import qualified Data.HashSet              as HashSet
 
 import           Network.Socket.ByteString (recv, sendAll)
 
 import           System.Nix.Store.Remote.Types
-import           System.Nix.Hash
 import           System.Nix.Util
 
-
-genericIncremental :: (MonadIO m) => m (Maybe B.ByteString) -> Get a -> m a
-genericIncremental getsome parser = go decoder
-  where decoder = runGetIncremental parser
-        go (Done _leftover _consumed x) = do
-          return x
-        go (Partial k) = do
-          chunk <- getsome
-          go (k chunk)
-        go (Fail _leftover _consumed msg) = do
-          error msg
-
-getSocketIncremental :: Get a -> MonadStore a
-getSocketIncremental = genericIncremental sockGet
-
 sockPut :: Put -> MonadStore ()
 sockPut p = do
   soc <- ask
   liftIO $ sendAll soc $ LBS.toStrict $ runPut p
 
-sockGet :: MonadStore (Maybe BSC.ByteString)
-sockGet = do
-  soc <- ask
-  liftIO $ Just <$> recv soc 8
-
-sockGetInt :: Integral a => MonadStore a
-sockGetInt = getSocketIncremental getInt
+sockGet :: Get a -> MonadStore a
+sockGet = go . runGetIncremental
+  where
+    go :: Decoder a -> MonadStore a
+    go (Done _leftover _consumed x) = return x
+    go (Partial cont) = do
+      sock <- ask
+      chunk <- liftIO (recv sock 8)
+      go (cont (Just chunk))
+    go (Fail _leftover _consumed msg) =
+      throwError (ParseError msg)
+
+sockGetInt :: MonadStore Int
+sockGetInt = sockGet getInt
 
 sockGetBool :: MonadStore Bool
 sockGetBool = (== (1 :: Int)) <$> sockGetInt
 
 sockGetStr :: MonadStore LBS.ByteString
-sockGetStr = getSocketIncremental getByteStringLen
+sockGetStr = sockGet getByteStringLen
 
 sockGetStrings :: MonadStore [LBS.ByteString]
-sockGetStrings = getSocketIncremental getByteStrings
+sockGetStrings = sockGet getByteStrings
 
 lBSToText :: LBS.ByteString -> Text
 lBSToText = T.pack . BSC.unpack . LBS.toStrict