arena (empty) → 0.1

raw patch · 7 files changed

+655/−0 lines, 7 filesdep +arenadep +basedep +bytessetup-changed

Dependencies added: arena, base, bytes, bytestring, containers, criterion, digest, directory, filepath, mtl, persistent-vector, safe, semigroups, unix

Files

LICENSE +30/−0
Setup.hs +2/−0
arena.cabal +79/−0
bench/bench.hs +32/−0
src/Database/Arena.hs +330/−0
tests/example.hs +128/−0
tests/test.hs +54/−0

+ LICENSE view

@@ -0,0 +1,30 @@+Copyright (c) 2015-2016, davean++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 davean 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.

+ Setup.hs view

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

+ arena.cabal view

@@ -0,0 +1,79 @@+name:                arena+version:             0.1+synopsis:            A journaled data store+license:             BSD3+license-file:        LICENSE+author:              davean+maintainer:          davean <davean@xkcd.com>+copyright:           Copyright (C) 2015-2016 davean+stability:           provisional+category:            Database+build-type:          Simple+cabal-version:       >=1.22+bug-reports:         oss@xkcd.com+description:+  @arena@ provides durable storage of data and summaries of that data.++  On insert, data is written to a journal. Each piece of data is added to a running summary of the current journal. When the summary indicates the correct amount of data has accumulated, the journal data is moved, as a block accompanied by its summary, to long-term storage. The data type, summary type, and accumulation policy are configurable.++source-repository head+  type: git+  location: http://git.xkrd.net/distributed-systems/arena++library+  hs-source-dirs:      src+  default-language:    Haskell2010+  exposed-modules:+        Database.Arena+  build-depends:+        base >=4.8 && <5+      , bytestring >= 0.10.4.1+      , filepath >= 1.2.0.0+      , semigroups >= 0.17+      , mtl >= 2.2+      , bytes >= 0.14+      , digest >= 0.0.1.2+      , directory >= 1.2.5.0+      , safe >= 0.3+      , containers >= 0.5.4.0+      , unix >= 2.7.1.0+      , persistent-vector >= 0.1.1++test-suite test+  type: exitcode-stdio-1.0+  default-language: Haskell2010+  hs-source-dirs: tests+  main-is: test.hs+  build-depends:+      base >=4.8 && <5+    , arena+    , directory+    , mtl+    , semigroups++test-suite example+  type: exitcode-stdio-1.0+  default-language: Haskell2010+  hs-source-dirs: tests+  main-is: example.hs+  build-depends:+      base >=4.8 && <5+    , arena+    , bytes+    , containers+    , directory+    , mtl+    , semigroups++benchmark bench+  type: exitcode-stdio-1.0+  default-language: Haskell2010+  hs-source-dirs: bench+  main-is: bench.hs+  build-depends:+      base >=4.8 && <5+    , arena+    , directory+    , criterion >= 1.1.1.0+    , mtl+    , semigroups

+ bench/bench.hs view

