diff --git a/ChangeLog.md b/ChangeLog.md new file mode 100644 --- /dev/null +++ b/ChangeLog.md @@ -0,0 +1,3 @@ +# Changelog for iotests + +## Unreleased changes diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright Author name here (c) 2018 + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * 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. + + * Neither the name of Author name here nor the names of other + 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 +OWNER 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,70 @@ +# random-access-file README + +This package is aimed to provide some number of different implementations of +random file access methods with the same interface. + +It can be of use for implementing multithread read-write random access for +large files, for example for DB engines. + +The following implementations are provided: + +* Simple: trivial wrapper around standard System.IO calls. Created mostly for + demonstrative purposes. Specific thing about System.IO calls is that they + work with one Handle per file, and that Handle contains a pointer to current + position in the file. So it is not possible for several threads to read or + write to different positions in the file. This implementation is made + thread-safe by adding a global per-file lock: several threads can read and/or + write to different positions of the file, but all access will be serialized. +* Threaded: file access using Posix pread(3), pwrite(3) calls. This + implementation is thread-safe. It is using block-level locks; each block can + be accessed for write by single thread, or for read by many threads. Size of + blocks being locked is adjustable. +* MMaped: file access using mmap(2) call. This implementation is thread-safe. + It is using block-level locks also. +* Cached: File access using application-level page cache, which can be used + over Threaded or MMaped file access. Size of cache pages and capacity of the + cache are adjustable. Performance of this implementation depends very + seriously on your usage pattern and page size. + +## Benchmark results + +``` +Benchmark random-access-file-benchmark: RUNNING... +benchmarking main/simple +time 691.1 ms (-610.8 ms .. 1.689 s) + 0.639 R² (0.044 R² .. 1.000 R²) +mean 1.781 s (1.088 s .. 2.106 s) +std dev 543.2 ms (192.5 ms .. 732.0 ms) +variance introduced by outliers: 73% (severely inflated) + +benchmarking main/threaded +time 228.5 ms (196.1 ms .. 248.2 ms) + 0.993 R² (0.978 R² .. 1.000 R²) +mean 250.5 ms (239.5 ms .. 257.1 ms) +std dev 11.31 ms (3.916 ms .. 16.04 ms) +variance introduced by outliers: 16% (moderately inflated) + +benchmarking main/mmaped +time 166.0 ms (147.6 ms .. 182.9 ms) + 0.989 R² (0.963 R² .. 1.000 R²) +mean 162.5 ms (157.6 ms .. 168.0 ms) +std dev 7.960 ms (4.764 ms .. 11.38 ms) +variance introduced by outliers: 12% (moderately inflated) + +benchmarking main/cached/threaded +time 1.272 s (575.8 ms .. 1.865 s) + 0.951 R² (0.925 R² .. 1.000 R²) +mean 1.517 s (1.365 s .. 1.626 s) +std dev 150.9 ms (64.02 ms .. 207.5 ms) +variance introduced by outliers: 23% (moderately inflated) + +benchmarking main/cached/mmaped +time 199.2 ms (78.98 ms .. 332.2 ms) + 0.863 R² (0.708 R² .. 1.000 R²) +mean 237.7 ms (190.4 ms .. 285.1 ms) +std dev 62.14 ms (31.83 ms .. 85.03 ms) +variance introduced by outliers: 58% (severely inflated) + +Benchmark random-access-file-benchmark: FINISH +``` + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/benchmark/Main.hs b/benchmark/Main.hs new file mode 100644 --- /dev/null +++ b/benchmark/Main.hs @@ -0,0 +1,72 @@ +{-# LANGUAGE OverloadedStrings #-} + +import Control.Monad +import Control.Concurrent +import qualified Data.ByteString as B +import qualified Data.Vector as V +import Data.Vector ((!)) +import System.Random +import System.Random.MWC +import System.IO +import Criterion.Main + +import System.IO.RandomAccessFile + +mkRandomOffsets :: Int -> (Offset, Offset) -> IO (V.Vector Offset) +mkRandomOffsets n (from, to) = + withSystemRandom . asGenIO $ \gen -> V.fromList `fmap` replicateM n (uniformR (from, to) gen) + +createFile :: FilePath -> IO () +createFile path = do + h <- openFile path ReadWriteMode + B.hPut h $ B.replicate (100*1024*1024) 0 + hClose h + +concurrently :: Int -> (Int -> IO ()) -> IO () +concurrently n action = do + vars <- replicateM n newEmptyMVar + forM_ (zip [0..] vars) $ \(i, var) -> forkIO $ do + action i + putMVar var () + forM_ vars takeMVar + +execute :: FileAccess a => AccessParams a -> FilePath -> Bool -> IO () +execute params path doClose = do + + h <- initFile params path + -- writeZeros h (1024*1024) + offsets <- mkRandomOffsets (40*(500+1000)) (100, 100*900*1024) + + concurrently 40 $ \threadId -> do + forM_ [0..499] $ \i -> do + let offset = offsets ! (threadId*(500+1000) + i) + writeBytes h offset "abdefgh0123456789ABCDEFGIJK" + return () + + forM_ [0.. 999] $ \i -> do + let offset = offsets ! (threadId*(500+1000) + 500 + i) + readBytes h offset 512 + return () + + when doClose $ + closeFile h + +benchList name list benchmarks = bgroup name + [bgroup (name ++ ":" ++ show param) [fn param | fn <- benchmarks] + | param <- list] + +main :: IO () +main = defaultMain [ + env (createFile "test.data") $ \_ -> bgroup "main" [ + bench "simple" $ whnfIO $ execute SimpleParams "test.data" True, + benchList "raw" [1024, 4*1024, 16*1024] [ + \pageSize -> bench "threaded" $ whnfIO $ execute (ThreadedParams pageSize) "test.data" True, + \pageSize -> bench "mmaped" $ whnfIO $ execute (MMapedParams pageSize False) "test.data" True + ], + benchList "cached" [1024, 4*1024, 16*1024] [ + \pageSize -> bench "threaded" $ whnfIO $ execute ((dfltCached (ThreadedParams pageSize)) {cachePageSize = pageSize}) "test.data" True, + \pageSize -> bench "mmaped" $ whnfIO $ execute ((dfltCached (MMapedParams pageSize False)) {cachePageSize = pageSize}) "test.data" True + ] + ] + ] + diff --git a/random-access-file.cabal b/random-access-file.cabal new file mode 100644 --- /dev/null +++ b/random-access-file.cabal @@ -0,0 +1,80 @@ +cabal-version: 1.12 + +-- This file has been generated from package.yaml by hpack version 0.31.1. +-- +-- see: https://github.com/sol/hpack +-- +-- hash: d3a26fdb60817d0a049223a739346a7af2298e7b58ed74c1d10b57f96e40d2c4 + +name: random-access-file +version: 0.1.0.0 +synopsis: Random file access methods, supporting application-level page cache. +description: Please see the README on GitHub at +category: System +homepage: https://github.com/portnov/random-access-file#readme +bug-reports: https://github.com/portnov/random-access-file/issues +author: Ilya V. Portnov +maintainer: portnov84@rambler.ru +copyright: 2018 Ilya Portnov +license: BSD3 +license-file: LICENSE +build-type: Simple +extra-source-files: + README.md + ChangeLog.md + +source-repository head + type: git + location: https://github.com/portnov/random-access-file + +library + exposed-modules: + System.IO.RandomAccessFile + System.IO.RandomAccessFile.Cached + System.IO.RandomAccessFile.Common + System.IO.RandomAccessFile.MMap + System.IO.RandomAccessFile.Simple + System.IO.RandomAccessFile.Threaded + other-modules: + Paths_random_access_file + hs-source-dirs: + src + ghc-options: -fwarn-unused-imports + build-depends: + base >=4.7 && <5 + , bytestring + , concurrent-extra + , containers + , directory + , lrucaching + , stm + , unix + , unix-bytestring + , unix-memory + default-language: Haskell2010 + +benchmark random-access-file-benchmark + type: exitcode-stdio-1.0 + main-is: Main.hs + other-modules: + Paths_random_access_file + hs-source-dirs: + benchmark + ghc-options: -threaded -rtsopts -with-rtsopts=-N -fwarn-unused-imports + build-depends: + base >=4.7 && <5 + , bytestring + , concurrent-extra + , containers + , criterion + , directory + , lrucaching + , mwc-random + , random + , random-access-file + , stm + , unix + , unix-bytestring + , unix-memory + , vector + default-language: Haskell2010 diff --git a/src/System/IO/RandomAccessFile.hs b/src/System/IO/RandomAccessFile.hs new file mode 100644 --- /dev/null +++ b/src/System/IO/RandomAccessFile.hs @@ -0,0 +1,15 @@ + +module System.IO.RandomAccessFile + (module System.IO.RandomAccessFile.Common, + module System.IO.RandomAccessFile.Simple, + module System.IO.RandomAccessFile.Threaded, + module System.IO.RandomAccessFile.MMap, + module System.IO.RandomAccessFile.Cached + ) where + +import System.IO.RandomAccessFile.Common +import System.IO.RandomAccessFile.Simple +import System.IO.RandomAccessFile.Threaded +import System.IO.RandomAccessFile.MMap +import System.IO.RandomAccessFile.Cached + diff --git a/src/System/IO/RandomAccessFile/Cached.hs b/src/System/IO/RandomAccessFile/Cached.hs new file mode 100644 --- /dev/null +++ b/src/System/IO/RandomAccessFile/Cached.hs @@ -0,0 +1,232 @@ +{-# LANGUAGE PackageImports #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeFamilies #-} + +module System.IO.RandomAccessFile.Cached + (Cached (..), + AccessParams (..), + dfltCached + ) where + +import Control.Monad +import Control.Concurrent +import Control.Concurrent.STM +import qualified Control.Concurrent.ReadWriteLock as RWL +import qualified Data.Map as M +import qualified Data.ByteString as B +import qualified Data.ByteString.Lazy as L +import qualified Data.LruCache as LRU +import System.Posix.IO +import System.Directory +import Text.Printf + +import System.IO.RandomAccessFile.Common + +data Page = Page {pData :: B.ByteString, pLock :: RWL.RWLock} + deriving (Eq) + +instance Show Page where + show p = "[Page]" + +data CacheData = CacheData { + cdDirty :: M.Map Offset Page + , cdClean :: LRU.LruCache Offset Page + } + +type Cache = TVar CacheData + +data Cached a = Cached { + cBackend :: a + , cCachePageSize :: Size + , cCapacity :: Int + , cCanClose :: TVar Bool + , cCloseLock :: MVar () + , cCache :: Cache + } + +lookupC :: Offset -> CacheData -> Maybe (Page, CacheData) +lookupC key c = + case M.lookup key (cdDirty c) of + Just page -> Just (page, c) + Nothing -> case LRU.lookup key (cdClean c) of + Nothing -> Nothing + Just (page, lru') -> Just (page, c {cdClean = lru'}) + +putDirty :: Offset -> Page -> CacheData -> CacheData +putDirty key page c = + c {cdDirty = M.insert key page (cdDirty c)} + +putClean :: Offset -> Page -> CacheData -> CacheData +putClean key page c = + c {cdClean = LRU.insert key page (cdClean c)} + +markAllClean :: CacheData -> CacheData +markAllClean c = c {cdClean = update (cdClean c), cdDirty = M.empty} + where + update lru = foldr (uncurry LRU.insert) lru (M.assocs $ cdDirty c) + +mkPage :: Size -> B.ByteString +mkPage sz = B.replicate (fromIntegral sz) 0 + +instance FileAccess a => FileAccess (Cached a) where + data AccessParams (Cached a) = + CachedBackend { + backendParams :: AccessParams a + , cachePageSize :: Size + , cacheCapacity :: Int + , cacheFlushPeriod :: Int + } + + initFile (CachedBackend params cachePageSize capacity flushPeriod) path = do + ex <- doesFileExist path + a <- initFile params path + when (not ex) $ do + writeBytes a 0 $ mkPage cachePageSize + + var <- atomically $ newTVar $ CacheData M.empty (LRU.empty capacity) + let fileMode = Just 0o644 + let flags = defaultFileFlags + canClose <- newTVarIO False + closeLock <- newEmptyMVar + forkIO $ dumpQueue a canClose closeLock var + return $ Cached a cachePageSize capacity canClose closeLock var + where + dumpQueue a canCloseVar closeLock var = do + threadDelay flushPeriod + pages <- atomically $ do + cache <- readTVar var + let pages = M.assocs $ cdDirty cache + cache' = markAllClean cache + writeTVar var cache' + return pages + if null pages + then do + -- check if there was a closeFile call + canClose <- atomically $ readTVar canCloseVar + if canClose + then do + closeFile a + putMVar closeLock () -- signal to closeFile call + else dumpQueue a canCloseVar closeLock var + else do + forM_ pages $ \(offset, page) -> do + writeBytes a offset (pData page) + -- syncFile a + dumpQueue a canCloseVar closeLock var + + currentFileSize handle = + currentFileSize (cBackend handle) + + readBytes handle offset size = do + let cachePageSize = cCachePageSize handle + dataOffset0 = offset `mod` cachePageSize + pageOffset0 = offset - dataOffset0 + dataOffset1 = (offset + size) `mod` cachePageSize + pageOffset1 = (offset + size) - dataOffset1 + pageOffsets = [pageOffset0, pageOffset0 + cachePageSize .. pageOffset1] + inputs = flip map pageOffsets $ \page -> + if page == pageOffset0 + then (page, dataOffset0, min size (cachePageSize - dataOffset0)) + else if page == pageOffset1 + then (page, 0, dataOffset1) + else (page, 0, cachePageSize) + -- printf "PO: %s\n" (show pageOffsets) + -- printf "I: %s\n" (show inputs) + fragments <- forM inputs $ \(pageOffset, dataOffset, sz) -> + readDataAligned handle pageOffset dataOffset sz + return $ B.concat fragments + + writeBytes handle offset bstr = do + let size = fromIntegral $ B.length bstr + cachePageSize = cCachePageSize handle + dataOffset0 = offset `mod` cachePageSize + pageOffset0 = offset - dataOffset0 + dataOffset1 = (offset + size) `mod` cachePageSize + pageOffset1 = (offset + size) - dataOffset1 + pageOffsets = [pageOffset0, pageOffset0 + cachePageSize .. pageOffset1] + inputs = flip map pageOffsets $ \page -> + if page == pageOffset0 + then (page, dataOffset0, min size (cachePageSize - dataOffset0)) + else if page == pageOffset1 + then (page, 0, dataOffset1) + else (page, 0, cachePageSize) + fragments = flip map inputs $ \(pageOffset, dataOffset, sz) -> + let strOffset = pageOffset + dataOffset - offset + fragment = B.take (fromIntegral sz) $ B.drop (fromIntegral strOffset) bstr + in (pageOffset, dataOffset, fragment) + -- printf "PO: %s\n" (show pageOffsets) + -- printf "I: %s\n" (show inputs) + -- printf "F: %s\n" (show fragments) + forM_ fragments $ \(pageOffset, dataOffset, fragment) -> + writeDataAligned handle pageOffset dataOffset fragment + + syncFile handle = do + syncFile (cBackend handle) + + closeFile handle = do + -- signal to dumpQueue thread that we are ready + -- for file to be closed + atomically $ writeTVar (cCanClose handle) True + -- wait for signal from dumpQueue thread + takeMVar (cCloseLock handle) + +readDataAligned :: FileAccess a => Cached a -> Offset -> Offset -> Size -> IO B.ByteString +readDataAligned handle pageOffset dataOffset size = do + let a = cBackend handle + var = cCache handle + cachePageSize = cCachePageSize handle + mbCached <- atomically $ do + cache <- readTVar var + return $ lookupC pageOffset cache + case mbCached of + Nothing -> do + page <- readBytes a pageOffset cachePageSize + let result = B.take (fromIntegral size) $ B.drop (fromIntegral dataOffset) page + lock <- RWL.new + atomically $ modifyTVar var $ \cache -> + putClean pageOffset (Page page lock) cache + return result + Just (page, cache') -> do + withLock (pLock page) ReadAccess $ do + atomically $ writeTVar var cache' + let result = B.take (fromIntegral size) $ B.drop (fromIntegral dataOffset) $ pData page + return result + +writeDataAligned :: FileAccess a => Cached a -> Offset -> Offset -> B.ByteString -> IO () +writeDataAligned handle pageOffset dataOffset bstr = do + -- printf "WA: page %d, data %d, len %d\n" pageOffset dataOffset (B.length bstr) + fsize <- currentFileSize handle + let a = cBackend handle + var = cCache handle + cachePageSize = cCachePageSize handle + size = fromIntegral $ B.length bstr + mbCached <- atomically $ do + cache <- readTVar var + return $ lookupC pageOffset cache + case mbCached of + Nothing -> do + page <- if (pageOffset + dataOffset + size) >= fsize + then return $ mkPage cachePageSize + else readBytes a pageOffset cachePageSize + let page' = B.take (fromIntegral dataOffset) page `B.append` + bstr `B.append` + B.drop (fromIntegral dataOffset + B.length bstr) page + when (fromIntegral (B.length page) /= B.length page') $ + fail $ printf "W/N: %d /= %d! data: %d, page: %d, len: %d" (B.length page) (B.length page') pageOffset dataOffset (B.length bstr) + lock <- RWL.new + atomically $ modifyTVar var $ \cache -> + putDirty pageOffset (Page page' lock) cache + + Just (page, cache') -> do + withLock (pLock page) WriteAccess $ do + let pageData = pData page + let pageData' = B.take (fromIntegral dataOffset) pageData `B.append` bstr `B.append` B.drop (fromIntegral dataOffset + B.length bstr) pageData + let page' = page {pData = pageData'} + when (B.length pageData /= B.length pageData') $ + fail $ printf "W/J: %d /= %d!" (B.length pageData) (B.length pageData') + atomically $ modifyTVar var $ \cache -> + putDirty pageOffset page' cache + +dfltCached :: FileAccess a => AccessParams a -> AccessParams (Cached a) +dfltCached params = CachedBackend params 4096 1024 (10*1000) + diff --git a/src/System/IO/RandomAccessFile/Common.hs b/src/System/IO/RandomAccessFile/Common.hs new file mode 100644 --- /dev/null +++ b/src/System/IO/RandomAccessFile/Common.hs @@ -0,0 +1,97 @@ +{-# LANGUAGE TypeFamilies #-} + +module System.IO.RandomAccessFile.Common where + +import Control.Monad +import Control.Concurrent +import Control.Concurrent.STM +import qualified Control.Concurrent.ReadWriteLock as RWL +import Control.Exception +import qualified Data.Map as M +import qualified Data.ByteString as B +import Data.List +import Data.Word + +type Offset = Word64 +type Size = Word64 + +class FileAccess a where + data AccessParams a + + initFile :: AccessParams a -> FilePath -> IO a + readBytes :: a -> Offset -> Size -> IO B.ByteString + writeBytes :: a -> Offset -> B.ByteString -> IO () + + currentFileSize :: a -> IO Size + + syncFile :: a -> IO () + syncFile _ = return () + + closeFile :: a -> IO () + +writeZeros :: FileAccess a => a -> Size -> IO () +writeZeros h size = + writeBytes h 0 $ B.replicate (fromIntegral size) 0 + +data AccessType = ReadAccess | WriteAccess + +type FileLocks = M.Map Offset RWL.RWLock + +withQSem :: QSem -> IO a -> IO a +withQSem sem = + bracket_ (waitQSem sem) (signalQSem sem) + +withLock :: RWL.RWLock -> AccessType -> IO a -> IO a +withLock lock ReadAccess action = + bracket_ + (RWL.acquireRead lock) + (RWL.releaseRead lock) + action +withLock lock WriteAccess action = + bracket_ + (RWL.acquireWrite lock) + (RWL.releaseWrite lock) + action + +withLocks :: [RWL.RWLock] -> AccessType -> IO a -> IO a +withLocks locks ReadAccess action = + bracket_ + (forM_ locks RWL.acquireRead) + (forM_ locks RWL.releaseRead) + action +withLocks locks WriteAccess action = + bracket_ + (forM_ locks RWL.acquireWrite) + (forM_ locks RWL.releaseWrite) + action + +withLock_ :: Bool -> RWL.RWLock -> AccessType -> IO a -> IO a +withLock_ False _ _ action = action +withLock_ True lock access action = withLock lock access action + +underBlockLock :: TVar FileLocks -> AccessType -> Offset -> IO a -> IO a +underBlockLock locksVar access n action = do + newLock <- RWL.new + lock <- atomically $ do + locks <- readTVar locksVar + case M.lookup n locks of + Just lock -> return lock + Nothing -> do + writeTVar locksVar $ M.insert n newLock locks + return newLock + withLock lock access action + +underBlockLocks :: TVar FileLocks -> AccessType -> [Offset] -> IO a -> IO a +underBlockLocks locksVar access ns action = do + newLocks <- replicateM (length ns) RWL.new + locks <- atomically $ do + locks <- readTVar locksVar + forM (zip [0..] $ sort ns) $ \(idx,n) -> + case M.lookup n locks of + Just lock -> return lock + Nothing -> do + let newLock = newLocks !! idx + writeTVar locksVar $ M.insert n newLock locks + return newLock + withLocks locks access action + diff --git a/src/System/IO/RandomAccessFile/MMap.hs b/src/System/IO/RandomAccessFile/MMap.hs new file mode 100644 --- /dev/null +++ b/src/System/IO/RandomAccessFile/MMap.hs @@ -0,0 +1,181 @@ +{-# LANGUAGE PackageImports #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeFamilies #-} + +module System.IO.RandomAccessFile.MMap + (MMaped, + AccessParams (..), + extendFile + ) where + +import Control.Monad +import Control.Concurrent.STM +import qualified Control.Concurrent.ReadWriteLock as RWL +import qualified Data.Map as M +import qualified Data.ByteString as B +import Data.ByteString.Unsafe +import Data.Int +import System.IO +import System.Posix.Types +import System.Posix.IO +import Foreign.Ptr +import Foreign.C.Types +import Foreign.Marshal +import System.Posix.Memory +import Text.Printf + +import System.IO.RandomAccessFile.Common + +data MMaped = MMaped { + mmFile :: TVar Fd + , mmPath :: FilePath + , mmData :: TVar (Ptr CChar) + , mmExtendable :: Bool + , mmFileSize :: TVar CSize + , mmLockPageSize :: Size + , mmLocks :: TVar FileLocks + , mmResizeLock :: RWL.RWLock + } + +mmap :: CSize -> Fd -> IO (Ptr CChar) +mmap size fd = + memoryMap Nothing size [MemoryProtectionRead, MemoryProtectionWrite] MemoryMapShared (Just fd) 0 + +-- | Resize file to be at least of specified size. +-- Does nothing if size of file is already greater or equal +-- to specified. +-- While file is resized, all reading and writing to it are locked. +extendFile :: MMaped -> Size -> IO () +extendFile handle newSize = when (mmExtendable handle) $ do + let sizeVar = mmFileSize handle + ptrVar = mmData handle + page = mmLockPageSize handle + ptr <- atomically $ readTVar ptrVar + oldSize <- atomically $ readTVar sizeVar + let delta :: Int64 + delta = fromIntegral newSize - fromIntegral oldSize + + when (delta > 0) $ + -- We are acquiring "general" lock here to + -- make sure that there will be no newly created + -- page-level locks + withLock (mmResizeLock handle) WriteAccess $ do + -- Acquire all existing page-level locks in Write mode + -- so that noone may read or write the file: we are going to + -- unmap it + locks <- atomically $ readTVar (mmLocks handle) + withLocks (M.elems locks) WriteAccess $ do + + -- Unmap existing mapping + memorySync ptr oldSize [MemorySyncSync, MemorySyncInvalidate] + memoryUnmap ptr oldSize + + -- Open file again + h <- openFile (mmPath handle) ReadWriteMode + -- Write zeros to the end + size <- hFileSize h + hSeek h AbsoluteSeek size + B.hPut h $ B.replicate (fromIntegral page) 0 + hFlush h + fd <- handleToFd h + -- MMap file again + let newSize' = oldSize + fromIntegral page + ptr' <- mmap (fromIntegral newSize') fd + + atomically $ do + writeTVar (mmFile handle) fd + writeTVar ptrVar ptr' + writeTVar sizeVar (fromIntegral newSize') + +instance FileAccess MMaped where + data AccessParams MMaped = MMapedParams Size Bool + + initFile (MMapedParams lockPageSize extendable) path = do + locks <- atomically $ newTVar M.empty + let fileMode = Just 0o644 + let flags = defaultFileFlags + handle <- openFile path ReadWriteMode + size <- do + sz <- hFileSize handle + if sz == 0 + then do + B.hPut handle $ B.replicate (fromIntegral lockPageSize) 0 + hFlush handle + return lockPageSize + else return $ fromIntegral sz +-- printf "Init size: %d\n" size + sizeVar <- newTVarIO (fromIntegral size) + fd <- handleToFd handle +-- fd <- openFd path ReadWrite fileMode flags +-- handle <- fdToHandle fd + fdVar <- newTVarIO fd + ptr <- mmap (fromIntegral size) fd + ptrVar <- newTVarIO ptr + resizeLock <- RWL.new + return $ MMaped fdVar path ptrVar extendable sizeVar lockPageSize locks resizeLock + + readBytes handle offset size = do + withLock_ (mmExtendable handle) (mmResizeLock handle) ReadAccess $ do -- just check that file is not currently being resized + let ptrVar = mmData handle + lockPageSize = mmLockPageSize handle + locks = mmLocks handle + (ptr, fsize) <- atomically $ do + p <- readTVar ptrVar + s <- readTVar (mmFileSize handle) + return (p, s) + when (offset + size > fromIntegral fsize) $ + fail $ printf "readBytes: read after EOF: offset %d, size %s, file size %s" offset (show size) (show fsize) + let dataOffset0 = offset `mod` lockPageSize + pageOffset0 = offset - dataOffset0 + dataOffset1 = (offset + size) `mod` lockPageSize + pageOffset1 = (offset + size) - dataOffset1 + pageOffsets = [pageOffset0, pageOffset0 + lockPageSize .. pageOffset1] + + underBlockLocks locks ReadAccess pageOffsets $ do + let bstrPtr = plusPtr ptr (fromIntegral offset) + -- unsafePackCStringLen (bstrPtr, fromIntegral size) + B.packCStringLen (bstrPtr, fromIntegral size) + + writeBytes handle offset bstr = do + let ptrVar = mmData handle + sizeVar = mmFileSize handle + lockPageSize = mmLockPageSize handle + locks = mmLocks handle + fsize <- atomically $ readTVar sizeVar + let size = fromIntegral $ B.length bstr + dataOffset0 = offset `mod` lockPageSize + pageOffset0 = offset - dataOffset0 + dataOffset1 = (offset + size) `mod` lockPageSize + pageOffset1 = (offset + size) - dataOffset1 + pageOffsets = [pageOffset0, pageOffset0 + lockPageSize .. pageOffset1] + -- printf "write: offset %d, size %d, fsize %s\n" offset size (show fsize) + if (offset + size) > fromIntegral fsize + then do + extendFile handle (offset+size) + writeBytes handle offset bstr + else + -- just check that file is not currently being resized + withLock_ (mmExtendable handle) (mmResizeLock handle) ReadAccess $ + underBlockLocks locks WriteAccess pageOffsets $ do + ptr <- atomically $ readTVar ptrVar + unsafeUseAsCStringLen bstr $ \(bstrPtr,len) -> + copyBytes (plusPtr ptr (fromIntegral offset)) bstrPtr len + + currentFileSize handle = do + sz <- atomically $ readTVar $ mmFileSize handle + return $ fromIntegral sz + + syncFile handle = do + (ptr, size) <- atomically $ do + p <- readTVar (mmData handle) + s <- readTVar (mmFileSize handle) + return (p,s) + memorySync ptr size [MemorySyncSync, MemorySyncInvalidate] + + closeFile handle = do + (ptr, size) <- atomically $ do + p <- readTVar (mmData handle) + s <- readTVar (mmFileSize handle) + return (p,s) + memoryUnmap ptr size + diff --git a/src/System/IO/RandomAccessFile/Simple.hs b/src/System/IO/RandomAccessFile/Simple.hs new file mode 100644 --- /dev/null +++ b/src/System/IO/RandomAccessFile/Simple.hs @@ -0,0 +1,38 @@ +{-# LANGUAGE TypeFamilies #-} +module System.IO.RandomAccessFile.Simple where + +import Control.Concurrent +import Control.Exception +import qualified Data.ByteString as B +import System.IO + +import System.IO.RandomAccessFile.Common + +data Simple = Simple Handle QSem + +instance FileAccess Simple where + data AccessParams Simple = SimpleParams + + initFile _ path = do + handle <- openFile path ReadWriteMode + sem <- newQSem 1 + return $ Simple handle sem + + readBytes (Simple handle sem) offset size = do + bracket_ (waitQSem sem) (signalQSem sem) $ do + hSeek handle AbsoluteSeek (fromIntegral offset) + bstr <- B.hGet handle $ fromIntegral size + return bstr + + writeBytes (Simple handle sem) offset bstr = do + bracket_ (waitQSem sem) (signalQSem sem) $ do + hSeek handle AbsoluteSeek (fromIntegral offset) + B.hPut handle bstr + + currentFileSize (Simple handle sem) = + fromIntegral `fmap` hFileSize handle + + closeFile (Simple handle sem) = + bracket_ (waitQSem sem) (signalQSem sem) $ do + hClose handle + diff --git a/src/System/IO/RandomAccessFile/Threaded.hs b/src/System/IO/RandomAccessFile/Threaded.hs new file mode 100644 --- /dev/null +++ b/src/System/IO/RandomAccessFile/Threaded.hs @@ -0,0 +1,88 @@ +{-# LANGUAGE PackageImports #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeFamilies #-} + +module System.IO.RandomAccessFile.Threaded where + +import Control.Monad +import Control.Concurrent.STM +import Control.Exception +import qualified Data.Map as M +import qualified Data.ByteString as B +import Data.Int +import System.Posix.Types +import System.Posix.IO +import System.IO +import "unix-bytestring" System.Posix.IO.ByteString +import Text.Printf + +import System.IO.RandomAccessFile.Common + +data Threaded = Threaded { + tFile :: Fd + , tFileSize :: TVar Size + , tLockSize :: Size + , tLocks :: TVar FileLocks + } + +instance FileAccess Threaded where + data AccessParams Threaded = ThreadedParams Size + + initFile (ThreadedParams lockPageSize) path = do + locks <- atomically $ newTVar M.empty + let fileMode = Just 0o644 + let flags = defaultFileFlags + handle <- openFile path ReadWriteMode + size <- hFileSize handle + -- printf "Size: %d\n" size + sizeVar <- newTVarIO (fromIntegral size) + fd <- handleToFd handle + -- fd <- openFd path ReadWrite fileMode flags + return $ Threaded fd sizeVar lockPageSize locks + + readBytes (Threaded fd fileSizeVar lockPageSize locks) offset size = do + let dataOffset0 = offset `mod` lockPageSize + pageOffset0 = offset - dataOffset0 + dataOffset1 = (offset + size) `mod` lockPageSize + pageOffset1 = (offset + size) - dataOffset1 + pageOffsets = [pageOffset0, pageOffset0 + lockPageSize .. pageOffset1] + underBlockLocks locks ReadAccess pageOffsets $ + fdPread fd (fromIntegral size) (fromIntegral offset) + `catch` (\(e :: SomeException) -> do + printf "pread: offset %d, len %d: %s\n" + offset size (show e) + throw e) + + writeBytes (Threaded fd fileSizeVar lockPageSize locks) offset bstr = do + let size = fromIntegral $ B.length bstr + dataOffset0 = offset `mod` lockPageSize + pageOffset0 = offset - dataOffset0 + dataOffset1 = (offset + size) `mod` lockPageSize + pageOffset1 = (offset + size) - dataOffset1 + pageOffsets = [pageOffset0, pageOffset0 + lockPageSize .. pageOffset1] + + pwrite bytes off = + (fdPwrite fd bytes (fromIntegral off) >> return ()) + `catch` (\(e :: SomeException) -> + printf "pwrite: offset %d, len %d: %s\n" + offset (B.length bstr) (show e)) + fsize <- atomically $ readTVar fileSizeVar + underBlockLocks locks WriteAccess pageOffsets $ do + let delta :: Int64 + delta = max 0 $ fromIntegral (offset + size) - fromIntegral fsize + when (delta > 0) $ + pwrite (B.replicate (fromIntegral delta) 0) fsize + pwrite bstr offset + when (delta > 0) $ do + -- printf "New size: %d+%d\n" fsize delta + atomically $ writeTVar fileSizeVar $ fromIntegral (fromIntegral fsize + delta) + + currentFileSize h = do + let var = tFileSize h + atomically $ readTVar var + + closeFile (Threaded fd _ _ _) = closeFd fd + +dfltThreaded :: AccessParams Threaded +dfltThreaded = ThreadedParams 4096 +