diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,5 @@ +# Revision history for blockio-uring + +## 0.1.0.0 -- 2025-06-09 + +* First release diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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. diff --git a/benchmark/Bench.hs b/benchmark/Bench.hs new file mode 100644 --- /dev/null +++ b/benchmark/Bench.hs @@ -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 + diff --git a/blockio-uring.cabal b/blockio-uring.cabal new file mode 100644 --- /dev/null +++ b/blockio-uring.cabal @@ -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 diff --git a/src/System/IO/BlockIO.hs b/src/System/IO/BlockIO.hs new file mode 100644 --- /dev/null +++ b/src/System/IO/BlockIO.hs @@ -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") diff --git a/src/System/IO/BlockIO/URing.hs b/src/System/IO/BlockIO/URing.hs new file mode 100644 --- /dev/null +++ b/src/System/IO/BlockIO/URing.hs @@ -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 + diff --git a/src/System/IO/BlockIO/URingFFI.hsc b/src/System/IO/BlockIO/URingFFI.hsc new file mode 100644 --- /dev/null +++ b/src/System/IO/BlockIO/URingFFI.hsc @@ -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 + +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 () + diff --git a/test/test-internals.hs b/test/test-internals.hs new file mode 100644 --- /dev/null +++ b/test/test-internals.hs @@ -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) + ] diff --git a/test/test.hs b/test/test.hs new file mode 100644 --- /dev/null +++ b/test/test.hs @@ -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