@@ -0,0 +1,32 @@+{-# LANGUAGE OverloadedStrings   #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections       #-}++module Main where++import           Criterion.Main+import           Data.Semigroup+import           Database.Arena+import           System.Directory++prepDirectory :: IO ()+prepDirectory = do+  createDirectoryIfMissing True "test_data"+  removeDirectoryRecursive "test_data"+  createDirectoryIfMissing True "test_data/journal"+  createDirectoryIfMissing True "test_data/data"++main = defaultMain [+        bgroup "inserts" [ bench "20"    . nfIO $ inserts 20+                         , bench "10000" . nfIO $ inserts 10000]+       ]+++startTestArena :: Int -> IO (ArenaDB (Sum Int, Sum Int) Int Int)+startTestArena n = startArena ((1,) . Sum) (getSum . snd) ((succ n <=) . getSum . fst) "test_data"++inserts :: Int -> IO ()+inserts n = do+  prepDirectory+  ad <- startTestArena n+  runArena ad $ mapM_ addData [1..n]

+ src/Database/Arena.hs view

@@ -0,0 +1,330 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE RecordWildCards            #-}+{-# LANGUAGE ScopedTypeVariables        #-}++module Database.Arena+    (+      ArenaLocation(..)++    , ArenaT+    , Arena+    , ArenaDB+    , runArenaT+    , runArena+    , startArena, initArena+    , addData+    , accessData+    ) where++import           Control.Applicative+import           Control.Concurrent.MVar+import           Control.Monad+import           Control.Monad.Reader+import           Control.Monad.Trans+import           Data.Bytes.Get+import           Data.Bytes.Put+import           Data.Bytes.Serial+import qualified Data.ByteString         as BS+import qualified Data.ByteString.Lazy    as BSL+import           Data.Digest.CRC32+import           Data.Either+import           Data.Foldable+import           Data.IORef+import           Data.List+import           Data.Maybe+import           Data.Semigroup+import qualified Data.Set                as Set+import           Data.String+import           Data.Typeable           (cast)+import qualified Data.Vector.Persistent  as PV+import           Data.Word+import           GHC.IO.Device+import           GHC.IO.Exception+import qualified GHC.IO.FD               as FD+import           GHC.IO.Handle+import qualified GHC.IO.Handle.FD        as FD+import           GHC.IO.Handle.Internals+import           GHC.IO.Handle.Types+import           Safe+import           System.Directory+import           System.FilePath+import           System.IO+import           System.IO.Error+import           System.Posix.IO         (handleToFd)+import           System.Posix.Types      (Fd (Fd))+import           System.Posix.Unistd     (fileSynchroniseDataOnly)++-- | The base directory for the arena files to be stored under.+newtype ArenaLocation = ArenaLocation { getArenaLocation :: FilePath }+  deriving (Eq, Ord, Read, Show, IsString)++type ArenaID     = Word32+type JournalHash = Word32++data JournalFrame =+    JF {+      jLength :: Word32+      -- ^ The amount of data stored in the frame.+    , jHash :: JournalHash+      -- ^ A hash of the data stored in the frame so we can detect write corruption.+      --   This is sufficient for the entire frame as all other elements are fixed+      --   size and directly impact the data's hash.+    , jData :: BS.ByteString+      -- ^ The user's serialized data.+    }++mkJournal :: Serial a => a -> JournalFrame+mkJournal a = JF (fromIntegral . BS.length $ d) (crc32 d) d+  where d = runPutS . serialize $ a++instance Serial JournalFrame where+    serialize (JF l h d) = serialize l *> serialize h *> putByteString d+    deserialize = do+      l <- deserialize+      h <- deserialize+      d <- getByteString . fromIntegral $ l+      if h == crc32 d+      then return $ JF l h d+      else fail "Journal Frame failed CRC check---Also, uh, this method shouldn't be in Monad"++data OpenJournal a b =+  OJ {+    ojArenaID :: ArenaID+    -- ^ The current datablock generation, starting from zero and strictly monotonic.+  , ojHandle :: Handle+    -- ^ The journal file, opened and positioned for writing+  , ojSummary :: Option b+    -- ^ The current summary, if any values have been stored in the journal+  , ojVals :: PV.Vector a+    -- ^ The values currently held in the journal on disk, in a 'PV.Vector' (instead of a list) to+    --   improve GC performance, and provide fast snoc to avoid the use of reverse.+  }++readDataFile :: (MonadIO m, Serial a) => ArenaLocation -> ArenaID -> m [a]+readDataFile l ai = liftIO $ do+  d <- BSL.readFile (dataFile l ai)+  return $ runGetL (many deserialize) d++syncHandle :: Handle -> IO ()+syncHandle h = do+    hFlush h+    unsafeSyncHandle h+  where+    unsafeSyncHandle h@(DuplexHandle {}) =+      ioError (ioeSetErrorString (mkIOError IllegalOperation+                                  "unsafeSyncHandle" (Just h) Nothing)+               "unsafeSyncHandle only works on file descriptors")+    unsafeSyncHandle h@(FileHandle _ m) = do+      withMVar m $ \h_@Handle__{haType=_,..} -> do+        case cast haDevice of+          Nothing -> ioError (ioeSetErrorString (mkIOError IllegalOperation+                                                 "unsafeSyncHandle" (Just h) Nothing)+                              ("handle is not a file descriptor"))+          Just fd -> do+            flushWriteBuffer h_+            fileSynchroniseDataOnly . Fd . FD.fdFD $ fd++withFileSync :: FilePath -> (Handle -> IO r) -> IO r+withFileSync fp f = liftIO $ withFile fp WriteMode go+  where go h = f h <* syncHandle h++data ArenaDB summary finalized d = ArenaDB {+      adSummarize      :: d -> summary+      -- ^ Convert the user's data to an instance of the summary semigroup+    , adFinalize       :: summary -> finalized+      -- ^ Removal of data only used for the merge operation or determining fullness,+      --   changing the summary to a version used purely for the datablock indexing.+    , adArenaFull      :: summary -> Bool+      -- ^ Determine if we've reached a datablock boundary.+    , adArenaLocation  :: ArenaLocation+      -- ^ The location on the filesystem in which we store the data and journal.+    , adDataRef        :: IORef [(finalized, IO [d])]+      -- ^ The list of datablocks, seen as their summary and accessors, for read operations.+    , adCurrentJournal :: MVar (OpenJournal d summary)+      -- ^ The journal for writing new data, in an 'MVar' so we can lock it for consistency when+      --   doing writes or data access.+    }++-- | * @s@: The summary semigroup for the journal data.+--   * @f@: The finalized summary associated with a datablock.+--   * @d@: The user's datatype stored in the arena.+newtype ArenaT s f d m a = ArenaT { unArenaT :: ReaderT (ArenaDB s f d) m a }+    deriving (Functor, Applicative, Monad, MonadTrans, MonadReader (ArenaDB s f d), MonadIO)++runArenaT :: ArenaDB s f d -> ArenaT s f d m a -> m a+runArenaT ad = flip runReaderT ad . unArenaT++type Arena s f d a = ArenaT s f d IO a++runArena :: ArenaDB s f d -> Arena s f d a -> IO a+runArena = runArenaT++summarize :: Monad m => d -> ArenaT s f d m s+summarize d = asks adSummarize <*> pure d++finalize :: Monad m => s -> ArenaT s f d m f+finalize s = asks adFinalize <*> pure s++journalDir, dataDir, tempJournal :: ArenaLocation -> FilePath+journalDir (ArenaLocation fp) = fp            </> "journal"+dataDir    (ArenaLocation fp) = fp            </> "data"+tempJournal al                = journalDir al </> "temp"++journalFile, dataFile, dataSummaryFile :: ArenaLocation -> ArenaID -> FilePath+journalFile     al i = journalDir al </> show i+dataFile        al i = dataDir    al </> addExtension (show i) "data"+dataSummaryFile al i = dataDir    al </> addExtension (show i) "header"++journalDir', dataDir', tempJournal' :: Monad m => ArenaT s f d m FilePath+journalDir'  = asks (journalDir  . adArenaLocation)+dataDir'     = asks (dataDir     . adArenaLocation)+tempJournal' = asks (tempJournal . adArenaLocation)++journalFile', dataFile', dataSummaryFile' :: Monad m => ArenaID -> ArenaT s f d m FilePath+journalFile'     ai = asks (journalFile     . adArenaLocation) <*> pure ai+dataFile'        ai = asks (dataFile        . adArenaLocation) <*> pure ai+dataSummaryFile' ai = asks (dataSummaryFile . adArenaLocation) <*> pure ai++data TheFiles = TheFiles {+      theJournalDir+    , theDataDir+    , theTempJournal+    , theJournalFile+    , theDataFile+    , theDataSummaryFile :: FilePath+    } deriving (Eq, Ord, Read, Show)++theFiles :: Monad m => Maybe ArenaID -> ArenaT s f d m TheFiles+theFiles Nothing   = theFiles (Just $ -1)+theFiles (Just ai) =+  TheFiles <$> journalDir'     <*> dataDir'     <*> tempJournal'+           <*> journalFile' ai <*> dataFile' ai <*> dataSummaryFile' ai++-- | Setup the directory structure for Arena.+--   This is performed implicitly by 'startArena'.+initArena :: ArenaLocation -> IO ()+initArena al = do+  createDirectoryIfMissing True . journalDir $ al+  createDirectoryIfMissing True . dataDir $ al++-- | Launch an 'ArenaDB', using the given summarizing, finalizing, and block policy functions,+--   at the given 'ArenaLocation'.+--+--   __NB:__ Two 'ArenaDB's must not be run concurrently with a shared 'ArenaLocation'.+--   Data loss from the journal is likely to result.+startArena+  :: (Serial d, Serial f, Semigroup s)+    => (d -> s) -> (s -> f) -> (s -> Bool) -> ArenaLocation -> IO (ArenaDB s f d)+startArena adSummarize adFinalize adArenaFull adArenaLocation = do+    initArena adArenaLocation+    runArena fakeConf $ do+      adCurrentJournal <- internArenas >>= cleanJournal >>= liftIO . newMVar+      adDataRef        <- readAllData  >>= liftIO . newIORef+      return ArenaDB {..}+    where fakeConf = ArenaDB { adCurrentJournal = error "uninitialized current journal"+                               , adDataRef        = error "uninitialized data ref"+                               , .. }++readJournalFile :: (Serial d, MonadIO m) => ArenaID -> ArenaT s f d m (PV.Vector d)+readJournalFile ai = do+  d <- journalFile' ai >>= liftIO . BSL.readFile+  return . PV.fromList . map (head . rights . pure . runGetS deserialize . jData) . runGetL (many deserialize) $ d+++cleanJournal :: (MonadIO m, Serial d, Semigroup s) => ArenaID -> ArenaT s f d m (OpenJournal d s)+cleanJournal ai = do+  TheFiles {..} <- theFiles (Just ai)+  liftIO $ doesFileExist theTempJournal >>= (`when` removeFile theTempJournal)+  je <- liftIO $ doesFileExist theJournalFile+  case je of+    False -> do+     jh <- liftIO $ openFile theJournalFile WriteMode+     return $ OJ ai jh (Option Nothing) mempty+    True -> do+      as <- readJournalFile ai+      liftIO $ withFileSync theTempJournal $ \h ->+        mapM_ (BS.hPutStr h . runPutS . serialize . mkJournal) as+      liftIO $ renameFile theTempJournal theJournalFile+      jh <- liftIO $ openFile theJournalFile AppendMode+      as' <- if null as+            then return $ Option Nothing+            else Option . Just <$> regenerateSummary as+      return $ OJ ai jh as' as++readArenaIDs :: MonadIO m => FilePath -> m [ArenaID]+readArenaIDs dir = liftIO go+    where go = mapMaybe readMay <$> listDirectory dir++internArenas :: (MonadIO m, Serial d, Serial f, Semigroup s) => ArenaT s f d m ArenaID+internArenas = do+  js <- journalDir' >>= readArenaIDs+  if null js+  then return 0+  else do+    let latest = maximum js+        old    = delete latest js+    mapM_ internArenaFile old+    return latest++regenerateSummary :: (Semigroup s, Monad m) => PV.Vector d -> ArenaT s f d m s+regenerateSummary ds = foldl1 (<>) <$> traverse summarize ds++internArenaFile :: (MonadIO m, Serial f, Serial d, Semigroup s) => ArenaID -> ArenaT s f d m ()+internArenaFile ai = do+  TheFiles {..} <- theFiles (Just ai)+  as  <- readJournalFile ai+  fas <- regenerateSummary as >>= finalize+  liftIO $ do withFileSync theDataFile  $ \h -> mapM_ (BS.hPutStr h . runPutS . serialize) as+              withFileSync theDataSummaryFile $ \h -> BS.hPutStr h . runPutS . serialize $ fas+              removeFile theJournalFile+++readDataFile' :: (MonadIO m, Serial d) => ArenaID -> ArenaT s f d m (IO [d])+readDataFile' ai = asks (readDataFile . adArenaLocation) <*> pure ai++readAllData :: (MonadIO m, Serial f, Serial d) => ArenaT s f d m [(f, IO [d])]+readAllData = do+  TheFiles {..} <- theFiles Nothing+  ds <- Set.fromList . mapMaybe (readMay . dropExtension) <$> liftIO (listDirectory theDataDir)+  forM (Set.toList ds) $ \d -> do+                dsf <- dataSummaryFile' d+                c <- runGetL deserialize <$> liftIO (BSL.readFile dsf)+                rdf <- readDataFile' d+                return (c, rdf)++-- | Access an atomic snapshot of the 'ArenaDB' as a list of summaries and accessors+--   for their associated data. One datablock is the journal, and thus does not satisfy+--   the block policy.+--+--   The returned IO actions in the list provide access to a state consistent with the time+--   when the list of accessors was returned, but do not hold more then the journal at that+--   time's contents in memory.+accessData :: MonadIO m => ArenaT s f d m [(f, IO [d])]+accessData = do+  ArenaDB {..} <- ask+  liftIO . withMVar adCurrentJournal $ \(OJ _ _ s as) -> do+               d <- readIORef adDataRef+               return $ if null as+                        then d+                        else (adFinalize . fromJust . getOption $ s, return . toList $ as) : d++-- | Durably insert a piece of data into the 'ArenaDB'.+--   This funtion returns after the data is sync'd to disk.+addData :: (MonadIO m, Serial d, Serial f, Semigroup s) => d -> ArenaT s f d m ()+addData d = do+  conf@ArenaDB {..} <- ask+  liftIO . modifyMVar_ adCurrentJournal $ \(OJ ai h s ds) -> do+              BS.hPutStr h . runPutS . serialize . mkJournal $ d+              let s' = s <> (pure . adSummarize $ d)+              syncHandle h+              case adArenaFull . fromJust . getOption $ s' of+                False -> return $ OJ ai h s' (ds `PV.snoc` d)+                True -> do+                  hClose h+                  let (ArenaT rdr) = internArenaFile ai -- :: Arena s f d ()+                  runReaderT rdr conf+                  atomicModifyIORef' adDataRef $ \ods ->+                    ((adFinalize . fromJust . getOption $ s', readDataFile adArenaLocation ai):ods, ())+                  let ai' = ai + 1+                  h' <- openFile (journalFile adArenaLocation ai') WriteMode+                  return $ OJ ai' h' mempty mempty

+ tests/example.hs view

@@ -0,0 +1,128 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}++module Main where++import           Control.Monad.Trans+import           GHC.Generics+import           Data.Semigroup+import           Data.Bytes.Serial+import           Database.Arena+import           Data.List+import           Data.Map (Map)+import qualified Data.Map as Map+import           Data.Ord++-- | Suppose we have a bunch of (simplified) web server log entries, each containing the source IP,+--   time, and number of bytes served for an HTTP request:+data LogEntry =+  LogEntry {+    leIP          :: String+  , leTimeStamp   :: Time+  , leBytesServed :: Int+  } deriving (Generic, Show)++type Time = Int++-- | We might query this data to find the amount of data we've served over a given time period:+bytesServedInPeriod :: Time -> Time -> [LogEntry] -> Int+bytesServedInPeriod start end les = sum [ bytes | LogEntry _ ts bytes <- les+                                                , ts >= start+                                                , ts <= end ]++-- | To execute this query more more efficiently, we'd like to batch up our entries and store+--   them alongside an index of some salient information about them.  Specifically, we'd like to+--   know how much data we've served, over what time period we served it, and the source IP that+--   made the most requests (the "heaviest hitter").+data LogIndex =+  LogIndex {+    liHeavyHitter :: String+  , liBytesServed :: Int+  , liStartTime   :: Time+  , liEndTime     :: Time+  } deriving (Generic, Show)++-- | Conceptually, our data can now look like @[(LogIndex, [LogEntry])]@+fastBytesServedInPeriod :: Time -> Time -> [(LogIndex, [LogEntry])] -> Int+fastBytesServedInPeriod start end blocks = sum fullyContained + sum partiallyContained+    where fullyContained = [ bs | (LogIndex _ bs liStart liEnd, _) <- blocks+                                , liStart >= start && liEnd <= end ]+          partiallyContained = [ bytesServedInPeriod start end es+                                 | (LogIndex _ bs liStart liEnd, es) <- blocks+                                 , (liStart <= start && start <= liEnd) || (liStart <= end && end <= liEnd)]++-- | To build these indexed blocks <https://en.wikipedia.org/wiki/Online_algorithm online>, we need+--   to keep track of some more information.  In particular, we need to keep count of the log+--   entries for each IP, the total bytes served, the start and end times, and the total count of+--   the log entries as we process them.+data LogSummary =+  LogSummary {+    lsHeavyHitters :: Map String Int+  , lsBytesServed  :: Sum Int+  , lsStartTime    :: Min Time+  , lsEndTime      :: Max Time+  , lsCount        :: Sum Int+  }++-- | By making 'LogSummary' a 'Semigroup', @arena@ can maintain the running summary we require+--   automatically.+instance Semigroup LogSummary where+    LogSummary hh bs st et c <> LogSummary hh' bs' st' et' c' =+       LogSummary (Map.unionWith (+) hh hh') (bs <> bs') (st <> st') (et <> et') (c <> c')++-- | All we need to be able to do is summarize individual 'LogEntry's:+summarizeEntry :: LogEntry -> LogSummary+summarizeEntry (LogEntry ip t bytes) =+  LogSummary (Map.singleton ip 1) (Sum bytes) (Min t) (Max t) (Sum 1)++-- | Of course we need to be able to convert our summaries into 'LogIndex'es:+summaryToIndex :: LogSummary -> LogIndex+summaryToIndex (LogSummary hh (Sum bs) (Min st) (Max et) _) =+    LogIndex heaviestHitter bs st et+  where+    heaviestHitter = fst . maximumBy (comparing snd) . Map.toList $ hh++-- | Now we just need to know when we're done building a block.  Let's assume that 5000 entries per+--   block is a reasonable policy:+blockDone :: LogSummary -> Bool+blockDone ls = lsCount ls == 5000++-- | In order for @arena@ to actually write our data to disk, we need to be able to serialize it.+--   We can just let GHC generics get us the instances we need:+instance Serial LogEntry+instance Serial LogIndex++-- | We can have @arena@ tie all this together for us:+type LogArena a = Arena LogSummary LogIndex LogEntry a++makeLogDB :: ArenaLocation -> IO (ArenaDB LogSummary LogIndex LogEntry)+makeLogDB = startArena summarizeEntry summaryToIndex blockDone++-- | Now we can use @arena@'s 'accessData' function to pull our indexed blocks off disk.  Notice+--   that 'accessData' gives back the slightly different type then above:+--   @[(LogIndex, IO [LogEntry])]@.  This is because @arena@ won't read the actual 'LogEntry's+--   (of which there might be many) unless you specifically force the corresponding IO action.+queryBytesServed :: Time -> Time -> LogArena Int+queryBytesServed start end = do+  blocks <- accessData+  let fullyContained = [ bs | (LogIndex _ bs liStart liEnd, _) <- blocks+                            , liStart >= start && liEnd <= end ]+      partiallyContained = [ bytesServedInPeriod start end <$> es+                              | (LogIndex _ bs liStart liEnd, es) <- blocks+                              , (liStart <= start && start <= liEnd) || (liStart <= end && end <= liEnd)]+  forced <- liftIO . sequence $ partiallyContained+  return $ sum fullyContained + sum forced++-- | Now to use the database, we just add and query data:+addAndQueryData :: LogArena ()+addAndQueryData = do+  mapM_ addData [LogEntry (show $ n `mod` 23) n ((n * n - 1) `mod` 29)+                     | n <- [1..7000]] -- sufficiently random looking data+  46666 <- queryBytesServed 2345 5678+  return ()++-- | Make the database in @/tmp@ and go!+main :: IO ()+main = do+  db <- makeLogDB "/tmp/log_arena"+  runArena db addAndQueryData

