random-access-file (empty) → 0.1.0.0
raw patch · 12 files changed
+908/−0 lines, 12 filesdep +basedep +bytestringdep +concurrent-extrasetup-changed
Dependencies added: base, bytestring, concurrent-extra, containers, criterion, directory, lrucaching, mwc-random, random, random-access-file, stm, unix, unix-bytestring, unix-memory, vector
Files
- ChangeLog.md +3/−0
- LICENSE +30/−0
- README.md +70/−0
- Setup.hs +2/−0
- benchmark/Main.hs +72/−0
- random-access-file.cabal +80/−0
- src/System/IO/RandomAccessFile.hs +15/−0
- src/System/IO/RandomAccessFile/Cached.hs +232/−0
- src/System/IO/RandomAccessFile/Common.hs +97/−0
- src/System/IO/RandomAccessFile/MMap.hs +181/−0
- src/System/IO/RandomAccessFile/Simple.hs +38/−0
- src/System/IO/RandomAccessFile/Threaded.hs +88/−0
+ ChangeLog.md view
@@ -0,0 +1,3 @@+# Changelog for iotests++## Unreleased changes
+ LICENSE view
@@ -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.
+ README.md view
@@ -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+```+
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ benchmark/Main.hs view
@@ -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+ ]+ ]+ ]+
+ random-access-file.cabal view
@@ -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 <https://github.com/portnov/random-access-file#readme>+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
+ src/System/IO/RandomAccessFile.hs view
@@ -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+
+ src/System/IO/RandomAccessFile/Cached.hs view
@@ -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)+
+ src/System/IO/RandomAccessFile/Common.hs view
@@ -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+
+ src/System/IO/RandomAccessFile/MMap.hs view
@@ -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+
+ src/System/IO/RandomAccessFile/Simple.hs view
@@ -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+
+ src/System/IO/RandomAccessFile/Threaded.hs view
@@ -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+