Merge pull request #782 from IntersectMBO/jdral/blockio-pinned-buffer

`blockio`: guard that buffers are pinned

Author: jorisdral

blockio/CHANGELOG.md

@@ -1,5 +1,5 @@
 # Revision history for blockio
 
-## 0.1.0.0 -- YYYY-mm-dd
+## 0.1.0.0 -- 2025-07-09
 
 * First version. Released on an unsuspecting world.

blockio/src-linux/System/FS/BlockIO/Async.hs

@@ -12,15 +12,16 @@ import           Foreign.C.Error
 import           GHC.IO.Exception
 import           GHC.Stack
 import           System.FS.API (BufferOffset (..), FsErrorPath, FsPath,
-                     Handle (..), HasFS (..), SomeHasFS (..), ioToFsError)
+                     Handle (..), HasFS (..), ioToFsError)
 import qualified System.FS.BlockIO.API as API
 import           System.FS.BlockIO.API (IOOp (..), IOResult (..), LockMode,
                      ioopHandle)
 import qualified System.FS.BlockIO.IO.Internal as IOI
 import           System.FS.IO (HandleIO)
 import           System.FS.IO.Handle
 import qualified System.IO.BlockIO as I
-import           System.IO.Error (ioeSetErrorString, isResourceVanishedError)
+import           System.IO.Error (ioeGetErrorType, ioeSetErrorString,
+                     isResourceVanishedError)
 import           System.Posix.Types
 
 -- | IO instantiation of 'HasBlockIO', using @blockio-uring@.
@@ -64,17 +65,22 @@ submitIO ::
   -> IO (VU.Vector IOResult)
 submitIO hasFS ioctx ioops = do
     ioops' <- mapM ioopConv ioops
-    ress <- I.submitIO ioctx ioops' `catch` rethrowClosedError
+    ress <- I.submitIO ioctx ioops' `catch` rethrowFsError
     hzipWithM rethrowErrno ioops ress
   where
-    rethrowClosedError :: IOError -> IO a
-    rethrowClosedError e@IOError{} =
-        -- Pattern matching on the error is brittle, because the structure of
-        -- the exception might change between versions of @blockio-uring@.
-        -- Nonetheless, it's better than nothing.
-        if isResourceVanishedError e && ioe_location e == "IOCtx closed"
-          then throwIO (IOI.mkClosedError (SomeHasFS hasFS) "submitIO")
-          else throwIO e
+    rethrowFsError :: IOError -> IO a
+    rethrowFsError e@IOError{}
+      -- Pattern matching on the error is brittle, because the structure of
+      -- the exception might change between versions of @blockio-uring@.
+      -- Nonetheless, it's better than nothing.
+      | isResourceVanishedError e
+      , ioe_location e == "IOCtx closed"
+      = throwIO (IOI.mkClosedError hasFS "submitIO")
+      | ioeGetErrorType e == InvalidArgument
+      , ioe_location e == "MutableByteArray is unpinned"
+      = throwIO (IOI.mkNotPinnedError hasFS "submitIO")
+      | otherwise
+      = throwIO e
 
     rethrowErrno ::
          HasCallStack

blockio/src/System/FS/BlockIO/API.hs