+ tests/test.hs view

@@ -0,0 +1,54 @@+{-# LANGUAGE ScopedTypeVariables, TupleSections, OverloadedStrings #-}+module Main where++import System.Mem+import Control.Concurrent++import Control.Monad.Trans+import Data.List+import Data.Semigroup+import System.Directory+import Database.Arena+import qualified Control.Exception as E++assert :: Bool -> String -> IO ()+assert True _ = return ()+assert False err = E.throwIO . E.AssertionFailed $ err++assert' :: MonadIO m => Bool -> String -> m ()+assert' b s = liftIO $ assert b s++prepDirectory :: IO ()+prepDirectory = do+  createDirectoryIfMissing True "test_data"+  removeDirectoryRecursive "test_data"+  createDirectoryIfMissing True "test_data/journal"+  createDirectoryIfMissing True "test_data/data"++main :: IO ()+main = do+  prepDirectory+  tok <- startArena ((1,) . Sum) (getSum . snd) ((5 <) . getSum . fst) "test_data"+  as <- runArena tok monadTest1+  runArena tok $ monadTest2 as++monadTest1 :: Arena (Sum Int, Sum Int) Int Int [(Int, [Int])]+monadTest1 = do+  addData 65+  sm <- (sum . fmap fst) <$> accessData+  assert' (sm == 65) $ "Count was wrong: " ++ show sm+  mapM_ addData [1..14]+  sm <- (sum . fmap fst) <$> accessData+  as <- accessData >>= mapM (\(c, act) -> (c,) <$> liftIO act)+  liftIO $! do putStrLn . show . sort $ as+               assert' (sm == (65 + sum [1..14])) "Larger sum incorrect!"+               performMajorGC+               threadDelay 100000+               performMajorGC+  return as++monadTest2 :: [(Int, [Int])] -> Arena (Sum Int, Sum Int) Int Int ()+monadTest2 as = do+  as' <- accessData >>= mapM (\(c, act) -> (c,) <$> liftIO act)+  liftIO $! do putStrLn . show . sort $ as'+               assert ((sort as) == (sort as')) "doesn't match old data"