liszt 0.1 → 0.2
raw patch · 9 files changed
+481/−232 lines, 9 filesdep +cerealdep +cpudep +gauge
Dependencies added: cereal, cpu, gauge, vector, vector-th-unbox
Files
- README.md +53/−0
- app/server.hs +0/−1
- benchmarks/bench.hs +9/−0
- liszt.cabal +62/−14
- src/Database/Liszt.hs +34/−4
- src/Database/Liszt/Internal.hs +174/−137
- src/Database/Liszt/Internal/Decoder.hs +70/−0
- src/Database/Liszt/Network.hs +49/−36
- src/Database/Liszt/Tracker.hs +30/−40
README.md view
@@ -5,6 +5,7 @@ ## Insertion For the sake of reliability, insertion is performed locally.+Insertions can be batched up to one commit in order to reduce overheads. ```haskell commitFile :: FilePath -> Transaction a -> IO a@@ -29,3 +30,55 @@ $ liszt foo.liszt message -b 0 -f "%p\n" hello, world ```++# Representation++Liszt employs a 2-3 tree (special case of B tree) of skew binary random access lists as its on-disk format.++A liszt file consists of six types of contents: keys, payloads, tags, nodes, trees, and a footer.++`Key`s, `Tag`s, and `Payload`s are bytestrings with arbitrary length.++A pointer to a content with type `t` is denoted by `P t`.+It's a pair of the byte offset and the length of the content, encoded in unsigned VLQ.++A node is one of the following:++* Empty node: `0x00`+* 1-leaf: `0x01 (P Key) Spine`+* 2-leaf: `0x02 (P Key) Spine (P Key) Spine`+* 2-node: `0x12 (P Node) (P Key) Spine (P Node)`+* 3-node: `0x13 (P Node) (P Key) Spine (P Node) (P Key) Spine (P Node)`++Spine is a list of pairs of:++* Rank: the number of payloads in a tree+* Pointer to a tree++Spine prefixed by the length of the list.++* `Spine`: `Int [Int (P Tree)]`++Tree is either++* Tip: `0x80 Int Tag (P Payload)`+* Bin: `0x81 Int Tag (P Payload) (P Node) (P Node)`++where the Int represents the length of the `Tag`.++A footer is a root node padded to 256 bytes long, and the padding must end with the following byte sequence:++```+8df4 c865 fa3d 300a f3f8 9962 e049 e379 489c e4cd 2b52 a630 5584 004c 4953 5a54+```++(last 5 bytes is LISZT)++Committing is performed in the following steps. This append-only scheme grants robustness to some extent.++* Read the root node from the file.+* Append payloads.+* Append a new footer.++When an exception occurs in the middle, it will put the original footer to prevent corruption+(TODO: find the last footer in case of corrupted file).
app/server.hs view
@@ -28,4 +28,3 @@ (_, _, es) -> do name <- getProgName die $ unlines es ++ usageInfo name options-
+ benchmarks/bench.hs view
@@ -0,0 +1,9 @@+{-# LANGUAGE OverloadedStrings #-}+import Gauge.Main+import Control.Monad+import Database.Liszt++main = withLiszt "bench.liszt" $ \h -> do+ defaultMain+ [ bench "insert" $ nfIO $ commit h $ do+ forM_ [1..10] $ \i -> insert "test" (i :: Int) ]
liszt.cabal view
@@ -1,11 +1,11 @@--- This file has been generated from package.yaml by hpack version 0.20.0.+-- This file has been generated from package.yaml by hpack version 0.28.2. -- -- see: https://github.com/sol/hpack ----- hash: 5561a69ed2e98f7b84def2db25e066ef420542febb624a57bdbe15bb4c7c521a+-- hash: 7cbc3b7d828fd5acf9d0c70234dd19f9cbd19c7e12d3f47b7d96f40b339fc2fd name: liszt-version: 0.1+version: 0.2 synopsis: Append only key-list database description: Please see the README on GitHub at <https://github.com/fumieval/liszt#readme> category: Database@@ -18,7 +18,6 @@ license-file: LICENSE build-type: Simple cabal-version: >= 1.10- extra-source-files: ChangeLog.md README.md@@ -28,13 +27,23 @@ location: https://github.com/fumieval/liszt library+ exposed-modules:+ Database.Liszt+ Database.Liszt.Tracker+ Database.Liszt.Internal+ Database.Liszt.Network+ other-modules:+ Database.Liszt.Internal.Decoder+ Paths_liszt hs-source-dirs: src build-depends: base >=4.7 && <5 , binary , bytestring+ , cereal , containers+ , cpu , deepseq , directory , exceptions@@ -49,25 +58,24 @@ , text , transformers , unordered-containers+ , vector+ , vector-th-unbox , winery- exposed-modules:- Database.Liszt- Database.Liszt.Tracker- Database.Liszt.Internal- Database.Liszt.Network- other-modules:- Paths_liszt default-language: Haskell2010 executable liszt main-is: client.hs+ other-modules:+ Paths_liszt hs-source-dirs: app build-depends: base >=4.7 && <5 , binary , bytestring+ , cereal , containers+ , cpu , deepseq , directory , exceptions@@ -83,21 +91,25 @@ , text , transformers , unordered-containers+ , vector+ , vector-th-unbox , winery- other-modules:- Paths_liszt default-language: Haskell2010 executable lisztd main-is: server.hs+ other-modules:+ Paths_liszt hs-source-dirs: app- ghc-options: -threaded+ ghc-options: -threaded -O2 build-depends: base >=4.7 && <5 , binary , bytestring+ , cereal , containers+ , cpu , deepseq , directory , exceptions@@ -113,7 +125,43 @@ , text , transformers , unordered-containers+ , vector+ , vector-th-unbox , winery+ default-language: Haskell2010++benchmark bench+ type: exitcode-stdio-1.0+ main-is: bench.hs other-modules: Paths_liszt+ hs-source-dirs:+ benchmarks+ ghc-options: -O2+ build-depends:+ base >=4.7 && <5+ , binary+ , bytestring+ , cereal+ , containers+ , cpu+ , deepseq+ , directory+ , exceptions+ , filepath+ , fsnotify+ , gauge+ , liszt+ , network+ , reflection+ , scientific+ , sendfile+ , stm+ , stm-delay+ , text+ , transformers+ , unordered-containers+ , vector+ , vector-th-unbox+ , winery default-language: Haskell2010
src/Database/Liszt.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE LambdaCase #-} module Database.Liszt ( openLiszt, closeLiszt,@@ -13,23 +14,30 @@ insertRaw, commit, commitFile,- -- * Reader+ -- * Local reader+ RawPointer,+ count,+ fetchRange,+ fetchPayload,+ -- * Remote reader Offset(..), Request(..), defRequest, Connection, withConnection,- fetch+ fetch,+ LisztError(..) ) where +import Control.Monad.IO.Class import Database.Liszt.Internal import Database.Liszt.Network import Database.Liszt.Tracker import Data.Winery -- | Commit a 'Transaction' to a file.-commitFile :: FilePath -> Transaction a -> IO a-commitFile path m = withLiszt path $ \h -> commit h m+commitFile :: (MonadIO m) => FilePath -> Transaction a -> m a+commitFile path m = liftIO $ withLiszt path $ \h -> commit h m -- | Insert a value. insert :: Serialise a => Key -> a -> Transaction ()@@ -42,3 +50,25 @@ insertTagged :: (Serialise t, Serialise a) => Key -> t -> a -> Transaction () insertTagged k t v = insertRaw k (toEncoding t) (toEncoding v) {-# INLINE insertTagged #-}++-- | The number of entries in the stream+count :: MonadIO m => LisztHandle -> Key -> m Int+count h k = liftIO $ do+ root <- fetchRoot h+ maybe 0 spineLength <$> lookupSpine h k root++fetchRange :: MonadIO m => LisztHandle -> Key -> Int -> Int -> m [(Int, Tag, RawPointer)]+fetchRange h key i_ j_ = liftIO $ do+ root <- fetchRoot h+ lookupSpine h key root >>= \case+ Nothing -> return []+ Just spine -> do+ let len = spineLength spine+ let normalise x+ | x < 0 = min (len - 1) $ max 0 $ len + x+ | otherwise = min (len - 1) x+ let j = normalise j_+ let i = normalise i_+ spine' <- dropSpine h (len - j - 1) spine+ result <- takeSpine h (j - i + 1) spine' []+ return [(k, t, rp) | (k, (t, rp)) <- zip [i..] result]
src/Database/Liszt/Internal.hs view
@@ -17,13 +17,14 @@ , TransactionState -- * Reading , availableKeys- -- * Frame- , Frame(..)- , decodeFrame- , forceSpine+ -- * Node+ , Node(..)+ , peekNode+ , LisztDecodingException(..) -- * Fetching , Fetchable(..) , RawPointer(..)+ , fetchPayload -- * Footer , fetchRoot , footerSize@@ -33,6 +34,7 @@ , Spine , spineLength , QueryResult+ , wholeSpine , takeSpine , dropSpine , takeSpineWhile@@ -45,14 +47,15 @@ import Control.Monad.Catch import Control.Monad.IO.Class import Control.Monad.Trans.State.Strict+import Data.Bifunctor import qualified Data.ByteString as B import qualified Data.ByteString.Internal as B import qualified Data.IntMap.Strict as IM import Data.IORef-import Data.Proxy+import Data.Monoid import Data.Word-import Data.Winery-import qualified Data.Winery.Internal.Builder as WB+import Data.Winery.Internal.Builder+import Database.Liszt.Internal.Decoder import Foreign.ForeignPtr import Foreign.Ptr import GHC.Generics (Generic)@@ -63,32 +66,81 @@ -- | Tag is an extra value attached to a payload. This can be used to perform -- a binary search.-type Tag = WB.Encoding+type Tag = B.ByteString type Spine a = [(Int, a)] -newtype KeyPointer = KeyPointer RawPointer deriving (Show, Eq, Serialise, NFData)+newtype KeyPointer = KeyPointer RawPointer deriving (Show, Eq, NFData) data RawPointer = RP !Int !Int deriving (Show, Eq, Generic)-instance Serialise RawPointer instance NFData RawPointer where rnf r = r `seq` () -data Frame a = Empty+data Node t a = Empty | Leaf1 !KeyPointer !(Spine a) | Leaf2 !KeyPointer !(Spine a) !KeyPointer !(Spine a) | Node2 !a !KeyPointer !(Spine a) !a | Node3 !a !KeyPointer !(Spine a) !a !KeyPointer !(Spine a) !a- | Tree !Tag !RawPointer !a !a- | Leaf !Tag !RawPointer+ | Tip !t !RawPointer+ | Bin !t !RawPointer !a !a deriving (Generic, Show, Eq, Functor, Foldable, Traversable) -instance Serialise a => Serialise (Frame a)+encodeNode :: Node Encoding RawPointer -> Encoding+encodeNode Empty = word8 0x00+encodeNode (Leaf1 (KeyPointer p) s) = word8 0x01 <> encodeRP p <> encodeSpine s+encodeNode (Leaf2 (KeyPointer p) s (KeyPointer q) t) = word8 0x02+ <> encodeRP p <> encodeSpine s <> encodeRP q <> encodeSpine t+encodeNode (Node2 l (KeyPointer p) s r) = word8 0x12+ <> encodeRP l <> encodeRP p <> encodeSpine s <> encodeRP r+encodeNode (Node3 l (KeyPointer p) s m (KeyPointer q) t r) = word8 0x13+ <> encodeRP l <> encodeRP p <> encodeSpine s+ <> encodeRP m <> encodeRP q <> encodeSpine t <> encodeRP r+encodeNode (Tip t p) = word8 0x80 <> unsignedVarInt (getSize t) <> t <> encodeRP p+encodeNode (Bin t p l r) = word8 0x81 <> unsignedVarInt (getSize t) <> t+ <> encodeRP p <> encodeRP l <> encodeRP r -decodeFrame :: B.ByteString -> Frame RawPointer-decodeFrame = either (error . show) id $ getDecoder- $ schema (Proxy :: Proxy (Frame RawPointer))+encodeRP :: RawPointer -> Encoding+encodeRP (RP p l) = unsignedVarInt p <> unsignedVarInt l +encodeSpine :: Spine RawPointer -> Encoding+encodeSpine s = unsignedVarInt (length s) <> foldMap (\(r, p) -> unsignedVarInt r <> encodeRP p) s++data LisztDecodingException = LisztDecodingException deriving Show+instance Exception LisztDecodingException++peekNode :: Ptr Word8 -> IO (Node Tag RawPointer)+peekNode = runDecoder $ decodeWord8 >>= \case+ 0x00 -> return Empty+ 0x01 -> Leaf1 <$> kp <*> spine+ 0x02 -> Leaf2 <$> kp <*> spine <*> kp <*> spine+ 0x12 -> Node2 <$> rp <*> kp <*> spine <*> rp+ 0x13 -> Node3 <$> rp <*> kp <*> spine <*> rp <*> kp <*> spine <*> rp+ 0x80 -> Tip <$> bs <*> rp+ 0x81 -> Bin <$> bs <*> rp <*> rp <*> rp+ _ -> throwM LisztDecodingException+ where+ kp = fmap KeyPointer rp+ rp = RP <$> decodeVarInt <*> decodeVarInt+ spine = do+ len <- decodeVarInt+ replicateM len $ (,) <$> decodeVarInt <*> rp+ bs = do+ len <- decodeVarInt+ Decoder $ \src -> do+ fp <- B.mallocByteString len+ withForeignPtr fp $ \dst -> B.memcpy dst src len+ return $ DecodeResult (src `plusPtr` len) (B.PS fp 0 len)++instance Bifunctor Node where+ bimap f g = \case+ Empty -> Empty+ Leaf1 k s -> Leaf1 k (map (fmap g) s)+ Leaf2 j s k t -> Leaf2 j (map (fmap g) s) k (map (fmap g) t)+ Node2 l k s r -> Node2 (g l) k (map (fmap g) s) (g r)+ Node3 l j s m k t r -> Node3 (g l) j (map (fmap g) s) (g m) k (map (fmap g) t) (g r)+ Bin t p l r -> Bin (f t) p (g l) (g r)+ Tip t p -> Tip (f t) p+ data LisztHandle = LisztHandle { hPayload :: !Handle , refBuffer :: MVar (Int, ForeignPtr Word8)@@ -123,8 +175,8 @@ data TransactionState = TS { dbHandle :: !LisztHandle , freshId :: !Int- , pending :: !(IM.IntMap (Frame StagePointer))- , currentRoot :: !(Frame StagePointer)+ , pending :: !(IM.IntMap (Node Encoding StagePointer))+ , currentRoot :: !(Node Encoding StagePointer) , currentPos :: !Int } @@ -144,19 +196,19 @@ writeIORef refModified True cont hPayload -insertRaw :: Key -> Tag -> Encoding -> Transaction ()+insertRaw :: Key -> Encoding -> Encoding -> Transaction () insertRaw key tag payload = do lh <- gets dbHandle ofs <- gets currentPos liftIO $ append lh $ \h -> hPutEncoding h payload root <- gets currentRoot- modify $ \ts -> ts { currentPos = ofs + WB.getSize payload }- insertF key (insertSpine tag (ofs `RP` WB.getSize payload)) root >>= \case+ modify $ \ts -> ts { currentPos = ofs + getSize payload }+ insertF key (insertSpine tag (ofs `RP` getSize payload)) root >>= \case Pure t -> modify $ \ts -> ts { currentRoot = t } Carry l k a r -> modify $ \ts -> ts { currentRoot = Node2 l k a r } -addFrame :: Frame StagePointer -> Transaction StagePointer-addFrame f = state $ \ts -> (Uncommited (freshId ts), ts+addNode :: Node Encoding StagePointer -> Transaction StagePointer+addNode f = state $ \ts -> (Uncommited (freshId ts), ts { freshId = freshId ts + 1 , pending = IM.insert (freshId ts) f (pending ts) }) @@ -176,7 +228,7 @@ root <- fetchRoot h do (a, TS _ _ pendings root' offset1) <- runStateT transaction- $ TS h 0 IM.empty (fmap Commited root) offset0+ $ TS h 0 IM.empty (bimap bytes Commited root) offset0 let substP (Commited ofs) = return ofs substP (Uncommited i) = case IM.lookup i pendings of@@ -202,15 +254,15 @@ qv' <- traverse (traverse substP) qv r' <- substP r write (Node3 l' pk pv' m' qk qv' r')- substF (Tree t p l r) = do+ substF (Bin t p l r) = do l' <- substP l r' <- substP r- write (Tree t p l' r')- substF (Leaf t p) = write (Leaf t p)+ write (Bin t p l' r')+ substF (Tip t p) = write (Tip t p) writeFooter offset1 $ substF root' return ((), a)- `onException` writeFooter offset0 (write root)+ `onException` writeFooter offset0 (write $ first bytes root) where writeFooter ofs m = do modified <- readIORef (refModified h)@@ -222,26 +274,26 @@ writeIORef (refModified h) False h' = hPayload h- write :: Frame RawPointer -> StateT Int IO RawPointer+ write :: Node Encoding RawPointer -> StateT Int IO RawPointer write f = do ofs <- get- let e = toEncoding f+ let e = encodeNode f liftIO $ hPutEncoding h' e- put $! ofs + WB.getSize e- return $ RP ofs (WB.getSize e)+ put $! ofs + getSize e+ return $ RP ofs (getSize e) fetchKeyT :: KeyPointer -> Transaction Key fetchKeyT p = gets dbHandle >>= \h -> liftIO (fetchKey h p) -fetchStage :: StagePointer -> Transaction (Frame StagePointer)+fetchStage :: StagePointer -> Transaction (Node Encoding StagePointer) fetchStage (Commited p) = do h <- gets dbHandle- liftIO $ fmap Commited <$> fetchFrame h p+ liftIO $ bimap bytes Commited <$> fetchNode h p fetchStage (Uncommited i) = gets pending >>= return . maybe (error "fetch: not found") id . IM.lookup i insertF :: Key -> (Spine StagePointer -> Transaction (Spine StagePointer))- -> Frame StagePointer+ -> Node Encoding StagePointer -> Transaction (Result StagePointer) insertF k u Empty = Pure <$> (Leaf1 <$> allocKey k <*> u []) insertF k u (Leaf1 pk pv) = do@@ -260,8 +312,8 @@ LT -> do v <- u [] kp <- allocKey k- l <- addFrame $ Leaf1 kp v- r <- addFrame $ Leaf1 qk qv+ l <- addNode $ Leaf1 kp v+ r <- addNode $ Leaf1 qk qv return $ Carry l pk pv r EQ -> do v <- u pv@@ -271,8 +323,8 @@ case compare k vqk of LT -> do v <- u []- l <- addFrame $ Leaf1 pk pv- r <- addFrame $ Leaf1 qk qv+ l <- addNode $ Leaf1 pk pv+ r <- addNode $ Leaf1 qk qv kp <- allocKey k return $ Carry l kp v r EQ -> do@@ -281,8 +333,8 @@ GT -> do v <- u [] kp <- allocKey k- l <- addFrame $ Leaf1 pk pv- r <- addFrame $ Leaf1 kp v+ l <- addNode $ Leaf1 pk pv+ r <- addNode $ Leaf1 kp v return $ Carry l qk qv r insertF k u (Node2 l pk0 pv0 r) = do vpk0 <- fetchKeyT pk0@@ -291,7 +343,7 @@ fl <- fetchStage l insertF k u fl >>= \case Pure l' -> do- l'' <- addFrame l'+ l'' <- addNode l' return $ Pure $ Node2 l'' pk0 pv0 r Carry l' ck cv r' -> return $ Pure $ Node3 l' ck cv r' pk0 pv0 r EQ -> do@@ -301,7 +353,7 @@ fr <- fetchStage r insertF k u fr >>= \case Pure r' -> do- r'' <- addFrame r'+ r'' <- addNode r' return $ Pure $ Node2 l pk0 pv0 r'' Carry l' ck cv r' -> return $ Pure $ Node3 l pk0 pv0 l' ck cv r' insertF k u (Node3 l pk0 pv0 m qk0 qv0 r) = do@@ -311,11 +363,11 @@ fl <- fetchStage l insertF k u fl >>= \case Pure l' -> do- l'' <- addFrame l'+ l'' <- addNode l' return $ Pure $ Node3 l'' pk0 pv0 m qk0 qv0 r Carry l' ck cv r' -> do- bl <- addFrame (Node2 l' ck cv r')- br <- addFrame (Node2 m qk0 qv0 r)+ bl <- addNode (Node2 l' ck cv r')+ br <- addNode (Node2 m qk0 qv0 r) return $ Pure $ Node2 bl pk0 pv0 br EQ -> do v <- u pv0@@ -327,11 +379,11 @@ fm <- fetchStage m insertF k u fm >>= \case Pure m' -> do- m'' <- addFrame m'+ m'' <- addNode m' return $ Pure $ Node3 l pk0 pv0 m'' qk0 qv0 r Carry l' ck cv r' -> do- bl <- addFrame $ Node2 l pk0 pv0 l'- br <- addFrame $ Node2 r' qk0 qv0 r+ bl <- addNode $ Node2 l pk0 pv0 l'+ br <- addNode $ Node2 r' qk0 qv0 r return $ Pure $ Node2 bl ck cv br EQ -> do v <- u qv0@@ -340,34 +392,34 @@ fr <- fetchStage r insertF k u fr >>= \case Pure r' -> do- r'' <- addFrame r'+ r'' <- addNode r' return $ Pure $ Node3 l pk0 pv0 m qk0 qv0 r'' Carry l' ck cv r' -> do- bl <- addFrame $ Node2 l pk0 pv0 m- br <- addFrame $ Node2 l' ck cv r'+ bl <- addNode $ Node2 l pk0 pv0 m+ br <- addNode $ Node2 l' ck cv r' return $ Pure $ Node2 bl qk0 qv0 br-insertF _ _ (Tree _ _ _ _) = fail "Unexpected Tree"-insertF _ _ (Leaf _ _) = fail "Unexpected Leaf"+insertF _ _ (Bin _ _ _ _) = fail "Unexpected Tree"+insertF _ _ (Tip _ _) = fail "Unexpected Leaf" -data Result a = Pure (Frame a)+data Result a = Pure (Node Encoding a) | Carry !a !KeyPointer !(Spine a) !a -insertSpine :: Tag -> RawPointer -> Spine StagePointer -> Transaction (Spine StagePointer)+insertSpine :: Encoding -> RawPointer -> Spine StagePointer -> Transaction (Spine StagePointer) insertSpine tag p ((m, x) : (n, y) : ss) | m == n = do- t <- addFrame $ Tree tag p x y+ t <- addNode $ Bin tag p x y return $ (2 * m + 1, t) : ss insertSpine tag p ss = do- t <- addFrame $ Leaf tag p+ t <- addNode $ Tip tag p return $ (1, t) : ss -------------------------------------------------------------------------------- -- Fetching class Fetchable a where- fetchFrame :: LisztHandle -> a -> IO (Frame a)+ fetchNode :: LisztHandle -> a -> IO (Node Tag a) instance Fetchable RawPointer where- fetchFrame h (RP ofs len) = fetchFrame'+ fetchNode h (RP ofs len) = fetchNode' (hSeek (hPayload h) AbsoluteSeek (fromIntegral ofs)) h len fetchKey :: LisztHandle -> KeyPointer -> IO Key@@ -381,8 +433,8 @@ modifyIORef' keyCache (IM.insert ofs key) return key -fetchFrame' :: IO () -> LisztHandle -> Int -> IO (Frame RawPointer)-fetchFrame' seek h len = modifyMVar (refBuffer h) $ \(blen, buf) -> do+fetchNode' :: IO () -> LisztHandle -> Int -> IO (Node Tag RawPointer)+fetchNode' seek h len = modifyMVar (refBuffer h) $ \(blen, buf) -> do seek (blen', buf') <- if blen < len then do@@ -391,11 +443,11 @@ else return (blen, buf) f <- withForeignPtr buf' $ \ptr -> do _ <- hGetBuf (hPayload h) ptr len- let f = decodeFrame $ B.PS buf' 0 len- forceSpine f+ peekNode ptr return ((blen', buf'), f)+{-# INLINE fetchNode' #-} -lookupSpine :: Fetchable a => LisztHandle -> Key -> Frame a -> IO (Maybe (Spine a))+lookupSpine :: Fetchable a => LisztHandle -> Key -> Node tag a -> IO (Maybe (Spine a)) lookupSpine h k (Leaf1 p v) = do vp <- fetchKey h p if k == vp then return (Just v) else return Nothing@@ -407,40 +459,45 @@ lookupSpine h k (Node2 l p v r) = do vp <- fetchKey h p case compare k vp of- LT -> fetchFrame h l >>= lookupSpine h k+ LT -> fetchNode h l >>= lookupSpine h k EQ -> return (Just v)- GT -> fetchFrame h r >>= lookupSpine h k+ GT -> fetchNode h r >>= lookupSpine h k lookupSpine h k (Node3 l p u m q v r) = do vp <- fetchKey h p case compare k vp of- LT -> fetchFrame h l >>= lookupSpine h k+ LT -> fetchNode h l >>= lookupSpine h k EQ -> return (Just u) GT -> do vq <- fetchKey h q case compare k vq of- LT -> fetchFrame h m >>= lookupSpine h k+ LT -> fetchNode h m >>= lookupSpine h k EQ -> return (Just v)- GT -> fetchFrame h r >>= lookupSpine h k+ GT -> fetchNode h r >>= lookupSpine h k lookupSpine _ _ _ = return Nothing -availableKeys :: LisztHandle -> Frame RawPointer -> IO [Key]+availableKeys :: LisztHandle -> Node t RawPointer -> IO [Key] availableKeys _ Empty = return [] availableKeys h (Leaf1 k _) = pure <$> fetchKey h k availableKeys h (Leaf2 j _ k _) = sequence [fetchKey h j, fetchKey h k] availableKeys h (Node2 l k _ r) = do- lks <- fetchFrame h l >>= availableKeys h+ lks <- fetchNode h l >>= availableKeys h vk <- fetchKey h k- rks <- fetchFrame h r >>= availableKeys h+ rks <- fetchNode h r >>= availableKeys h return $ lks ++ vk : rks availableKeys h (Node3 l j _ m k _ r) = do- lks <- fetchFrame h l >>= availableKeys h+ lks <- fetchNode h l >>= availableKeys h vj <- fetchKey h j- mks <- fetchFrame h m >>= availableKeys h+ mks <- fetchNode h m >>= availableKeys h vk <- fetchKey h k- rks <- fetchFrame h r >>= availableKeys h+ rks <- fetchNode h r >>= availableKeys h return $ lks ++ vj : mks ++ vk : rks availableKeys _ _ = fail "availableKeys: unexpected frame" +fetchPayload :: LisztHandle -> RawPointer -> IO B.ByteString+fetchPayload LisztHandle{..} (RP ofs len) = do+ hSeek hPayload AbsoluteSeek (fromIntegral ofs)+ B.hGet hPayload len+ -------------------------------------------------------------------------------- -- Footer (root node) @@ -451,10 +508,10 @@ emptyFooter = B.pack [0,14,171,160,140,150,185,18,22,70,203,145,129,232,42,76,81,176,163,195,96,209,8,74,97,123,57,136,107,174,241,142,100,164,181,138,253,170,25,77,12,191,212,224,142,167,215,73,48,0,2,170,226,114,8,29,141,85,243,179,81,11,59,246,62,189,202,254,56,140,227,195,189,118,152,26,106,81,4,121,152,72,247,119,111,128,75,242,29,96,157,190,170,1,57,77,61,132,72,8,233,94,254,18,197,152,128,15,174,9,91,78,125,21,72,250,179,176,176,47,230,45,255,228,214,223,28,61,123,159,104,233,131,39,88,245,13,242,228,48,17,119,159,173,71,172,238,69,137,141,153,133,79,24,81,242,19,21,209,44,120,69,180,103,100,185,189,191,50,132,52,229,248,12,207,134,45,241,2,217,112,21,239,65,39,30,33,16,147,152,52,204,221,56,87,191,235,235,173,181,106,165,37,52,245,221,13,80,91,207,224,95,157,222,3,210,54,28,99,1,7,50,189,163,141,244,200,101,250,61,48,10,243,248,153,98,224,73,227,121,72,156,228,205,43,82,166,48,85,132,0,76,73,83,90,84] isFooter :: B.ByteString -> Bool-isFooter bs = B.drop 128 bs == B.drop 128 emptyFooter+isFooter bs = B.drop (footerSize - 64) bs == B.drop (footerSize - 64) emptyFooter -fetchRoot :: LisztHandle -> IO (Frame RawPointer)-fetchRoot h = fetchFrame'+fetchRoot :: LisztHandle -> IO (Node Tag RawPointer)+fetchRoot h = fetchNode' (hSeek (hPayload h) SeekFromEnd (-fromIntegral footerSize)) h footerSize --------------------------------------------------------------------------------@@ -470,15 +527,15 @@ dropSpine _ 0 s = return s dropSpine h n0 ((siz0, t0) : xs0) | siz0 <= n0 = dropSpine h (n0 - siz0) xs0- | otherwise = dropTree n0 siz0 t0 xs0+ | otherwise = dropBin n0 siz0 t0 xs0 where- dropTree 0 siz t xs = return $ (siz, t) : xs- dropTree n siz t xs = fetchFrame h t >>= \case- Tree _ _ l r+ dropBin 0 siz t xs = return $ (siz, t) : xs+ dropBin n siz t xs = fetchNode h t >>= \case+ Bin _ _ l r | n == 1 -> return $ (siz', l) : (siz', r) : xs- | n <= siz' -> dropTree (n - 1) siz' l ((siz', r) : xs)- | otherwise -> dropTree (n - siz' - 1) siz' r xs- Leaf _ _ -> return xs+ | n <= siz' -> dropBin (n - 1) siz' l ((siz', r) : xs)+ | otherwise -> dropBin (n - siz' - 1) siz' r xs+ Tip _ _ -> return xs _ -> error $ "dropTree: unexpected frame" where siz' = siz `div` 2@@ -488,26 +545,30 @@ takeSpine _ _ [] ps = return ps takeSpine h n ((siz, t) : xs) ps | n >= siz = takeAll h t ps >>= takeSpine h (n - siz) xs- | otherwise = takeTree h n siz t ps+ | otherwise = takeBin h n siz t ps -takeTree :: Fetchable a => LisztHandle -> Int -> Int -> a -> [QueryResult] -> IO [QueryResult]-takeTree _ n _ _ ps | n <= 0 = return ps-takeTree h n siz t ps = fetchFrame h t >>= \case- Tree tag p l r+takeBin :: Fetchable a => LisztHandle -> Int -> Int -> a -> [QueryResult] -> IO [QueryResult]+takeBin _ n _ _ ps | n <= 0 = return ps+takeBin h n siz t ps = fetchNode h t >>= \case+ Bin tag p l r | n == 1 -> return $ (tag, p) : ps- | n <= siz' -> takeTree h (n - 1) siz' l ((tag, p) : ps)+ | n <= siz' -> takeBin h (n - 1) siz' l ((tag, p) : ps) | otherwise -> do ps' <- takeAll h l ((tag, p) : ps)- takeTree h (n - siz' - 1) siz' r ps'- Leaf tag p -> return $ (tag, p) : ps+ takeBin h (n - siz' - 1) siz' r ps'+ Tip tag p -> return $ (tag, p) : ps _ -> error $ "takeTree: unexpected frame" where siz' = siz `div` 2 +wholeSpine :: Fetchable a => LisztHandle -> Spine a -> [QueryResult] -> IO [QueryResult]+wholeSpine h ((_, s) : ss) r = wholeSpine h ss r >>= takeAll h s+wholeSpine _ [] r = return r+ takeAll :: Fetchable a => LisztHandle -> a -> [QueryResult] -> IO [QueryResult]-takeAll h t ps = fetchFrame h t >>= \case- Tree tag p l r -> takeAll h l ((tag, p) : ps) >>= takeAll h r- Leaf tag p -> return ((tag, p) : ps)+takeAll h t ps = fetchNode h t >>= \case+ Bin tag p l r -> takeAll h l ((tag, p) : ps) >>= takeAll h r+ Tip tag p -> return ((tag, p) : ps) _ -> error $ "takeAll: unexpected frame" takeSpineWhile :: Fetchable a@@ -516,11 +577,11 @@ -> Spine a -> [QueryResult] -> IO [QueryResult] takeSpineWhile cond h = go where- go (t0 : ((siz, t) : xs)) ps = fetchFrame h t >>= \case- Leaf tag p+ go (t0 : ((siz, t) : xs)) ps = fetchNode h t >>= \case+ Tip tag p | cond tag -> takeAll h (snd t0) ps >>= go xs . ((tag, p):) | otherwise -> inner t0 ps- Tree tag p l r+ Bin tag p l r | cond tag -> takeAll h (snd t0) ps >>= go ((siz', l) : (siz', r) : xs) . ((tag, p):) | otherwise -> inner t0 ps@@ -530,11 +591,11 @@ go [t] ps = inner t ps go [] ps = return ps - inner (siz, t) ps = fetchFrame h t >>= \case- Leaf tag p+ inner (siz, t) ps = fetchNode h t >>= \case+ Tip tag p | cond tag -> return $ (tag, p) : ps | otherwise -> return ps- Tree tag p l r+ Bin tag p l r | cond tag -> go [(siz', l), (siz', r)] ((tag, p) : ps) | otherwise -> return ps _ -> error "takeSpineWhile: unexpected frame"@@ -547,50 +608,26 @@ -> Spine a -> IO (Maybe (Int, QueryResult, Spine a)) dropSpineWhile cond h = go 0 where- go !total (t0@(siz0, _) : ts@((siz, t) : xs)) = fetchFrame h t >>= \case- Leaf tag _+ go !total (t0@(siz0, _) : ts@((siz, t) : xs)) = fetchNode h t >>= \case+ Tip tag _ | cond tag -> go (total + siz0 + siz) xs- | otherwise -> dropTree total t0 ts- Tree tag _ l r+ | otherwise -> dropBin total t0 ts+ Bin tag _ l r | cond tag -> go (total + siz0 + 1) $ (siz', l) : (siz', r) : xs- | otherwise -> dropTree total t0 ts+ | otherwise -> dropBin total t0 ts _ -> error "dropSpineWhile: unexpected frame" where siz' = siz `div` 2- go total (t : ts) = dropTree total t ts+ go total (t : ts) = dropBin total t ts go _ [] = return Nothing - dropTree !total (siz, t) ts = fetchFrame h t >>= \case- Leaf tag p+ dropBin !total (siz, t) ts = fetchNode h t >>= \case+ Tip tag p | cond tag -> go (total + 1) ts | otherwise -> return $ Just (total, (tag, p), ts)- Tree tag p l r+ Bin tag p l r | cond tag -> go (total + 1) $ (siz', l) : (siz', r) : ts | otherwise -> return $ Just (total, (tag, p), (siz', l) : (siz', r) : ts) _ -> error "dropSpineWhile: unexpected frame" where siz' = siz `div` 2------------------------------------------------------------------------------------- Magical--copyByteString :: B.ByteString -> IO B.ByteString-copyByteString (B.PS fp ofs len) = do- fp' <- B.mallocByteString len- withForeignPtr fp' $ \dst -> withForeignPtr fp- $ \src -> B.memcpy dst (src `plusPtr` ofs) len- return (B.PS fp' 0 len)--forceSpine :: NFData a => Frame a -> IO (Frame a)-forceSpine f = case f of- Empty -> return Empty- Leaf1 _ s -> rnf s `seq` return f- Leaf2 _ s _ t -> rnf s `seq` rnf t `seq` return f- Node2 _ _ s _ -> rnf s `seq` return f- Node3 _ _ s _ _ t _ -> rnf s `seq` rnf t `seq` return f- Tree tag p l r -> do- tag' <- WB.bytes <$> copyByteString (WB.toByteString tag)- return (Tree tag' p l r)- Leaf tag p -> do- tag' <- WB.bytes <$> copyByteString (WB.toByteString tag)- return (Leaf tag' p)
+ src/Database/Liszt/Internal/Decoder.hs view
@@ -0,0 +1,70 @@+module Database.Liszt.Internal.Decoder (Decoder(..)+ , runDecoder+ , DecodeResult(..)+ , decodeWord8+ , decodeVarInt+ , decodeInt) where++import Control.Monad.Catch+import Foreign.Ptr+import Foreign.Storable+import Data.Bits+import Data.Word+import System.Endian++newtype Decoder a = Decoder { unDecoder :: Ptr Word8 -> IO (DecodeResult a) }++runDecoder :: Decoder a -> Ptr Word8 -> IO a+runDecoder m p = do+ DecodeResult _ a <- unDecoder m p+ return a++data DecodeResult a = DecodeResult {-# UNPACK #-} !(Ptr Word8) !a++instance MonadThrow Decoder where+ throwM e = Decoder $ \_ -> throwM e++instance Functor Decoder where+ fmap f m = Decoder $ \p -> do+ DecodeResult p' a <- unDecoder m p+ return (DecodeResult p' (f a))+ {-# INLINE fmap #-}++instance Applicative Decoder where+ pure a = Decoder $ \p -> pure $ DecodeResult p a+ {-# INLINE pure #-}+ m <*> n = Decoder $ \p -> do+ DecodeResult p' f <- unDecoder m p+ DecodeResult p'' a <- unDecoder n p'+ return $ DecodeResult p'' (f a)+ {-# INLINE (<*>) #-}++instance Monad Decoder where+ return = pure+ m >>= k = Decoder $ \p -> do+ DecodeResult p' a <- unDecoder m p+ unDecoder (k a) p'+ {-# INLINE (>>=) #-}+ (>>) = (*>)++decodeWord8 :: Decoder Word8+decodeWord8 = Decoder $ \ptr -> do+ a <- peek ptr+ return $ DecodeResult (ptr `plusPtr` 1) a+{-# INLINE decodeWord8 #-}++decodeVarInt :: Decoder Int+decodeVarInt = decodeWord8 >>= go+ where+ go n+ | testBit n 7 = do+ m <- decodeWord8 >>= go+ return $! unsafeShiftL m 7 .|. clearBit (fromIntegral n) 7+ | otherwise = return $ fromIntegral n+{-# INLINE decodeVarInt #-}++decodeInt :: Decoder Int+decodeInt = Decoder $ \ptr -> do+ a <- peek (castPtr ptr)+ return $ DecodeResult (ptr `plusPtr` 8) (fromIntegral $ fromBE64 a)+{-# INLINE decodeInt #-}
src/Database/Liszt/Network.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE LambdaCase, OverloadedStrings #-} module Database.Liszt.Network ( startServer , Connection@@ -8,40 +8,44 @@ , fetch) where import Control.Concurrent-import Control.Exception+import Control.Exception (evaluate, throw, throwIO) import Control.Monad+import Control.Monad.Catch+import Control.Monad.IO.Class import Database.Liszt.Tracker import Database.Liszt.Internal (hPayload, RawPointer(..))-import Data.Binary-import Data.Binary.Get+import Data.Serialize+import Data.Serialize.Get import Data.Winery import qualified Data.Winery.Internal.Builder as WB import qualified Data.ByteString.Char8 as B-import qualified Data.ByteString.Lazy as BL import qualified Network.Socket.SendFile.Handle as SF import qualified Network.Socket.ByteString as SB import qualified Network.Socket as S+import System.FilePath ((</>)) import System.IO import Text.Read (readMaybe) respond :: Tracker -> S.Socket -> IO () respond tracker conn = do msg <- SB.recv conn 4096- req <- try (evaluate $ decodeCurrent msg) >>= \case- Left e -> throwIO $ WineryError e- Right a -> return a- unless (B.null msg) $ handleRequest tracker req $ \lh lastSeqNo offsets -> do- let count = length offsets- _ <- SB.send conn $ encodeResp $ Right count- forM_ (zip [lastSeqNo - count + 1..] offsets) $ \(i, (tag, RP pos len)) -> do- SB.sendAll conn $ WB.toByteString $ mconcat- [ WB.word64 (fromIntegral i)- , WB.word64 (fromIntegral $ WB.getSize tag), tag- , WB.word64 $ fromIntegral len]- SF.sendFile' conn (hPayload lh) (fromIntegral pos) (fromIntegral len)+ unless (B.null msg) $ do+ req <- try (evaluate $ decodeCurrent msg) >>= \case+ Left e -> throwIO $ WineryError e+ Right a -> return a+ handleRequest tracker req $ \lh lastSeqNo offsets -> do+ let count = length offsets+ _ <- SB.send conn $ encodeResp $ Right count+ forM_ (zip [lastSeqNo - count + 1..] offsets) $ \(i, (tag, RP pos len)) -> do+ SB.sendAll conn $ WB.toByteString $ mconcat+ [ WB.word64 (fromIntegral i)+ , WB.word64 (fromIntegral $ B.length tag), WB.bytes tag+ , WB.word64 $ fromIntegral len]+ SF.sendFile' conn (hPayload lh) (fromIntegral pos) (fromIntegral len)+ respond tracker conn startServer :: Int -> FilePath -> IO ()-startServer port prefix = withLisztReader prefix $ \env -> do+startServer port prefix = withLisztReader $ \env -> do let hints = S.defaultHints { S.addrFlags = [S.AI_NUMERICHOST, S.AI_NUMERICSERV], S.addrSocketType = S.Stream } addr:_ <- S.getAddrInfo (Just hints) (Just "0.0.0.0") (Just $ show port) bracket (S.socket (S.addrFamily addr) (S.addrSocketType addr) (S.addrProtocol addr)) S.close $ \sock -> do@@ -52,10 +56,13 @@ forever $ do (conn, _) <- S.accept sock forkFinally (do- path <- decode . BL.fromStrict <$> SB.recv conn 4096- withTracker env path $ \t -> do+ bs <- SB.recv conn 4096+ path <- case decode bs of+ Right a -> return a+ Left _ -> throwM InvalidRequest+ withTracker env (prefix </> path) $ \t -> do SB.sendAll conn $ B.pack "READY"- forever $ respond t conn)+ respond t conn) $ \result -> do case result of Left ex -> case fromException ex of@@ -65,37 +72,43 @@ S.close conn encodeResp :: Either String Int -> B.ByteString-encodeResp = BL.toStrict . encode+encodeResp = encode newtype Connection = Connection (MVar S.Socket) -withConnection :: String -> Int -> B.ByteString -> (Connection -> IO r) -> IO r+withConnection :: (MonadIO m, MonadMask m) => String -> Int -> B.ByteString -> (Connection -> m r) -> m r withConnection host port path = bracket (connect host port path) disconnect -connect :: String -> Int -> B.ByteString -> IO Connection-connect host port path = do+connect :: MonadIO m => String -> Int -> B.ByteString -> m Connection+connect host port path = liftIO $ do let hints = S.defaultHints { S.addrFlags = [S.AI_NUMERICSERV], S.addrSocketType = S.Stream } addr:_ <- S.getAddrInfo (Just hints) (Just host) (Just $ show port) sock <- S.socket (S.addrFamily addr) (S.addrSocketType addr) (S.addrProtocol addr) S.connect sock $ S.addrAddress addr- SB.sendAll sock $ BL.toStrict $ encode path- _ <- SB.recv sock 4096- Connection <$> newMVar sock+ SB.sendAll sock $ encode path+ resp <- SB.recv sock 4096+ case resp of+ "READY" -> Connection <$> newMVar sock+ e -> case readMaybe <$> decode e of+ Right (Just (Left e')) -> throw (e' :: LisztError)+ Right (Just (Right ())) -> fail $ "connect: Unexpected response: " ++ show e+ _ -> fail $ "connect: Unexpected response: " ++ show e -disconnect :: Connection -> IO ()-disconnect (Connection sock) = takeMVar sock >>= S.close+disconnect :: MonadIO m => Connection -> m ()+disconnect (Connection sock) = liftIO $ takeMVar sock >>= S.close -fetch :: Connection -> Request -> IO [(Int, B.ByteString, B.ByteString)]-fetch (Connection msock) req = modifyMVar msock $ \sock -> do+fetch :: MonadIO m => Connection -> Request -> m [(Int, B.ByteString, B.ByteString)]+fetch (Connection msock) req = liftIO $ modifyMVar msock $ \sock -> do SB.sendAll sock $ serialiseOnly req- go sock $ runGetIncremental $ get >>= \case+ bs <- SB.recv sock 4096+ go sock $ flip runGetPartial bs $ get >>= \case Left e -> case readMaybe e of Just e' -> throw (e' :: LisztError) Nothing -> fail $ "Unknown error: " ++ show e Right n -> replicateM n ((,,) <$> get <*> get <*> get) where- go sock (Done _ _ a) = return (sock, a)+ go sock (Done a _) = return (sock, a) go sock (Partial cont) = do bs <- SB.recv sock 4096- if B.null bs then go sock $ cont Nothing else go sock $ cont $ Just bs- go _ (Fail _ _ str) = fail $ show req ++ ": " ++ str+ if B.null bs then go sock $ cont "" else go sock $ cont bs+ go _ (Fail str _) = fail $ show req ++ ": " ++ str
src/Database/Liszt/Tracker.hs view
@@ -32,7 +32,6 @@ import Data.Text (Text) import Data.Winery import Foreign.ForeignPtr-import qualified Data.Winery.Internal.Builder as WB import GHC.Generics (Generic) import System.Directory import System.FilePath@@ -74,25 +73,25 @@ instance Exception LisztError data Tracker = Tracker- { vRoot :: !(TVar (Frame CachePointer))+ { vRoot :: !(TVar (Node Tag CachePointer)) , vUpdated :: !(TVar Bool) , vPending :: !(TVar [STM (IO ())]) , vReaders :: !(TVar Int) , followThread :: !ThreadId- , trackerPath :: !B.ByteString+ , filePath :: !FilePath , streamHandle :: !LisztHandle , cache :: !Cache } data Cache = Cache- { primaryCache :: TVar (IM.IntMap (Frame CachePointer))- , secondaryCache :: TVar (IM.IntMap (Frame CachePointer))+ { primaryCache :: TVar (IM.IntMap (Node Tag CachePointer))+ , secondaryCache :: TVar (IM.IntMap (Node Tag CachePointer)) } newtype CachePointer = CachePointer RawPointer instance Given Cache => Fetchable CachePointer where- fetchFrame h (CachePointer p@(RP ofs _)) = join $ atomically $ do+ fetchNode h (CachePointer p@(RP ofs _)) = join $ atomically $ do let Cache{..} = given pcache <- readTVar primaryCache case IM.lookup ofs pcache of@@ -104,7 +103,7 @@ writeTVar primaryCache $! IM.insert ofs x pcache return (pure x) Nothing -> return $ do- x <- fmap CachePointer <$> fetchFrame h p+ x <- fmap CachePointer <$> fetchNode h p atomically $ modifyTVar' primaryCache $ IM.insert ofs x return x @@ -113,9 +112,8 @@ readTVar primaryCache >>= writeTVar secondaryCache writeTVar primaryCache IM.empty -createTracker :: WatchManager -> FilePath -> B.ByteString -> IO Tracker-createTracker man prefix trackerPath = do- let filePath = prefix </> B.unpack trackerPath+createTracker :: WatchManager -> FilePath -> IO Tracker+createTracker man filePath = do exist <- doesFileExist filePath unless exist $ throwIO FileNotFound streamHandle <- openLiszt filePath@@ -128,34 +126,27 @@ _ -> False) $ const $ void $ atomically $ writeTVar vUpdated True - fptr <- B.mallocByteString 4096- let wait = atomically $ do b <- readTVar vUpdated unless b retry writeTVar vUpdated False - hSeek (hPayload streamHandle) SeekFromEnd (-fromIntegral footerSize)- let seekRoot prevSize = do- n <- withForeignPtr fptr $ \p -> hGetBuf (hPayload streamHandle) p 4096- if n == 0- then do- let bs = B.PS fptr (max 0 $ prevSize - footerSize) footerSize- if isFooter bs- then try (evaluate $ decodeFrame bs) >>= \case- Left (_ :: DecodeException) -> do- wait- seekRoot 0- Right a -> forceSpine a- else do- wait- seekRoot 0- else seekRoot n+ let seekRoot = do+ hSeek (hPayload streamHandle) SeekFromEnd (-fromIntegral footerSize)+ bs@(B.PS fp _ _) <- B.hGet (hPayload streamHandle) footerSize+ if isFooter bs+ then try (withForeignPtr fp peekNode) >>= \case+ Left LisztDecodingException -> do+ putStrLn "WARNING: isFooter passed, but it's not valid"+ wait+ seekRoot+ Right a -> return a+ else wait >> seekRoot cache <- Cache <$> newTVarIO IM.empty <*> newTVarIO IM.empty followThread <- forkFinally (forever $ do- newRoot <- fmap CachePointer <$> seekRoot 0+ newRoot <- fmap CachePointer <$> seekRoot join $ atomically $ do flipCache cache writeTVar vRoot newRoot@@ -169,16 +160,15 @@ data LisztReader = LisztReader { watchManager :: WatchManager- , vTrackers :: TVar (HM.HashMap B.ByteString Tracker)- , prefix :: FilePath+ , vTrackers :: TVar (HM.HashMap FilePath Tracker) } -withLisztReader :: FilePath -> (LisztReader -> IO ()) -> IO ()-withLisztReader prefix k = do+withLisztReader :: (LisztReader -> IO ()) -> IO ()+withLisztReader k = do vTrackers <- newTVarIO HM.empty withManager $ \watchManager -> k LisztReader{..} -acquireTracker :: LisztReader -> B.ByteString -> IO Tracker+acquireTracker :: LisztReader -> FilePath -> IO Tracker acquireTracker LisztReader{..} path = join $ atomically $ do streams <- readTVar vTrackers case HM.lookup path streams of@@ -186,7 +176,7 @@ modifyTVar' (vReaders s) (+1) return (return s) Nothing -> return $ do- s <- createTracker watchManager prefix path+ s <- createTracker watchManager path atomically $ modifyTVar vTrackers (HM.insert path s) return s @@ -195,13 +185,13 @@ n <- readTVar vReaders if n <= 1 then do- modifyTVar' vTrackers (HM.delete trackerPath)+ modifyTVar' vTrackers (HM.delete filePath) return $ do killThread followThread closeLiszt streamHandle else return () <$ writeTVar vReaders (n - 1) -withTracker :: LisztReader -> B.ByteString -> (Tracker -> IO a) -> IO a+withTracker :: LisztReader -> FilePath -> (Tracker -> IO a) -> IO a withTracker env path = bracket (acquireTracker env path) (releaseTracker env) handleRequest :: Tracker@@ -230,20 +220,20 @@ SeqNo n -> takeSpine streamHandle (min reqLimit $ ofs - n + 1) spine' [] >>= cont streamHandle ofs WineryTag sch name p -> do dec <- handleWinery sch name- takeSpineWhile ((>=p) . dec . WB.toByteString) streamHandle spine' [] >>= cont streamHandle ofs+ takeSpineWhile ((>=p) . dec) streamHandle spine' [] >>= cont streamHandle ofs case reqTo of FromEnd ofs -> goSeqNo (len - ofs) SeqNo ofs -> goSeqNo ofs WineryTag sch name p -> do dec <- handleWinery sch name- dropSpineWhile ((>=p) . dec . WB.toByteString) streamHandle spine >>= \case+ dropSpineWhile ((>=p) . dec) streamHandle spine >>= \case Nothing -> cont streamHandle 0 [] Just (dropped, e, spine') -> case reqFrom of FromEnd n -> takeSpine streamHandle (min reqLimit $ n - dropped + 1) spine' [e] >>= cont streamHandle (len - dropped) SeqNo n -> takeSpine streamHandle (min reqLimit $ len - dropped - n + 1) spine' [e] >>= cont streamHandle (len - dropped) WineryTag sch' name' q -> do dec' <- handleWinery sch' name'- takeSpineWhile ((>=q) . dec' . WB.toByteString) streamHandle spine' [e] >>= cont streamHandle (len - dropped)+ takeSpineWhile ((>=q) . dec') streamHandle spine' [e] >>= cont streamHandle (len - dropped) where handleWinery :: Schema -> [Text] -> IO (B.ByteString -> Scientific) handleWinery sch names = either (throwIO . WinerySchemaError . show) pure