@@ -88,6 +88,9 @@ data HasBlockIO m h = HasBlockIO {
     --
     -- If any of the I\/O operations fails, an 'FsError' exception will be thrown.
     --
+    -- The buffers in the 'IOOp's should be pinned memory. If any buffer is
+    -- unpinned memory, an 'FsError' exception will be thrown.
+    --
   , submitIO :: HasCallStack => V.Vector (IOOp (PrimState m) h) -> m (VU.Vector IOResult)
 
     -- TODO: once file caching is disabled, subsequent reads/writes with

blockio/src/System/FS/BlockIO/IO/Internal.hs

@@ -5,18 +5,20 @@ module System.FS.BlockIO.IO.Internal (
     IOCtxParams (..)
   , defaultIOCtxParams
   , mkClosedError
+  , mkNotPinnedError
   , tryLockFileIO
   , createHardLinkIO
   ) where
 
 import           Control.DeepSeq (NFData (..))
 import           Control.Monad.Class.MonadThrow (MonadCatch (bracketOnError),
                      MonadThrow (..), bracketOnError, try)
-import           GHC.IO.Exception (IOErrorType (ResourceVanished))
+import           GHC.IO.Exception
+                     (IOErrorType (InvalidArgument, ResourceVanished))
 import qualified GHC.IO.Handle.Lock as GHC
 import           GHC.Stack (HasCallStack)
 import qualified System.FS.API as FS
-import           System.FS.API (FsError (..), FsPath, HasFS, SomeHasFS (..))
+import           System.FS.API (FsError (..), FsPath, HasFS)
 import           System.FS.BlockIO.API (LockFileHandle (..))
 import           System.FS.IO (HandleIO)
 import qualified System.IO as GHC
@@ -54,12 +56,27 @@ defaultIOCtxParams = IOCtxParams {
       ioctxConcurrencyLimit = 64 * 3
     }
 
-mkClosedError :: HasCallStack => SomeHasFS m -> String -> FsError
-mkClosedError (SomeHasFS hasFS) loc = FS.ioToFsError (FS.mkFsErrorPath hasFS (FS.mkFsPath [])) ioerr
-  where ioerr =
-          ioeSetErrorString
-            (mkIOError ResourceVanished loc Nothing Nothing)
-            ("HasBlockIO closed: " <> loc)
+{-------------------------------------------------------------------------------
+  Errors
+-------------------------------------------------------------------------------}
+
+mkClosedError :: HasCallStack => HasFS m h -> String -> FsError
+mkClosedError hasFS loc =
+    FS.ioToFsError (FS.mkFsErrorPath hasFS (FS.mkFsPath [])) ioerr
+  where
+    ioerr =
+        ioeSetErrorString
+          (mkIOError ResourceVanished loc Nothing Nothing)
+          ("HasBlockIO closed: " <> loc)
+
+mkNotPinnedError :: HasCallStack => HasFS m h -> String -> FsError
+mkNotPinnedError hasFS loc =
+    FS.ioToFsError (FS.mkFsErrorPath hasFS (FS.mkFsPath [])) ioerr
+  where
+    ioerr =
+        ioeSetErrorString
+          (mkIOError InvalidArgument loc Nothing Nothing)
+          ("MutableByteArray is unpinned: " <> loc)
 
 {-------------------------------------------------------------------------------
   File locks

blockio/src/System/FS/BlockIO/Serial.hs

@@ -6,6 +6,7 @@ import           Control.Concurrent.Class.MonadMVar
 import           Control.Monad (unless)
 import           Control.Monad.Class.MonadThrow
 import           Control.Monad.Primitive (PrimMonad, PrimState, RealWorld)
+import           Data.Primitive (MutableByteArray, isMutableByteArrayPinned)
 import qualified Data.Vector as V
 import qualified Data.Vector.Unboxed as VU
 import qualified Data.Vector.Unboxed.Mutable as VUM
@@ -59,7 +60,9 @@ data IOCtx m = IOCtx { ctxFS :: SomeHasFS m, openVar :: MVar m Bool }
 {-# SPECIALISE guardIsOpen :: IOCtx IO -> IO () #-}
 guardIsOpen :: (HasCallStack, MonadMVar m, MonadThrow m) => IOCtx m -> m ()
 guardIsOpen ctx = readMVar (openVar ctx) >>= \b ->
-    unless b $ throwIO (IOI.mkClosedError (ctxFS ctx) "submitIO")
+    case ctxFS ctx of
+      SomeHasFS hfs ->
+        unless b $ throwIO $ IOI.mkClosedError hfs "submitIO"
 
 {-# SPECIALISE initIOCtx :: SomeHasFS IO -> IO (IOCtx IO) #-}
 initIOCtx :: MonadMVar m => SomeHasFS m -> m (IOCtx m)
@@ -90,11 +93,24 @@ ioop ::
   => HasFS m h
   -> IOOp (PrimState m) h
   -> m IOResult
-ioop hfs (IOOpRead h off buf bufOff c) =
+ioop hfs (IOOpRead h off buf bufOff c) = do
+    guardPinned hfs buf "submitIO"
     IOResult <$> hGetBufExactlyAt hfs h buf bufOff c (fromIntegral off)
-ioop hfs (IOOpWrite h off buf bufOff c) =
+ioop hfs (IOOpWrite h off buf bufOff c) = do
+    guardPinned hfs buf "submitIO"
     IOResult <$> hPutBufExactlyAt hfs h buf bufOff c (fromIntegral off)
 
+{-# SPECIALISE guardPinned :: HasFS IO h -> MutableByteArray RealWorld -> String -> IO () #-}
+guardPinned ::
+     MonadThrow m
+  => HasFS m h
+  -> MutableByteArray (PrimState m)
+  -> String
+  -> m ()
+guardPinned hfs buf loc =
+    unless (isMutableByteArrayPinned buf) $
+      throwIO (IOI.mkNotPinnedError hfs loc)
+
 {-# SPECIALISE hmapM ::
      VUM.Unbox b
   => (a -> IO b)

blockio/test/Main.hs

@@ -5,20 +5,24 @@ module Main (main) where
 import           Control.Concurrent (modifyMVar_, newMVar, threadDelay,
                      withMVar)
 import           Control.Concurrent.Async
-import           Control.Exception (SomeException (SomeException), bracket, try)
+import           Control.Exception (Exception (..),
+                     SomeException (SomeException), bracket, try)
 import           Control.Monad
 import           Control.Monad.Primitive
 import           Data.ByteString (ByteString)
 import qualified Data.ByteString as BS
 import qualified Data.ByteString.Char8 as BSC
+import qualified Data.ByteString.Lazy as LBS
 import           Data.Foldable (traverse_)
 import           Data.Functor.Compose (Compose (Compose))
+import qualified Data.List as List
 import           Data.Maybe (catMaybes)
 import           Data.Primitive.ByteArray
 import           Data.Typeable
 import qualified Data.Vector as V
 import qualified Data.Vector.Unboxed as VU
 import           System.FS.API
+import qualified System.FS.API.Lazy as FS
 import qualified System.FS.API.Strict as FS
 import           System.FS.API.Strict (hPutAllStrict)
 import qualified System.FS.BlockIO.API as FS
@@ -40,7 +44,18 @@ tests = testGroup "blockio:test" [
     , testCase "example_closeIsIdempotent" example_closeIsIdempotent
     , testProperty "prop_readWrite" prop_readWrite
     , testProperty "prop_submitToClosedCtx" prop_submitToClosedCtx
+
+      -- Context
+    , testProperty "prop_submitIO_contextClosed" prop_submitIO_contextClosed
+
+      -- Pinned vs. unpinned buffers
+    , testProperty "prop_submitIO_buffersPinned" prop_submitIO_buffersPinned
+    , testProperty "prop_submitIO_buffersUnpinned" prop_submitIO_buffersUnpinned
+
+      -- File locks
     , testProperty "prop_tryLockFileExclusiveTwice" prop_tryLockFileExclusiveTwice
+
+      -- Storage synchronisation
     , testProperty "prop_synchronise" prop_synchronise
     , testProperty "prop_synchroniseFile_fileDoesNotExist"
         prop_synchroniseFile_fileDoesNotExist
@@ -118,6 +133,104 @@ prop_submitToClosedCtx bs = ioProperty $ withSystemTempDirectory "prop_a" $ \dir
                 Left _  -> Just $ tabulate "submitIO successful" [show False] $ counterexample "expected failure, but got success" (b === True)
                 Right _ -> Just $ tabulate "submitIO successful" [show True]  $ counterexample "expected success, but got failure" (b === False)
 
+{-------------------------------------------------------------------------------
+  Closed context
+-------------------------------------------------------------------------------}
+
+-- | Test that 'submitIO' on a closed context returns a "context closed" error
+prop_submitIO_contextClosed :: Property
+prop_submitIO_contextClosed =
+    ioProperty $
+    withTempIOHasBlockIO "prop_submitIO_unpinnedBuffers" $ \hfs hbio ->
+    FS.withFile hfs path (FS.ReadWriteMode FS.MustBeNew) $ \h -> do
+      void $ FS.hPutAll hfs h $ LBS.pack [1..100]
+      buf <- newByteArray 17
+      let ioops = V.fromList [
+              IOOpWrite h 0 buf 0 17
+            , IOOpRead h 0 buf 0 17
+            ]
+      close hbio
+      eith <- try @FsError $ submitIO hbio ioops
+      pure $ case eith of
+        Left e
+          | isClosedError e
+          -> property True
+          | otherwise
+          -> counterexample ("Unexpected error: " <> displayException e) False
+        Right _
+          -> counterexample ("Unexpected success") False
+  where
+    path = FS.mkFsPath ["temp-file"]
+
+-- TODO: add a property that checks @isClosedError . mkClosedError = True@
+isClosedError :: FsError -> Bool
+isClosedError e
+    -- TODO: add an FsResourceVanished constructor to FsErrorType?
+  | fsErrorType e == FsOther
+  , "HasBlockIO closed: " `List.isPrefixOf` (fsErrorString e)
+  = True
+  | otherwise
+  = False
+
+{-------------------------------------------------------------------------------
+  Pinned vs. unpinned buffers
+-------------------------------------------------------------------------------}
+
+-- | Test that 'submitIO' using pinned buffers returns /no/ "unpinned buffers"
+-- error
+prop_submitIO_buffersPinned :: Property
+prop_submitIO_buffersPinned =
+    ioProperty $
+    withTempIOHasBlockIO "prop_submitIO_pinnedBuffers" $ \hfs hbio ->
+    FS.withFile hfs path (FS.ReadWriteMode FS.MustBeNew) $ \h -> do
+      void $ FS.hPutAll hfs h $ LBS.pack [1..100]
+      buf <- newPinnedByteArray 17
+      let ioops = V.fromList [
+              IOOpWrite h 0 buf 0 17
+            , IOOpRead h 0 buf 0 17
+            ]
+      eith <- try @FsError $ submitIO hbio ioops
+      pure $ case eith of
+        Left e
+          -> counterexample ("Unexpected error: " <> displayException e) False
+        Right _
+          -> property True
+  where
+    path = FS.mkFsPath ["temp-file"]
+
+-- | Test that 'submitIO' using unpinned buffers returns an "unpinned buffers" error
+prop_submitIO_buffersUnpinned :: Property
+prop_submitIO_buffersUnpinned =
+    ioProperty $
+    withTempIOHasBlockIO "prop_submitIO_unpinnedBuffers" $ \hfs hbio ->
+    FS.withFile hfs path (FS.ReadWriteMode FS.MustBeNew) $ \h -> do
+      void $ FS.hPutAll hfs h $ LBS.pack [1..100]
+      buf <- newByteArray 17
+      let ioops = V.fromList [
+              IOOpWrite h 0 buf 0 17
+            , IOOpRead h 0 buf 0 17
+            ]
+      eith <- try @FsError $ submitIO hbio ioops
+      pure $ case eith of
+        Left e
+          | isNotPinnedError e
+          -> property True
+          | otherwise
+          -> counterexample ("Unexpected error: " <> displayException e) False
+        Right _
+          -> counterexample ("Unexpected success") False
+  where
+    path = FS.mkFsPath ["temp-file"]
+
+-- TODO: add a property that checks @isNotPinnedError . mkNotPinnedError = True@
+isNotPinnedError :: FsError -> Bool
+isNotPinnedError e
+  | fsErrorType e == FsInvalidArgument
+  , "MutableByteArray is unpinned: " `List.isPrefixOf` (fsErrorString e)
+  = True
+  | otherwise
+  = False
+
 {-------------------------------------------------------------------------------
   File locks
 -------------------------------------------------------------------------------}