blockio-uring (empty) → 0.1.0.0
raw patch · 10 files changed
+1710/−0 lines, 10 filesdep +asyncdep +basedep +blockio-uring
Dependencies added: async, base, blockio-uring, containers, primitive, quickcheck-classes, random, tasty, tasty-hunit, tasty-quickcheck, time, unix, vector
Files
- CHANGELOG.md +5/−0
- LICENSE +26/−0
- README.md +166/−0
- benchmark/Bench.hs +246/−0
- blockio-uring.cabal +140/−0
- src/System/IO/BlockIO.hs +553/−0
- src/System/IO/BlockIO/URing.hs +246/−0
- src/System/IO/BlockIO/URingFFI.hsc +127/−0
- test/test-internals.hs +61/−0
- test/test.hs +140/−0
+ CHANGELOG.md view
@@ -0,0 +1,5 @@+# Revision history for blockio-uring++## 0.1.0.0 -- 2025-06-09++* First release
+ LICENSE view
@@ -0,0 +1,26 @@+Copyright 2022-2025 Well-Typed LLP++Redistribution and use in source and binary forms, with or without modification,+are permitted provided that the following conditions are met:++1. Redistributions of source code must retain the above copyright notice, this+ list of conditions and the following disclaimer.++2. Redistributions in binary form must reproduce the above copyright notice,+ this list of conditions and the following disclaimer in the documentation+ and/or other materials provided with the distribution.++3. Neither the name of the copyright holder nor the names of its contributors+ may be used to endorse or promote products derived from this software without+ specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS “AS IS” AND+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,166 @@+# blockio-uring++This library supports disk I/O operations using the Linux `io_uring` API. The+library supports submitting large batches of I/O operations in one go. It also+supports submitting batches from multiple Haskell threads concurrently. The I/O+only blocks the calling thread, not all other Haskell threads. In this style,+using a combination of batching and concurrency, it is possible to saturate+modern SSDs, thus achieving maximum I/O throughput. This is particularly helpful+for performing lots of random reads.++The library only supports recent versions of Linux, because it uses the `io_uring`+kernel API. It only supports disk operations, not socket operations. The library+is tested only with Ubuntu (versions `22.04` and `24.04`), but other Linux+distributions should probably also work out of the box. Let us know if you run+into any problems!++## Installation++1. Install `liburing` version 2.1 or higher. Use your package manager of choice,+ or clone a recent version of https://github.com/axboe/liburing and install+ using `make install`.+2. Invoke `cabal build` or `cabal run`.++## Benchmarks++We can compare the I/O performance that can be achieved using library against+the best case of what the system can do. The most interesting comparison is+for performing random 4k reads, and measuring the IOPS -- the I/O operations+per second.++### Baseline using `fio`++We can use the `fio` (flexible I/O tester) tool to give us a baseline for the+best that the system can manage. While manufacturers often claim that SSDs can+hit certain speeds or IOPs for specific workloads (like random reads at queue+depth 32), `fio` gives a more realistic number that includes all the overheads+that are present in practice, such as overheads induced by file systems,+encrypted block devices, etc. This makes `fio` a sensible baseline.++The repo contains an `fio` configuration file for a random read benchmark using+`io_uring`, which we can use like so:++```bash+❯ fio ./benchmark/randread.fio+```++This will produce a page full of output, like so:++```+ read: IOPS=234k, BW=915MiB/s (960MB/s)(1024MiB/1119msec)+ slat (usec): min=41, max=398, avg=70.98, stdev=15.76+ clat (usec): min=59, max=4611, avg=342.75, stdev=199.02+ lat (usec): min=121, max=4863, avg=413.75, stdev=200.25+ clat percentiles (usec):+ | 1.00th=[ 86], 5.00th=[ 102], 10.00th=[ 120], 20.00th=[ 174],+ | 30.00th=[ 262], 40.00th=[ 289], 50.00th=[ 318], 60.00th=[ 343],+ | 70.00th=[ 371], 80.00th=[ 490], 90.00th=[ 586], 95.00th=[ 652],+ | 99.00th=[ 922], 99.50th=[ 1090], 99.90th=[ 1598], 99.95th=[ 2180],+ | 99.99th=[ 4015]+ bw ( KiB/s): min=938280, max=946296, per=100.00%, avg=942288.00, stdev=5668.17, samples=2+ iops : min=234570, max=236574, avg=235572.00, stdev=1417.04, samples=2+ lat (usec) : 100=4.32%, 250=21.96%, 500=54.23%, 750=16.10%, 1000=2.74%+ lat (msec) : 2=0.60%, 4=0.05%, 10=0.01%+ cpu : usr=10.20%, sys=64.67%, ctx=66125, majf=0, minf=138+ IO depths : 1=0.0%, 2=0.0%, 4=0.0%, 8=0.0%, 16=0.0%, 32=0.1%, >=64=100.0%+ submit : 0=0.0%, 4=0.0%, 8=0.0%, 16=0.0%, 32=100.0%, 64=0.0%, >=64=0.0%+ complete : 0=0.0%, 4=100.0%, 8=0.0%, 16=0.0%, 32=0.0%, 64=0.1%, >=64=0.0%+ issued rwts: total=262144,0,0,0 short=0,0,0,0 dropped=0,0,0,0+ latency : target=0, window=0, percentile=100.00%, depth=128+```++The headline number to focus on is this bit:++```+ read: IOPS=234k+```++### Caching++By default, the `fio` configuration file declares that the random read benchmark+should use direct I/O, which bypasses any caching of file contents by the+operating system. This means that `fio` measures the raw performance of the disk+hardware. If you change the configuration file to use direct I/O, then the+reported statistics will be different because of caching. It is possible to+manually drop caches before running the benchmark to reduce the effects of+caching by invoking:++```bash+❯ sudo sysctl -w vm.drop_caches=1+```++### Haskell benchmarks++The `bench` executable expects three arguments: (1) a benchmark name, either+`low` or `high`, (2) a caching mode, either `Cache` or `NoCache`, and (3) a+filepath to a data file.++* The `low` benchmark targets low-level, internal code of the library. The+`high` benchmark uses the public API that is exposed from `blockio-uring`.++* `NoCache` enables direct I/O, while `Cache` disables it.++* The filepath to a data file will be used to read bytes from during the+ benchmark.++For a fair comparison with `fio` we use the same file that `fio` generated for+its own benchmark, `./benchmark/benchfile.0.0`. For example, we can invoke the+high-level benchmark with direct I/O as follows:++```bash+❯ cabal run bench -- high NoCache ./benchmark/benchfile.0.0+```++This will report some statistics:++```+High-level API benchmark+File caching: False+Capabilities: 1+Threads : 4+Total I/O ops: 262016+Elapsed time: 1.216050853s+IOPS: 215465+Allocated total: 24883952+Allocated per: 95+```++Though the output prints a number of useful metrics, the headline number to+focus on here is this bit:++```+IOPS: 215465+```++### Comparing results++For comparisons, we are primarily interested in IOPS.++On a maintainer's laptop (@jdral) with direct I/O enabled and using a single OS+thread the numbers in question are:++* `fio`: 234k IOPS+* High-level Haskell benchmark: 215k IOPS+* Low-level Haskell benchmark: 208k IOPS++So as a rough conclusion, we can get about 92% of the maximum IOPS using the+high-level Haskell API, or about 88% when using the low-level internals. On some+other machines, it was observed that the high-level benchmark could even match+the IOPs reported by `fio`. It might be unintuitive that the high-level+benchmark gets higher numbers than the low-level benchmark. However, the+high-level benchmarks uses multiple lightweight Haskell threads, which allows+the benchmark to have multiple I/O batches in flight at once even when using one+OS thread, while the low-level benchmark only submits I/O batches serially.++### Multi-threading++Note that much IOPs can be achieved by running the benchmarks with more than one+OS thread. This allows (more) I/O operations to be in flight at once, which+better utilises the bandwidth of modern SSDs. For example, if we want to use 4+OS threads for our benchmarks, then:++* In the `fio` configuration file we can add `numjobs=4` to fork 4 identical+ jobs that perform I/O concurrently.++* In the Haskell benchmarks we can configure the GHC run-time system to use 4 OS+ threads (capabilities) by passing `+RTS -N4 -RTS` to the benchmark executable.
+ benchmark/Bench.hs view
@@ -0,0 +1,246 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE LambdaCase #-}+{- HLINT ignore "Use camelCase" -}++module Main (main) where++import Data.Primitive+import qualified Data.Set as Set+import Control.Monad+import Control.Monad.Primitive (RealWorld)+import Control.Monad.ST (ST, stToIO)+import Control.Exception+import Control.Concurrent (getNumCapabilities)+import Control.Concurrent.Async as Async++import Foreign+import System.Posix.IO+import System.Posix.Files+import System.Posix.Types as Posix+import System.Posix.Fcntl++import System.Random as Random+import System.Environment+import System.Exit+import System.Mem (performMajorGC)+import Data.Time+import qualified GHC.Stats as RTS++import System.IO.BlockIO+import System.IO.BlockIO.URing hiding (submitIO)+import qualified System.IO.BlockIO.URing as URing++import qualified Data.Vector as V+import qualified Data.Vector.Mutable as VM++main :: IO ()+main = do+ args <- getArgs+ case args of+ [level, cache, filename]+ | Just main_func <- parseLevel level+ , Just useCache <- parseCacheFlag cache+ -> main_func useCache filename+ _ -> do+ putStrLn "Usage: Bench [low|high] [Cache|NoCache] [DataFile]"+ exitFailure+ where+ parseLevel = \case+ "low" -> Just main_lowlevel+ "high" -> Just main_highlevel+ _ -> Nothing+ parseCacheFlag = \case+ "Cache" -> Just True+ "NoCache" -> Just False+ _ -> Nothing++main_lowlevel :: Bool -> FilePath -> IO ()+main_lowlevel useCache filename = do+ putStrLn "Low-level API benchmark"+ fd <- openFd filename ReadOnly defaultFileFlags++ fileSetCaching fd useCache+ putStrLn $ "File caching: " ++ show useCache++ status <- getFdStatus fd+ let size = fileSize status+ lastBlock :: Int+ lastBlock = fromIntegral (size `div` 4096 - 1)+ nqueue = 64+ nbufs = 64 * 4+ withURing (URingParams nqueue nbufs) $ \uring ->+ allocaBytesAligned (4096 * nbufs) 4096 $ \bufptr -> do+ let submitBatch :: [(Int, Int)] -> IO ()+ submitBatch blocks = do+ sequence_+ [ prepareRead uring fd blockoff bufptr' 4096+ (IOOpId (fromIntegral i))+ | (i, block) <- blocks+ , let bufptr' = bufptr `plusPtr` ((i `mod` nbufs) * 4096)+ blockoff = fromIntegral (block * 4096)+ ]+ URing.submitIO uring++ collectBatch :: Int -> IO ()+ collectBatch n =+ replicateM_ n $ do+ (IOCompletion i count) <- awaitIO uring+ when (count /= IOResult 4096) $+ fail $ "I/O failure: I/O " ++ show i+ ++ " returned " ++ show count++ go [] = return ()+ go blocks = do+ let (batch, blocks') = splitAt 32 blocks+ submitBatch batch+ let n = case blocks' of+ [] -> length batch+ _ -> 32+ collectBatch n+ go blocks'++ rng <- initStdGen+ let blocks = zip [0..] (randomPermute rng [0..lastBlock])+ totalOps = lastBlock + 1+ (leadIn, blocks') = splitAt 64 blocks++ withReport totalOps $ do+ submitBatch leadIn+ go blocks'+ collectBatch 64+++{-# NOINLINE main_highlevel #-}+main_highlevel :: Bool -> FilePath -> IO ()+main_highlevel useCache filename = do+ putStrLn "High-level API benchmark"+ fd <- openFd filename ReadOnly defaultFileFlags++ fileSetCaching fd useCache+ putStrLn $ "File caching: " ++ show useCache++ status <- getFdStatus fd+ rng <- initStdGen+ ncaps <- getNumCapabilities+ let size = fileSize status+ lastBlock :: Int+ lastBlock = fromIntegral (size `div` 4096 - 1)+ params = IOCtxParams {+ ioctxBatchSizeLimit = 64,+ ioctxConcurrencyLimit = 64 * 4+ }+ ntasks = 4 * ncaps+ batchsz = 32+ nbatches = lastBlock `div` (ntasks * batchsz) -- batches per task+ totalOps = nbatches * batchsz * ntasks++ putStrLn $ "Capabilities: " ++ show ncaps+ putStrLn $ "Threads : " ++ show ntasks++ bracket (initIOCtx params) closeIOCtx $ \ioctx ->+ withReport totalOps $ do+ tasks <-+ forRngSplitM ntasks rng $ \ n !rng_task ->+ Async.asyncOn n $ do+ buf <- newAlignedPinnedByteArray (4096 * batchsz) 4096+ -- Each thread gets its own reusable vector+ (v, unsafeGenerateIOOpsBatch) <- stToIO $ mkGenerateIOOpsBatch fd buf lastBlock batchsz+ forRngSplitM_ nbatches rng_task $ \ _ !rng_batch -> do+ stToIO $ unsafeGenerateIOOpsBatch rng_batch+ submitIO ioctx v+ _ <- Async.waitAnyCancel tasks+ return ()++{-# NOINLINE mkGenerateIOOpsBatch #-}+-- | Create a reusable vector and an /unsafe/ function to fill the vector with+-- random elements.+--+-- The returned vector is immutable, but the filler function mutates the vector+-- in place. The user should make sure not to use (e.g., read) the vector while+-- it's being filled.+mkGenerateIOOpsBatch :: Posix.Fd+ -> MutableByteArray RealWorld+ -> Int+ -> Int+ -> ST RealWorld ( V.Vector (IOOp RealWorld)+ , -- fill the vector with new randomly generated IOOps+ Random.StdGen -> ST RealWorld ()+ )+mkGenerateIOOpsBatch !fd !buf !lastBlock !size = do+ v <- VM.new size+ v' <- V.unsafeFreeze v+ pure (v', \rng -> go v rng 0)+ where+ go :: V.MVector RealWorld (IOOp RealWorld) -> Random.StdGen -> Int -> ST RealWorld ()+ go !_ !_ !i | i == size = return ()+ go !v !rng !i = do+ let (!block, !rng') = Random.uniformR (0, lastBlock) rng+ !bufOff = i * 4096+ !blockoff = fromIntegral (block * 4096)+ VM.unsafeWrite v i $! IOOpRead fd blockoff buf bufOff 4096+ go v rng' (i+1)++{-# INLINE forRngSplitM_ #-}+forRngSplitM_ :: Monad m+ => Int+ -> Random.StdGen+ -> (Int -> Random.StdGen -> m a)+ -> m ()+forRngSplitM_ n rng0 action = go 0 rng0+ where+ go !i !_ | i == n = return ()+ go !i !rng = let (!rng', !rng'') = Random.splitGen rng+ in action i rng' >> go (i+1) rng''++{-# INLINE forRngSplitM #-}+forRngSplitM :: Monad m+ => Int+ -> Random.StdGen+ -> (Int -> Random.StdGen -> m a)+ -> m [a]+forRngSplitM n rng0 action = go [] 0 rng0+ where+ go acc !i !_ | i == n = return (reverse acc)+ go acc !i !rng = let (!rng', !rng'') = Random.splitGen rng+ in action i rng' >>= \x -> go (x:acc) (i+1) rng''++{-# INLINE withReport #-}+withReport :: Int -> IO () -> IO ()+withReport totalOps action = do+ performMajorGC+ beforeRTS <- RTS.getRTSStats+ beforeTime <- getCurrentTime+ action+ afterTime <- getCurrentTime+ performMajorGC+ afterRTS <- RTS.getRTSStats+ report beforeTime afterTime beforeRTS afterRTS totalOps++{-# NOINLINE report #-}+report :: UTCTime -> UTCTime -> RTS.RTSStats -> RTS.RTSStats -> Int -> IO ()+report beforeTime afterTime beforeRTS afterRTS totalOps = do+ putStrLn $ "Total I/O ops: " ++ show totalOps+ putStrLn $ "Elapsed time: " ++ show elapsed+ putStrLn $ "IOPS: " ++ show iops+ putStrLn $ "Allocated total: " ++ show allocated+ putStrLn $ "Allocated per: " ++ show apio+ where+ elapsed = afterTime `diffUTCTime` beforeTime+ allocated = RTS.allocated_bytes afterRTS - RTS.allocated_bytes beforeRTS++ iops, apio :: Int+ iops = round (fromIntegral totalOps / realToFrac elapsed :: Double)+ apio = round (fromIntegral allocated / realToFrac totalOps :: Double)++randomPermute :: Ord a => StdGen -> [a] -> [a]+randomPermute rng0 xs0 =+ go (Set.fromList xs0) rng0+ where+ go !xs !_rng | Set.null xs = []++ go !xs !rng = x : go xs' rng'+ where+ (i, rng') = uniformR (0, Set.size xs - 1) rng+ !x = Set.elemAt i xs+ !xs' = Set.deleteAt i xs+
+ blockio-uring.cabal view
@@ -0,0 +1,140 @@+cabal-version: 3.4+name: blockio-uring+version: 0.1.0.0+synopsis: Perform batches of asynchronous disk IO operations.+description:+ This library supports disk I/O operations using the Linux io_uring API. The+ library supports submitting large batches of I/O operations in one go. It also+ supports submitting batches from multiple Haskell threads concurrently. The+ I/O only blocks the calling thread, not all other Haskell threads. In this+ style, using a combination of batching and concurrency, it is possible to+ saturate modern SSDs, thus achieving maximum I/O throughput. This is+ particularly helpful for performing lots of random reads.++ The library only supports recent versions of Linux, because it uses the+ io_uring kernel API. It only supports disk operations, not socket operations.+ The library is tested only with Ubuntu (versions 22.04 and 24.04), but+ other Linux distributions should probably also work out of the box. Let us+ know if you run into any problems!++license: BSD-3-Clause+license-file: LICENSE+author: Duncan Coutts+maintainer: duncan@well-typed.com, joris@well-typed.com+copyright: (c) Well-Typed LLP 2022 - 2025+category: System+build-type: Simple+tested-with: GHC ==9.2 || ==9.4 || ==9.6 || ==9.8 || ==9.10 || ==9.12+extra-doc-files:+ CHANGELOG.md+ README.md++source-repository head+ type: git+ location: https://github.com/well-typed/blockio-uring++source-repository this+ type: git+ location: https://github.com/well-typed/blockio-uring+ tag: blockio-uring-0.1.0.0++common warnings+ ghc-options:+ -Wall -Wcompat -Wincomplete-uni-patterns+ -Wincomplete-record-updates -Wpartial-fields -Widentities+ -Wredundant-constraints -Wmissing-export-lists+ -Wno-unticked-promoted-constructors -Wunused-packages++ ghc-options: -Werror=missing-deriving-strategies++common language+ default-language: GHC2021+ default-extensions:+ DeriveAnyClass+ DerivingStrategies+ DerivingVia+ ExplicitNamespaces+ GADTs+ LambdaCase+ RecordWildCards+ RoleAnnotations+ ViewPatterns++library+ import: language, warnings+ exposed-modules: System.IO.BlockIO+ hs-source-dirs: src+ other-modules:+ System.IO.BlockIO.URing+ System.IO.BlockIO.URingFFI++ build-depends:+ , base >=4.16 && <4.22+ , primitive ^>=0.8 || ^>=0.9+ , vector ^>=0.13++ -- Annoyingly, liburing-2.1 has the wrong version in its liburing.pc file,+ -- namely version 2.0. So, even though we are actually only supporting >=2.1,+ -- we have to use >=2.0 here.+ pkgconfig-depends: liburing >=2.0 && <2.10++benchmark bench+ import: language, warnings+ type: exitcode-stdio-1.0+ hs-source-dirs: benchmark src+ main-is: Bench.hs+ build-depends:+ , async+ , base <5+ , containers+ , primitive+ , random+ , time+ , unix ^>=2.8.7.0+ , vector++ pkgconfig-depends: liburing+ other-modules:+ System.IO.BlockIO+ System.IO.BlockIO.URing+ System.IO.BlockIO.URingFFI++ ghc-options: -threaded -with-rtsopts=-T++test-suite test+ import: language, warnings+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: test.hs+ build-depends:+ , base <5+ , blockio-uring+ , primitive+ , tasty+ , tasty-hunit+ , tasty-quickcheck+ , vector++ ghc-options: -threaded++test-suite test-internals+ import: language, warnings+ type: exitcode-stdio-1.0+ hs-source-dirs: test src+ main-is: test-internals.hs+ build-depends:+ , base <5+ , primitive+ , quickcheck-classes+ , tasty+ , tasty-hunit+ , tasty-quickcheck+ , vector++ pkgconfig-depends: liburing+ other-modules:+ System.IO.BlockIO+ System.IO.BlockIO.URing+ System.IO.BlockIO.URingFFI++ ghc-options: -threaded -fno-ignore-asserts
+ src/System/IO/BlockIO.hs view
@@ -0,0 +1,553 @@+{-# LANGUAGE MultiWayIf #-}+{-# LANGUAGE TypeFamilies #-}++module System.IO.BlockIO (++ -- * I\/O context and initialisation+ IOCtx,+ IOCtxParams(..),+ defaultIOCtxParams,+ withIOCtx,+ initIOCtx,+ closeIOCtx,++ -- * Performing I\/O+ submitIO,+ IOOp(IOOpRead, IOOpWrite),+ IOResult(IOResult, IOError),+ ByteCount, Errno(..),++ ) where++import Data.Bits+import Data.Primitive.ByteArray+import qualified Data.Vector as V+import qualified Data.Vector.Mutable as VM+import qualified Data.Vector.Unboxed as VU+import qualified Data.Vector.Unboxed.Mutable as VUM++import Control.Monad+import Control.Monad.Primitive+import Control.Concurrent (forkOn, myThreadId, threadCapability,+ getNumCapabilities)+import Control.Concurrent.MVar+import Control.Concurrent.QSemN+import Control.Concurrent.Chan+import Control.Exception (mask_, throw, ArrayException(UndefinedElement),+ finally, assert, throwIO, bracket, onException)+import System.IO.Error+import GHC.IO.Exception (IOErrorType(ResourceVanished, InvalidArgument))+import GHC.Conc.Sync (labelThread)++import Foreign.Ptr (plusPtr)+import Foreign.C.Error (Errno(..))+import System.Posix.Types (Fd (..), FileOffset, ByteCount)+import System.Posix.Internals (hostIsThreaded)++import qualified System.IO.BlockIO.URing as URing+import System.IO.BlockIO.URing (IOResult(..))++-- | IO context: a handle used by threads submitting IO batches.+--+-- Internally, each GHC capability in the program creates its own independent IO+-- context. This means that each capability can process batches of I/O+-- operations independently. As such, running with more capabilities can+-- increase throughput.+newtype IOCtx = IOCtx (V.Vector IOCapCtx) -- one per RTS capability.++type CapNo = Int+data IOCapCtx = IOCapCtx {+ -- | This is initialised from the 'ioctxBatchSizeLimit' from the 'IOCtxParams'.+ ioctxBatchSizeLimit' :: !Int,++ -- | IO concurrency control: used by writers to reserve the+ -- right to submit an IO batch of a given size, and by the+ -- completion thread to return it on batch completion.+ ioctxQSemN :: !QSemN,++ -- | Locking of the writer end of the URing: used by writers+ -- while they are modifying the uring submission queue.+ ioctxURing :: !(MVar (Maybe URing.URing)),++ -- | Communication channel from writers to the completion thread:+ -- letting it know about new batches of IO that they have+ -- submitted.+ ioctxChanIOBatch :: !(Chan IOBatch),++ -- | Communication channel from the completion thread to writers:+ -- letting them grab the next batch index, which they need when+ -- submitting IO operations.+ ioctxChanIOBatchIx :: !(Chan IOBatchIx),++ -- | An MVar to synchronise on for shutdown+ ioctxCloseSync :: !(MVar ())+ }++-- | Parameters for instantiating an 'IOCtx' IO context.+--+-- Picking suitable parameters for high performance depends on the hardware that+-- is used and the workload of the program. There are benchmarks included in the+-- [@blockio-uring@](https://github.com/well-typed/blockio-uring) repository+-- that can help gauge the performance for a random 4k read workload, but some+-- trial and error with custom benchmarks might be required.+--+-- Ideally the concurrency limit should be a multiple of the batch size limit.+-- This should ensure that there can be multiple batches in flight at once. If+-- so, the SSD can more often than not pick up new I/O batches to perform as+-- soon as it has finished other batches.+--+-- Ideally the batch size limit should be equal to if not larger than the number+-- of I/O operations that an SSD can perform concurrently. This ensures that the+-- SSD can perform most if not all I/O operations in a batch concurrently.+--+-- Internally, each GHC capability in the program creates its own independent IO+-- context. These parameters apply separately to each capability's IO context.+-- For example, if the program runs with 2 capabilities, a batch size limit of+-- 64, and a concurrency limit of 256, then each capability gets its own IO+-- context, and each IO context can process batches of at most 64 I/O operations+-- at a time, and each IO context can only process 4 such batches concurrently.+data IOCtxParams = IOCtxParams {+ -- | The maximum size of a batch of I\/O operations that can be processed as+ -- a whole.+ --+ -- Note that his does /not/ affect the size of batches of I/O operations+ -- that can be submitted using 'submitIO'. There is no restriction on the+ -- size of batches submitted using 'submitIO'. If the size of a batch that+ -- is passed to 'submitIO' exceeds this limit, then internally these batches+ -- will split into sub-batches of size at most the limit, and each sub-batch+ -- is processed individually.+ ioctxBatchSizeLimit :: !Int,+ -- | The total number of I/O operations that can be processed concurrently.+ --+ -- If a use of 'submitIO' would lead to this limit being exceeded, then the+ -- call to 'submitIO' will block until enough I/O batches have been+ -- processed.+ ioctxConcurrencyLimit :: !Int+ }+ deriving stock Show++-- | Default parameters. Some manual tuning of parameters might be required to+-- achieve higher performance targets (see 'IOCtxParams' for hints).+defaultIOCtxParams :: IOCtxParams+defaultIOCtxParams =+ IOCtxParams {+ ioctxBatchSizeLimit = 64,+ ioctxConcurrencyLimit = 64 * 3+ }++validateIOCtxParams :: IOCtxParams -> Maybe String+validateIOCtxParams IOCtxParams{..} = ("IOCtxParams are invalid because " ++) <$>+ if+ | ioctxBatchSizeLimit <= 0+ -> Just "the batch size limit is non-positive"+ | ioctxBatchSizeLimit >= 2^(15 :: Int)+ -> Just "the batch size limit is greater than or equal to 2^15"+ | ioctxConcurrencyLimit <= 0+ -> Just "the concurrency limit is non-positive"+ | ioctxConcurrencyLimit >= 2^(16 :: Int)+ -> Just "the concurrency limit is greater than or equal to 2^16"+ | ioctxBatchSizeLimit > ioctxConcurrencyLimit+ -> Just "the batch size limit exceeds the concurrency limit"+ | otherwise+ -> Nothing++withIOCtx :: IOCtxParams -> (IOCtx -> IO a) -> IO a+withIOCtx params = bracket (initIOCtx params) closeIOCtx++initIOCtx :: IOCtxParams -> IO IOCtx+initIOCtx ioctxparams = do+ unless hostIsThreaded $ throwIO rtsNotThreaded+ forM_ (validateIOCtxParams ioctxparams) $ throwIO . mkInvalidArgumentError+ ncaps <- getNumCapabilities+ IOCtx <$> V.generateM ncaps (initIOCapCtx ioctxparams)+ where+ rtsNotThreaded =+ mkIOError+ illegalOperationErrorType+ "The run-time system should be threaded, make sure you are passing the -threaded flag"+ Nothing+ Nothing++ mkInvalidArgumentError :: String -> IOError+ mkInvalidArgumentError msg =+ mkIOError+ InvalidArgument+ msg+ Nothing+ Nothing++initIOCapCtx :: IOCtxParams -> CapNo -> IO IOCapCtx+initIOCapCtx IOCtxParams {+ ioctxBatchSizeLimit,+ ioctxConcurrencyLimit+ } capno = do+ mask_ $ do+ ioctxQSemN <- newQSemN ioctxConcurrencyLimit+ uring <- URing.setupURing (URing.URingParams ioctxBatchSizeLimit ioctxConcurrencyLimit)+ ioctxURing <- newMVar (Just uring)+ ioctxChanIOBatch <- newChan+ ioctxChanIOBatchIx <- newChan+ ioctxCloseSync <- newEmptyMVar+ t <- forkOn capno $+ -- Use forkOn to bind the thread to this capability+ completionThread+ uring+ ioctxCloseSync+ ioctxConcurrencyLimit+ ioctxQSemN+ ioctxChanIOBatch+ ioctxChanIOBatchIx+ labelThread t ("System.IO.BlockIO.completionThread " +++ "(for cap " ++ show capno ++ ")")+ let initialBatchIxs :: [IOBatchIx]+ initialBatchIxs = [0 .. ioctxConcurrencyLimit-1]+ writeList2Chan ioctxChanIOBatchIx initialBatchIxs+ return IOCapCtx {+ ioctxBatchSizeLimit' = ioctxBatchSizeLimit,+ ioctxQSemN,+ ioctxURing,+ ioctxChanIOBatch,+ ioctxChanIOBatchIx,+ ioctxCloseSync+ }++closeIOCtx :: IOCtx -> IO ()+closeIOCtx (IOCtx capctxs) = closeIOCapCtxs capctxs+ where+ -- If an exception was raised while closing one context, we still try to+ -- close the remaining IOCapCtxs+ closeIOCapCtxs xs+ | Just (x, ys) <- V.uncons xs+ = closeIOCapCtx x `finally` closeIOCapCtxs ys+ | otherwise+ = pure ()++closeIOCapCtx :: IOCapCtx -> IO ()+closeIOCapCtx IOCapCtx {ioctxURing, ioctxCloseSync} = do+ uringMay <- takeMVar ioctxURing+ case uringMay of+ Nothing -> putMVar ioctxURing Nothing+ Just uring -> do+ --TODO: there's a problem with the keepAlives here. By sending the+ -- Nop with maxBound we're telling the completionThread to shut down,+ -- but this may be the only thing keeping the IO buffers from being+ -- GCd. We could get heap corruption if we get a GC during shutdown.+ -- We need to prevent new operations being submitted, and wait for all+ -- existing operations to complete.+ URing.prepareNop uring (URing.IOOpId maxBound)+ URing.submitIO uring+ takeMVar ioctxCloseSync+ URing.closeURing uring+ putMVar ioctxURing Nothing++-- | The 'MutableByteArray' buffers within __must__ be pinned. Addresses into+-- these buffers are passed to @io_uring@, and the buffers must therefore not be+-- moved around. 'submitIO' will check that buffers are pinned, and will throw+-- errors if it finds any that are not pinned.+data IOOp s = IOOpRead !Fd !FileOffset !(MutableByteArray s) !Int !ByteCount+ | IOOpWrite !Fd !FileOffset !(MutableByteArray s) !Int !ByteCount++-- | Submit a batch of I\/O operations, and wait for them all to complete.+-- The sequence of results matches up with the sequence of operations.+-- Any I\/O errors are reported in the result list, not as IO exceptions.+--+-- Note that every operation in the batch is performed concurrently with+-- respect to each other (and any other concurrent batches): their effects+-- may be performed in any order. It is up to you to ensure the effects do+-- not interfere with each other (i.e. not mixing reads and writes to+-- overlapping areas of files).+--+-- It is permitted to submit multiple batches concurrently from different+-- Haskell threads. Submitting I\/O only blocks the calling Haskell thread, it+-- does not block other Haskell threads. The maximum concurrency is set when+-- the 'IOCtx' is created: submitting more operations than this will block until+-- enough previous operations have completed.+--+-- Performance tips:+--+-- * Use reasonable batch sizes to amortise the overheads over multiple+-- operations. Batch sizes that are within the I\/O limit of the+-- 'IOCtx' can be initiated with a single system call.+--+-- * Consider that most SSDs can perform up to 64 operations concurrently. So+-- use reasonable batch sizes, and submit multiple batches concurrently.+--+-- * Think of I\/O as a queue, with I\/O operations being added at one end,+-- and results arriving at the other: keep the queue full with 64 operations+-- in progress at once.+--+-- * Pipeline your I\/O submissions to keep the queue full: submit enough+-- batches to keep the queue full, and as batches complete, submit more.+-- For example follow a strategy of submitting batches up to double the+-- target SSD queue depth (i.e. 2x 64 = 128) and when it drains to nearly+-- the target depth, fill it up to double again. This way there is always+-- at least the target number in flight at once.+--+submitIO :: IOCtx -> V.Vector (IOOp RealWorld) -> IO (VU.Vector IOResult)+submitIO (IOCtx capctxs) !ioops = do+ -- Find out which capability the thread is currently running on and use+ -- that one. It does _not matter_ for correctness if the thread is migrated+ -- while the I/O is submitted or when waiting for completion. Migration+ -- happens sufficiently infrequently that it should not be a performance+ -- problem.+ tid <- myThreadId+ (capno, _) <- threadCapability tid+ let !capctx = capctxs V.! (capno `mod` V.length capctxs)+ submitCapIO capctx ioops++submitCapIO :: IOCapCtx -> V.Vector (IOOp RealWorld) -> IO (VU.Vector IOResult)+submitCapIO ioctx@IOCapCtx {ioctxBatchSizeLimit'} !ioops+ -- Typical small case. We can be more direct.+ | V.length ioops > 0 && V.length ioops <= ioctxBatchSizeLimit'+ = mask_ $ do+ iobatchCompletion <- newEmptyMVar+ prepAndSubmitIOBatch ioctx ioops iobatchCompletion+ takeMVar iobatchCompletion++submitCapIO ioctx@IOCapCtx {ioctxBatchSizeLimit'} !ioops0 =+ -- General case. Needs multiple batches and combining results, or the vector+ -- of I/O operations is empty.+ --+ --TODO: instead of the completion thread allocating result arrays+ -- allocate them in the calling thread and have the completionThread+ -- fill them in. Then for batches we can send in a bunch of slices of+ -- a contiguous array, and then we can avoid having to re-combine them+ -- at the end here.+ mask_ $ do+ iobatchCompletions <- prepAndSubmitIOBatches [] ioops0+ awaitIOBatches iobatchCompletions+ where+ prepAndSubmitIOBatches acc !ioops+ | V.null ioops = return acc+ | otherwise = do+ let batch = V.take ioctxBatchSizeLimit' ioops+ iobatchCompletion <- newEmptyMVar+ prepAndSubmitIOBatch ioctx batch iobatchCompletion+ prepAndSubmitIOBatches (iobatchCompletion:acc)+ (V.drop ioctxBatchSizeLimit' ioops)++ awaitIOBatches iobatchCompletions =+ VU.concat <$> mapM takeMVar (reverse iobatchCompletions)++-- Must be called with async exceptions masked. See mask_ above in submitIO.+prepAndSubmitIOBatch :: IOCapCtx+ -> V.Vector (IOOp RealWorld)+ -> MVar (VU.Vector IOResult)+ -> IO ()+prepAndSubmitIOBatch IOCapCtx {+ ioctxQSemN,+ ioctxURing,+ ioctxChanIOBatch,+ ioctxChanIOBatchIx+ }+ !iobatch !iobatchCompletion = do+ let !iobatchOpCount = V.length iobatch+ -- We're called with async exceptions masked, but 'waitQSemN' can block and+ -- receive exceptions. That's ok. But once we acquire the semaphore+ -- quantitiy we must eventully return it. There's two cases for returning:+ -- 1. we successfully submit the I/O and pass the information off to the+ -- completionThread which will signal the semaphore upon completion, or+ -- 2. we encounter an exception here in which case we need to undo the+ -- semaphore acquisition.+ -- For the latter case we use 'onException'. We also need to obtain a+ -- batch index. This should never block because we have as many tokens as+ -- QSemN initial quantitiy, and the batch ix is released before the QSemN+ -- is signaled in the completionThread.+ waitQSemN ioctxQSemN iobatchOpCount+ !iobatchIx <- readChan ioctxChanIOBatchIx+ -- Thus undoing the acquisition involves releasing the batch index and+ -- semaphore quantitiy (which themselves cannot blocks).+ let undoAcquisition = do writeChan ioctxChanIOBatchIx iobatchIx+ signalQSemN ioctxQSemN iobatchOpCount+ flip onException undoAcquisition $ do+ -- We can receive an async exception if takeMVar blocks. That's ok, we'll+ -- undo the acquisition.+ muring <- takeMVar ioctxURing+ -- From here on we cannot receive any async exceptions, because we do not+ -- do any more blocking operations. But we can encounter sync exceptions,+ -- so we may still need to release the mvar on exception.+ flip onException (putMVar ioctxURing muring) $ do+ uring <- maybe (throwIO closed) pure muring+ V.iforM_ iobatch $ \ioopix ioop -> case ioop of+ IOOpRead fd off buf bufOff cnt -> do+ guardPinned buf+ URing.prepareRead uring fd off+ (mutableByteArrayContents buf `plusPtr` bufOff)+ cnt (packIOOpId iobatchIx ioopix)+ IOOpWrite fd off buf bufOff cnt -> do+ guardPinned buf+ URing.prepareWrite uring fd off+ (mutableByteArrayContents buf `plusPtr` bufOff)+ cnt (packIOOpId iobatchIx ioopix)+ -- TODO: if submitIO or guardPinned throws an exception, we need to+ -- undo / clear the SQEs that we prepared.+ URing.submitIO uring++ -- More async exception safety: we want to inform the completionThread+ -- /if and only if/ we successfully submitted a batch of IO. So now that+ -- we have submitted a batch we need to inform the completionThread+ -- without interruptions. We're still masked, but writeChan does not+ -- throw exceptions and never blocks (unbounded channel) so we should+ -- not get async or sync exceptions.+ writeChan ioctxChanIOBatch+ IOBatch {+ iobatchIx,+ iobatchOpCount,+ iobatchCompletion,+ iobatchKeepAlives = iobatch+ }+ putMVar ioctxURing muring+ where+ guardPinned mba = unless (isMutableByteArrayPinned mba) $ throwIO notPinned+ closed = mkIOError ResourceVanished "IOCtx closed" Nothing Nothing+ notPinned = mkIOError InvalidArgument "MutableByteArray is unpinned" Nothing Nothing++data IOBatch = IOBatch {+ iobatchIx :: !IOBatchIx,+ iobatchOpCount :: !Int,+ iobatchCompletion :: !(MVar (VU.Vector IOResult)),+ -- | The list of I\/O operations is sent to the completion+ -- thread so that the buffers are kept alive while the kernel+ -- is using them.+ iobatchKeepAlives :: V.Vector (IOOp RealWorld)+ }++-- | We submit and processes the completions in batches. This is the index into+-- the tracking arrays of the batch.+type IOBatchIx = Int++-- | This is the index of an operation within a batch. The pair of the+-- 'IOBatchIx' and 'IOOpIx' is needed to identity a specific operation within+-- the tracking data structures.+type IOOpIx = Int++{-# INLINE packIOOpId #-}+-- | The pair of the 'IOBatchIx' and 'IOOpIx' is needed to identify a specific+-- operation within the tracking data structures. We pair up the batch index+-- and the intra-batch index into the operation identifier. This identifier is+-- submitted with the operation and returned with the completion. Thus upon+-- completion this allows us to match up the operation with the tracking data+-- structures and process the operation completion.+packIOOpId :: IOBatchIx -> IOOpIx -> URing.IOOpId+packIOOpId batchix opix =+ URing.IOOpId $ unsafeShiftL (fromIntegral batchix) 32+ .|. fromIntegral opix++{-# INLINE unpackIOOpId #-}+unpackIOOpId :: URing.IOOpId -> (IOBatchIx, IOOpIx)+unpackIOOpId (URing.IOOpId w64) =+ (batchix, opix)+ where+ batchix :: Int+ batchix = fromIntegral (unsafeShiftR w64 32)++ opix :: Int+ opix = fromIntegral (w64 .&. 0xffffffff)++-- Note: all these arrays are indexed by 'IOBatchIx'.+--+-- The 'counts' array keeps track (per batch) of the number of operations that+-- remain to complete. When we processes the last operation in a batch we can+-- complete the whole batch. Batch indexes that are not currently in use contain+-- a count value of -1. This is used to identify when an operation completes+-- that is for a previously unused batch index, and thus tells us we have a new+-- batch and we need to find and set up the tracking information appropriately.+--+-- The 'results' array keeps track (per-batch) of the results of individual I/O+-- operations. Each element is an array indexed by 'IOOpIx', containing the+-- 'IOResult' for that operation. This result array is accumulated and then+-- frozen and returned as the result for the batch. Batch indexes that are not+-- currently in use contain an invalid entry.+--+-- The 'completions' array keeps track (per batch) of the completion MVar used+-- to communicate the batch result back to the thread that submitted the batch.+--+-- The 'keepAlives' array ensures (per batch) that certain heap objects are kept+-- live for the duration of the I/O operations in the batch. Specifically, it is+-- the I/O buffers for each operation that we must keep live (otherwise if they+-- were GC'd the kernel could scribble on top of whatever got placed there+-- next). We reuse the original vector of IOOps that was submitted since this+-- conveniently exists anyway and it contains the IOOps which themselves contain+-- the I/O buffers. The 'keepAlives' entries are overwritten with 'invalidEntry'+-- once they are no longer needed.+--+-- Algorithm outline:+-- + wait for single IO result+-- + if the count is -1, grab new batches from the chan (and process them)+-- repeatedly until the batch count in question is found.+-- + if the count is positive, decrement and update result array+-- + if the count is now 0, also fill in the completion+-- + reset count to -1, and result entries to invalid+completionThread :: URing.URing+ -> MVar ()+ -> Int+ -> QSemN+ -> Chan IOBatch+ -> Chan IOBatchIx+ -> IO ()+completionThread !uring !done !maxc !qsem !chaniobatch !chaniobatchix = do+ counts <- VUM.replicate maxc (-1)+ results <- VM.replicate maxc invalidEntry+ completions <- VM.replicate maxc invalidEntry+ keepAlives <- VM.replicate maxc invalidEntry+ collectCompletion counts results completions keepAlives+ `finally` putMVar done ()+ where+ collectCompletion :: VUM.MVector RealWorld Int+ -> VM.MVector RealWorld (VUM.MVector RealWorld IOResult)+ -> VM.MVector RealWorld (MVar (VU.Vector IOResult))+ -> VM.MVector RealWorld (V.Vector (IOOp RealWorld))+ -> IO ()+ collectCompletion !counts !results !completions !keepAlives = do+ iocompletion <- URing.awaitIO uring+ let (URing.IOCompletion !ioopid !iores) = iocompletion+ unless (ioopid == URing.IOOpId maxBound) $ do+ let (!iobatchix, !ioopix) = unpackIOOpId ioopid+ count <- do+ c <- VUM.read counts iobatchix+ if c < 0 then collectIOBatches iobatchix+ else return c+ assert (count > 0) (return ())+ VUM.write counts iobatchix (count-1)+ result <- VM.read results iobatchix+ VUM.write result ioopix iores+ when (count == 1) $ do+ completion <- VM.read completions iobatchix+ VUM.write counts iobatchix (-1)+ VM.write results iobatchix invalidEntry+ VM.write completions iobatchix invalidEntry+ VM.write keepAlives iobatchix invalidEntry+ result' <- VU.unsafeFreeze result+ putMVar completion (result' :: VU.Vector IOResult)+ -- Important: release batch index _before_ we signal the QSemN.+ -- The other side needs the guarantee that the index is available+ -- once it acquires the QSemN.+ writeChan chaniobatchix iobatchix+ let !qrelease = VU.length result'+ signalQSemN qsem qrelease+ collectCompletion counts results completions keepAlives++ where+ collectIOBatches :: IOBatchIx -> IO Int+ collectIOBatches !iobatchixNeeded = do+ IOBatch{+ iobatchIx,+ iobatchOpCount,+ iobatchCompletion,+ iobatchKeepAlives+ } <- readChan chaniobatch+ oldcount <- VUM.read counts iobatchIx+ assert (oldcount == (-1)) (return ())+ VUM.write counts iobatchIx iobatchOpCount+ result <- VUM.replicate iobatchOpCount (IOResult (-1))+ VM.write results iobatchIx result+ VM.write completions iobatchIx iobatchCompletion+ VM.write keepAlives iobatchIx iobatchKeepAlives+ if iobatchIx == iobatchixNeeded+ then return $! iobatchOpCount+ else collectIOBatches iobatchixNeeded++ {-# NOINLINE invalidEntry #-}+ invalidEntry :: a+ invalidEntry =+ throw (UndefinedElement "System.IO.BlockIO.completionThread")
+ src/System/IO/BlockIO/URing.hs view
@@ -0,0 +1,246 @@+{-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE TypeFamilies #-}++module System.IO.BlockIO.URing (+ URing,+ URingParams(..),+ setupURing,+ closeURing,+ withURing,+ IOOpId(..),+ prepareRead,+ prepareWrite,+ prepareNop,+ submitIO,+ IOCompletion(..),+ IOResult(IOResult, IOError),+ awaitIO,+ ) where++import qualified Data.Vector.Generic as VG+import qualified Data.Vector.Generic.Mutable as VGM+import qualified Data.Vector.Primitive as VP+import qualified Data.Vector.Unboxed as VU+import qualified Data.Vector.Unboxed.Mutable as VUM+import qualified Data.Vector.Unboxed.Base++import Foreign+import Foreign.C+import Foreign.ForeignPtr.Unsafe+import System.IO.Error+import System.Posix.Types++import Control.Monad+import Control.Exception++import qualified System.IO.BlockIO.URingFFI as FFI+++--+-- Init+--++data URing = URing {+ -- | The uring itself.+ uringptr :: !(Ptr FFI.URing),++ -- | A pre-allocated buffer to help with FFI marshalling.+ cqeptrfptr :: {-# UNPACK #-} !(ForeignPtr (Ptr FFI.URingCQE))+ }+data URingParams = URingParams {+ sizeSQRing :: !Int,+ sizeCQRing :: !Int+ }++setupURing :: URingParams -> IO URing+setupURing URingParams { sizeSQRing, sizeCQRing } = do+ uringptr <- malloc+ cqeptrfptr <- mallocForeignPtr+ alloca $ \paramsptr -> do+ poke paramsptr params+ throwErrnoResIfNegRetry_ "setupURing" $+ FFI.io_uring_queue_init_params+ (fromIntegral sizeSQRing)+ uringptr+ paramsptr+ params' <- peek paramsptr+ -- liburing rounds up the size of the SQ ring to the nearest power of 2+ when (fromIntegral sizeSQRing > FFI.sq_entries params') $+ setupFailure uringptr $ "unexected SQ ring size "+ ++ show (sizeSQRing, FFI.sq_entries params')+ -- liburing rounds up the size of the CQ ring to the nearest power of 2+ when (fromIntegral sizeCQRing > FFI.cq_entries params') $ do+ setupFailure uringptr $ "unexected CQ ring size "+ ++ show (sizeCQRing, FFI.cq_entries params')+ return URing { uringptr, cqeptrfptr }+ where+ flags = FFI.iORING_SETUP_CQSIZE+ params = FFI.URingParams {+ FFI.sq_entries = 0,+ FFI.cq_entries = fromIntegral sizeCQRing,+ FFI.flags = flags,+ FFI.features = 0+ }+ setupFailure uringptr msg = do+ FFI.io_uring_queue_exit uringptr+ throwIO (userError $ "setupURing initialisation failure: " ++ msg)++closeURing :: URing -> IO ()+closeURing URing {uringptr} = do+ FFI.io_uring_queue_exit uringptr+ free uringptr++withURing :: URingParams -> (URing -> IO a) -> IO a+withURing params =+ bracket (setupURing params) closeURing+++--+-- Submitting I/O+--++-- | An identifier that is submitted with the I\/O operation and returned with+-- the completion.+newtype IOOpId = IOOpId Word64+ deriving stock (Eq, Ord, Bounded, Show)++prepareRead :: URing -> Fd -> FileOffset -> Ptr Word8 -> ByteCount -> IOOpId -> IO ()+prepareRead URing {uringptr} fd off buf len (IOOpId ioopid) = do+ sqeptr <- throwErrResIfNull "prepareRead" fullErrorType+ "URing I/O queue full" $+ FFI.io_uring_get_sqe uringptr+ FFI.io_uring_prep_read sqeptr fd buf (fromIntegral len) (fromIntegral off)+ FFI.io_uring_sqe_set_data sqeptr (fromIntegral ioopid)++prepareWrite :: URing -> Fd -> FileOffset -> Ptr Word8 -> ByteCount -> IOOpId -> IO ()+prepareWrite URing {uringptr} fd off buf len (IOOpId ioopid) = do+ sqeptr <- throwErrResIfNull "prepareWrite" fullErrorType+ "URing I/O queue full" $+ FFI.io_uring_get_sqe uringptr+ FFI.io_uring_prep_write sqeptr fd buf (fromIntegral len) (fromIntegral off)+ FFI.io_uring_sqe_set_data sqeptr (fromIntegral ioopid)++prepareNop :: URing -> IOOpId -> IO ()+prepareNop URing {uringptr} (IOOpId ioopid) = do+ sqeptr <- throwErrResIfNull "prepareNop" fullErrorType+ "URing I/O queue full" $+ FFI.io_uring_get_sqe uringptr+ FFI.io_uring_prep_nop sqeptr+ FFI.io_uring_sqe_set_data sqeptr (fromIntegral ioopid)++submitIO :: URing -> IO ()+submitIO URing {uringptr} =+ throwErrnoResIfNegRetry_ "submitIO" $+ FFI.io_uring_submit uringptr+++--+-- Types for completing I/O+--++data IOCompletion = IOCompletion !IOOpId !IOResult++newtype IOResult = IOResult_ Int+ deriving stock (Eq, Show)++{-# COMPLETE IOResult, IOError #-}++pattern IOResult :: ByteCount -> IOResult+pattern IOResult c <- (viewIOResult -> Just c)+ where+ IOResult count = IOResult_ ((fromIntegral :: CSize -> Int) count)++pattern IOError :: Errno -> IOResult+pattern IOError e <- (viewIOError -> Just e)+ where+ IOError (Errno e) = IOResult_ (fromIntegral (-e))++viewIOResult :: IOResult -> Maybe ByteCount+viewIOResult (IOResult_ c)+ | c >= 0 = Just ((fromIntegral :: Int -> CSize) c)+ | otherwise = Nothing++viewIOError :: IOResult -> Maybe Errno+viewIOError (IOResult_ e)+ | e < 0 = Just (Errno (fromIntegral e))+ | otherwise = Nothing+++--+-- Unboxed vector support for IOResult+--++newtype instance VUM.MVector s IOResult = MV_IOResult (VP.MVector s Int)+newtype instance VU.Vector IOResult = V_IOResult (VP.Vector Int)++deriving newtype instance VGM.MVector VUM.MVector IOResult+deriving newtype instance VG.Vector VU.Vector IOResult++instance VU.Unbox IOResult+++--+-- Completing I/O+--++-- | Must only be called from one thread at once.+awaitIO :: URing -> IO IOCompletion+awaitIO URing {uringptr, cqeptrfptr} = do+ -- We use unsafeForeignPtrToPtr and touchForeignPtr here rather than+ -- withForeignPtr because using withForeignPtr defeats GHCs CPR analysis+ -- which causes the 'IOCompletion' result to be allocated on the heap+ -- rather than returned in registers.++ let !cqeptrptr = unsafeForeignPtrToPtr cqeptrfptr+ -- Try non-blocking first (unsafe FFI call)+ peekres <- FFI.io_uring_peek_cqe uringptr cqeptrptr+ -- But if nothing is available, use a blocking call (safe FFI call)+-- when (peekres == 0) $ print ("awaitIO: non-blocking")+ when (peekres /= 0) $ do+ if Errno (-peekres) == eAGAIN+ then do --print ("awaitIO: blocking")+ throwErrnoResIfNegRetry_ "awaitIO (blocking)" $+ FFI.io_uring_wait_cqe uringptr cqeptrptr+ --print ("awaitIO: blocking complete")+ else throwIO $ errnoToIOError "awaitIO (non-blocking)"+ (Errno (-peekres)) Nothing Nothing+ cqeptr <- peek cqeptrptr+ FFI.URingCQE { FFI.cqe_data, FFI.cqe_res } <- peek cqeptr+ FFI.io_uring_cqe_seen uringptr cqeptr+ touchForeignPtr cqeptrfptr+ let opid = IOOpId (fromIntegral cqe_data)+ res = IOResult_ (fromIntegral cqe_res)+ return $! IOCompletion opid res+++--+-- Utils+--++throwErrnoResIfNegRetry_ :: String -> IO CInt -> IO ()+throwErrnoResIfNegRetry_ label action = go+ where+ go = do+ res <- action+ when (res < 0) $+ if Errno (-res) == eINTR+ then go+ else throwIO $+ errnoToIOError+ label+ (Errno (-res))+ Nothing Nothing++throwErrResIfNull :: String -> IOErrorType -> String -> IO (Ptr a) -> IO (Ptr a)+throwErrResIfNull location ioErrorType description action = do+ res <- action+ if res == nullPtr+ then throwIO $+ ioeSetErrorString+ (mkIOError+ ioErrorType+ location+ Nothing Nothing)+ description+ else return res+
+ src/System/IO/BlockIO/URingFFI.hsc view
@@ -0,0 +1,127 @@+{-# LANGUAGE CApiFFI #-}+{-# LANGUAGE InterruptibleFFI #-}+{-# LANGUAGE NamedFieldPuns #-}++{-# OPTIONS_GHC -fobject-code #-}+{-# OPTIONS_GHC -Wno-missing-export-lists #-}++module System.IO.BlockIO.URingFFI where++import Foreign+import Foreign.C+import Prelude hiding (head, tail)+import System.Posix.Types++#include <liburing.h>++data {-# CTYPE "liburing.h" "struct io_uring" #-} URing = URing++instance Storable URing where+ sizeOf _ = #{size struct io_uring}+ alignment _ = #{alignment struct io_uring}+ peek _ = return URing+ poke _ _ = return ()++foreign import capi unsafe "liburing.h io_uring_queue_init"+ io_uring_queue_init :: CUInt -> Ptr URing -> CUInt -> IO CInt++foreign import capi unsafe "liburing.h io_uring_queue_exit"+ io_uring_queue_exit :: Ptr URing -> IO ()++foreign import capi safe "liburing.h io_uring_queue_init_params"+ io_uring_queue_init_params :: CUInt -> Ptr URing -> Ptr URingParams -> IO CInt++iORING_SETUP_CQSIZE :: CUInt+iORING_SETUP_CQSIZE = #{const IORING_SETUP_CQSIZE}++data {-# CTYPE "liburing.h" "struct io_uring_params" #-}+ URingParams = URingParams {+ sq_entries :: !CUInt,+ cq_entries :: !CUInt,+ flags :: !CUInt,+ features :: !CUInt+ -- Note: this is a subset of all the fields. These are+ -- just the ones we need now or are likely too need.+ -- If you need more, just add them.+ }+ deriving stock (Show, Eq)++instance Storable URingParams where+ sizeOf _ = #{size struct io_uring_params}+ alignment _ = #{alignment struct io_uring_params}+ peek p = do sq_entries <- #{peek struct io_uring_params, sq_entries} p+ cq_entries <- #{peek struct io_uring_params, cq_entries} p+ flags <- #{peek struct io_uring_params, flags} p+ features <- #{peek struct io_uring_params, features} p+ return URingParams {..}+ poke p ps = do -- As we only cover a subset of the fields, we must clear+ -- the remaining fields we don't set to avoid them+ -- containing arbitrary values.+ fillBytes p 0 #{size struct io_uring_params}++ #{poke struct io_uring_params, sq_entries} p (sq_entries ps)+ #{poke struct io_uring_params, cq_entries} p (cq_entries ps)+ #{poke struct io_uring_params, flags} p (flags ps)+ #{poke struct io_uring_params, features} p (features ps)+++--+-- Submitting I/O+--++data {-# CTYPE "liburing.h" "struct io_uring_sqe" #-} URingSQE++foreign import capi unsafe "liburing.h io_uring_get_sqe"+ io_uring_get_sqe :: Ptr URing -> IO (Ptr URingSQE)++#ifdef LIBURING_HAVE_DATA64+foreign import capi unsafe "liburing.h io_uring_sqe_set_data64"+ io_uring_sqe_set_data :: Ptr URingSQE -> CULong -> IO ()+#else+io_uring_sqe_set_data :: Ptr URingSQE -> CULong -> IO ()+io_uring_sqe_set_data p user_data =+ do #{poke struct io_uring_sqe, user_data} p user_data+#endif++foreign import capi unsafe "liburing.h io_uring_prep_read"+ io_uring_prep_read :: Ptr URingSQE -> Fd -> Ptr Word8 -> CUInt -> CULong -> IO ()++foreign import capi unsafe "liburing.h io_uring_prep_write"+ io_uring_prep_write :: Ptr URingSQE -> Fd -> Ptr Word8 -> CUInt -> CULong -> IO ()++foreign import capi unsafe "liburing.h io_uring_prep_nop"+ io_uring_prep_nop :: Ptr URingSQE -> IO ()++foreign import capi unsafe "liburing.h io_uring_submit"+ io_uring_submit :: Ptr URing -> IO CInt+++--+-- Collecting I/O+--++data {-# CTYPE "liburing.h" "struct io_uring_cqe" #-}+ URingCQE = URingCQE {+ cqe_data :: !CULong,+ cqe_res :: !CInt+ }+ deriving stock Show++instance Storable URingCQE where+ sizeOf _ = #{size struct io_uring_cqe}+ alignment _ = #{alignment struct io_uring_cqe}+ peek p = do cqe_data <- #{peek struct io_uring_cqe, user_data} p+ cqe_res <- #{peek struct io_uring_cqe, res} p+ return URingCQE { cqe_data, cqe_res }+ poke _ _ = return ()+++foreign import capi safe "liburing.h io_uring_wait_cqe"+ io_uring_wait_cqe :: Ptr URing -> Ptr (Ptr URingCQE) -> IO CInt++foreign import capi unsafe "liburing.h io_uring_peek_cqe"+ io_uring_peek_cqe :: Ptr URing -> Ptr (Ptr URingCQE) -> IO CInt++foreign import capi unsafe "liburing.h io_uring_cqe_seen"+ io_uring_cqe_seen :: Ptr URing -> Ptr URingCQE -> IO ()+
+ test/test-internals.hs view
@@ -0,0 +1,61 @@+{-# OPTIONS_GHC -Wno-orphans #-}++{- HLINT ignore "Use camelCase" -}++module Main (main) where+++import Data.Proxy+import Data.Word (Word64)+import System.IO.BlockIO.URing+import qualified System.IO.BlockIO.URingFFI as FFI+import Test.QuickCheck.Classes+import Test.Tasty+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck++main :: IO ()+main = defaultMain tests++tests :: TestTree+tests = testGroup "test-internals" [+ testCase "example_simpleNoop 1" $ example_simpleNoop 1+ , testCase "example_simpleNoop maxBound" $ example_simpleNoop maxBound+ , testClassLaws "URingParams" $ storableLaws (Proxy @FFI.URingParams)+ ]++example_simpleNoop :: Word64 -> Assertion+example_simpleNoop n = do+ uring <- setupURing (URingParams 1 2)+ prepareNop uring (IOOpId n)+ submitIO uring+ completion <- awaitIO uring+ closeURing uring+ IOCompletion (IOOpId n) (IOResult 0) @=? completion++deriving stock instance Eq IOCompletion+deriving stock instance Show IOCompletion++{-------------------------------------------------------------------------------+ Storable+-------------------------------------------------------------------------------}++testClassLaws :: String -> Laws -> TestTree+testClassLaws typename laws = testClassLawsWith typename laws testProperty++testClassLawsWith ::+ String -> Laws+ -> (String -> Property -> TestTree)+ -> TestTree+testClassLawsWith typename Laws {lawsTypeclass, lawsProperties} k =+ testGroup ("class laws" ++ lawsTypeclass ++ " " ++ typename)+ [ k name prop+ | (name, prop) <- lawsProperties ]++instance Arbitrary FFI.URingParams where+ arbitrary = FFI.URingParams+ <$> arbitrary <*> arbitrary <*> arbitrary <*> arbitrary+ shrink (FFI.URingParams a b c d) = [+ FFI.URingParams a' b' c' d'+ | (a', b', c', d') <- shrink (a, b, c, d)+ ]
+ test/test.hs view
@@ -0,0 +1,140 @@+{-# OPTIONS_GHC -Wno-orphans #-}++{- HLINT ignore "Use camelCase" -}++module Main (main) where++import Control.Exception (Exception (displayException),+ IOException, SomeException, try)+import Control.Monad (void)+import Data.List (isPrefixOf)+import qualified Data.Primitive.ByteArray as P+import qualified Data.Vector as V+import GHC.IO.Exception (IOException (ioe_location))+import GHC.IO.FD (FD (..))+import GHC.IO.Handle.FD (handleToFd)+import System.IO+import System.IO.BlockIO+import Test.Tasty+import Test.Tasty.HUnit+import Test.Tasty.QuickCheck++main :: IO ()+main = defaultMain tests++tests :: TestTree+tests = testGroup "test"+ [ testCase "example_initClose" example_initClose+ , testCase "example_initReadClose 32" $ example_initReadClose 32+ , testCase "example_initReadClose 96" $ example_initReadClose 96+ , testCase "example_initReadClose 200" $ example_initReadClose 200+ , testCase "example_initEmptyClose" example_initEmptyClose+ , testCase "example_closeIsIdempotent" example_closeIsIdempotent+ , testProperty "prop_ValidIOCtxParams" prop_ValidIOCtxParams+ ]++example_initClose :: Assertion+example_initClose = do+ ctx <- initIOCtx defaultIOCtxParams+ closeIOCtx ctx++example_initReadClose :: Int -> Assertion+example_initReadClose size = do+ ctx <- initIOCtx defaultIOCtxParams+ withFile "blockio-uring.cabal" ReadMode $ \hdl -> do+ -- handleToFd is available since base-4.16.0.0+ FD { fdFD = fromIntegral -> fd } <- handleToFd hdl+ mba <- P.newPinnedByteArray 10 -- TODO: shouldn't use the same array for all ops :)+ void $ submitIO ctx $ V.replicate size $+ IOOpRead fd 0 mba 0 10+ closeIOCtx ctx++example_initEmptyClose :: Assertion+example_initEmptyClose = do+ ctx <- initIOCtx defaultIOCtxParams+ _ <- submitIO ctx V.empty+ closeIOCtx ctx++example_closeIsIdempotent :: Assertion+example_closeIsIdempotent = do+ ctx <- initIOCtx defaultIOCtxParams+ closeIOCtx ctx+ eith <- try (closeIOCtx ctx)+ case eith of+ Left e ->+ assertFailure ("Close on a closed context threw an error : " <> show (e :: SomeException))+ Right () ->+ pure ()++{-------------------------------------------------------------------------------+ Valid IOCtxParams+-------------------------------------------------------------------------------}++-- | We test @validateIOCtxParams@ through 'withIOCtx'. If any params slip+-- through the cracks, then the call to @setupURing@ inside 'withIOCtx' should+-- throw an unexpected exception, which causes the property to fail.+prop_ValidIOCtxParams :: IOCtxParams -> Property+prop_ValidIOCtxParams params@IOCtxParams{..} =+ checkCoverage $+ coverTable "Result" [("Success", 5)] $+ ioProperty $ do+ eith <- try @IOException $ withIOCtx params $ \_ctx -> pure ()+ pure $ case eith of+ Left e+ | "IOCtxParams are invalid" `isPrefixOf` ioe_location e+ &&+ not (+ inBoundsExcl ioctxBatchSizeLimit batchSizeLimitBoundsExcl &&+ inBoundsExcl ioctxConcurrencyLimit (concurrencyLimitBoundsExcl (ioctxBatchSizeLimit - 1))+ )+ -> tabulate "Result" [displayException e] True+ | otherwise+ -> counterexample ("Unknown exception: " ++ displayException e) False+ Right () -> tabulate "Result" ["Success"] True+ where+ inBoundsExcl x (lb, ub) = lb < x && x < ub++ batchSizeLimitBoundsExcl = (0, 2^(15 :: Int))+ concurrencyLimitBoundsExcl batchSizeLimit = (batchSizeLimit, 2^(16::Int))++instance Arbitrary IOCtxParams where+ arbitrary = IOCtxParams <$> genLimit <*> genLimit+ shrink (IOCtxParams a b) =+ [ IOCtxParams a' b' | (a', b') <- liftShrink2 shrinkLimit shrinkLimit (a, b) ]++genLimit :: Gen Int+genLimit = frequency [+ -- Generate powers of 2 to hit the upper bounds on the batch size limit+ -- and concurrency limit+ (10, genPowerOf2)+ -- Get some coverage of the whole range between the lower and upper+ -- bounds on the batch size limit and concurrency limit.+ , (10, chooseNonNegativeInt)+ -- Otherwise, generate integers like normal+ , (10, arbitrary)+ ]++shrinkLimit :: Int -> [Int]+shrinkLimit x = shrinkPowerOf2 x ++ shrinkNonNegativeInt x ++ shrink x++genPowerOf2 :: Gen Int+genPowerOf2 = do+ i <- chooseInt (minExponent, maxExponent)+ pure (2^i)++shrinkPowerOf2 :: Int -> [Int]+shrinkPowerOf2 x = [x' | i <- [ minExponent .. maxExponent ], let x' = 2^i, x' < x]++minExponent, maxExponent :: Int+minExponent = 0+maxExponent = 20++chooseNonNegativeInt :: Gen Int+chooseNonNegativeInt = chooseInt (minValue, maxValue)++shrinkNonNegativeInt :: Int -> [Int]+shrinkNonNegativeInt x = [ x' | NonNegative x' <- shrink (NonNegative x) ]++minValue, maxValue :: Int+minValue = 0+maxValue = 2^maxExponent