diff --git a/hsrc_lib/Database/VCache/Alloc.hs b/hsrc_lib/Database/VCache/Alloc.hs --- a/hsrc_lib/Database/VCache/Alloc.hs +++ b/hsrc_lib/Database/VCache/Alloc.hs @@ -86,18 +86,14 @@ _addr2vref :: (VCacheable a) => a -> VSpace -> Address -> Memory -> IO (Memory, VRef a) _addr2vref _dummy !vc !addr !m = do let ty = typeOf _dummy - mbCacheE <- loadVRefCache addr ty (mem_evrefs m) - case mbCacheE of + mbCache <- loadVRefCache addr ty (mem_vrefs m) + case mbCache of Just cache -> return (m, VRef addr cache vc ty get) Nothing -> do - mbCacheC <- loadVRefCache addr ty (mem_cvrefs m) - case mbCacheC of - Just cache -> return (m, VRef addr cache vc ty get) - Nothing -> do - cache <- newIORef NotCached - e <- mkVREph vc addr cache ty - let m' = m { mem_evrefs = addVREph e (mem_evrefs m) } - m' `seq` return (m', VRef addr cache vc ty get) + cache <- newIORef NotCached + e <- mkVREph vc addr cache ty + let m' = m { mem_vrefs = addVREph e (mem_vrefs m) } + m' `seq` return (m', VRef addr cache vc ty get) {-# NOINLINE _addr2vref #-} mkVREph :: VSpace -> Address -> IORef (Cache a) -> TypeRep -> IO VREph @@ -111,23 +107,24 @@ mbf = Map.lookup addr em >>= Map.lookup ty {-# INLINE loadVRefCache #-} --- When a VRef is GC'd from the Haskell layer, we need to delete it --- from the ephemeron table. Of course, while unlikely, another VRef --- may have since replaced the existing one. +-- When a VRef is GC'd from the Haskell layer, it must be deleted from +-- the ephemeron table. And deleting it from the cache will also help +-- the cache manager maintain a valid estimate of cache size. +-- +-- There is no guarantee that this operation is timely. It may be called +-- after the address is brought back into memory. So this function will +-- double check that it's still working with a 'dead' cache. clearVRef :: VSpace -> Address -> TypeRep -> IO () -clearVRef !vc !addr !ty = modifyMVarMasked_ (vcache_memory vc) $ \ m -> do - evrefs' <- tryDelVREph addr ty (mem_evrefs m) - cvrefs' <- tryDelVREph addr ty (mem_cvrefs m) - let m' = m { mem_evrefs = evrefs', mem_cvrefs = cvrefs' } - return $! m' - -tryDelVREph :: Address -> TypeRep -> VREphMap -> IO VREphMap -tryDelVREph !addr !ty !em = - case takeVREph addr ty em of +clearVRef !vc !addr !ty = delFromCache >> delFromMem where + delFromCache = modifyMVarMasked_ (vcache_cvrefs vc) delFrom + delFromMem = modifyMVarMasked_ (vcache_memory vc) $ \ m -> + delFrom (mem_vrefs m) >>= \ vrefs' -> + return $! m { mem_vrefs = vrefs' } + delFrom em = case takeVREph addr ty em of Nothing -> return em Just (VREph { vreph_cache = wk }, em') -> Weak.deRefWeak wk >>= \ mbc -> - if isJust mbc then return em -- replaced (improbable; race condition) + if isJust mbc then return em -- replaced since GC; do not delete else return em' -- removed -- This is certainly an unsafe operation in general, but we have @@ -237,30 +234,23 @@ (c', c' `seq` op) {-# NOINLINE newVRefIO #-} --- I've split the mem_vrefs into two partitions, evrefs and cvrefs. --- This shifts allows the cache manager to focus on just the cvrefs --- partition, which will typically be much smaller than evrefs. --- --- After a value is first cached, whichever thread was responsible must --- move the content from the mem_evrefs partition into mem_cvrefs. The --- cache manager may later move it back and then clear the cache. +-- Cached values should be represented in the vcache_cvrefs table to +-- support cache management (i.e. so the manager can focus on just +-- the subset of cached values). The cache manager or GC may remove +-- objects from the vcache_cvrefs table. -- --- Cached values may be held temporarily by mem_evrefs, i.e. prior to --- 'init' or just before the cache is cleared. But NotCached VRefs --- should never be held by mem_cvrefs. +-- I have an option here, to either create a new weak IORef for the +-- cache or to reuse the existing one. I'm choosing the latter for +-- now because I'm not sure how much a burden weak references add to +-- the GC. initVRefCache :: VRef a -> IO () -initVRefCache !vref = - let vc = vref_space vref in - let addr = vref_addr vref in - let ty = vref_type vref in - modifyMVarMasked_ (vcache_memory vc) $ \ m -> - case takeVREph addr ty (mem_evrefs m) of - Nothing -> fail $ show vref ++ " expected in mem_evrefs partition!" - Just (e, evrefs') -> - let cvrefs' = addVREph e (mem_cvrefs m) in - let m' = m { mem_evrefs = evrefs', mem_cvrefs = cvrefs' } in - return $! m' -{-# NOINLINE initVRefCache #-} +initVRefCache !r = do + let vc = vref_space r + vrefs <- mem_vrefs <$> readMVar (vcache_memory vc) + case Map.lookup (vref_addr r) vrefs >>= Map.lookup (vref_type r) of + Nothing -> fail $ "VCache bug: " ++ show r ++ " should be in mem_vrefs!" + Just e -> modifyMVarMasked_ (vcache_cvrefs vc) $ return . addVREph e +{-# INLINABLE initVRefCache #-} -- | Construct a new VRef without initializing the cache. newVRefIO' :: (VCacheable a) => VSpace -> a -> IO (VRef a) diff --git a/hsrc_lib/Database/VCache/Cache.hs b/hsrc_lib/Database/VCache/Cache.hs --- a/hsrc_lib/Database/VCache/Cache.hs +++ b/hsrc_lib/Database/VCache/Cache.hs @@ -41,13 +41,8 @@ -- find some use for benchmarks or staged applications. clearVRefsCache :: VSpace -> IO () clearVRefsCache vc = do - -- we must hold lock for long enough to move contents to mem_evrefs - ephMap <- modifyMVarMasked (vcache_memory vc) $ \ m -> do - let evrefs' = Map.unionWith (Map.union) (mem_cvrefs m) (mem_evrefs m) - let m' = m { mem_cvrefs = Map.empty, mem_evrefs = evrefs' } - m' `seq` return (m', mem_cvrefs m) - mapM_ (mapM_ clearVREphCache . Map.elems) (Map.elems ephMap) -{-# NOINLINE clearVRefsCache #-} + cvrefs <- swapMVar (vcache_cvrefs vc) Map.empty + mapM_ (mapM_ clearVREphCache . Map.elems) (Map.elems cvrefs) clearVREphCache :: VREph -> IO () clearVREphCache (VREph { vreph_cache = wc }) = @@ -56,7 +51,6 @@ Nothing -> return () Just cache -> writeIORef cache NotCached - -- | Immediately clear the cache associated with a VRef, allowing -- any contained data to be GC'd. Normally, VRef cached values are -- cleared either by a background thread or when the VRef itself @@ -65,13 +59,10 @@ clearVRefCache :: VRef a -> IO () clearVRefCache v = do let vc = vref_space v - modifyMVarMasked_ (vcache_memory vc) $ \ m -> do - case takeVREph (vref_addr v) (vref_type v) (mem_cvrefs m) of - Nothing -> return m -- was not cached - Just (e, cvrefs') -> do - let evrefs' = addVREph e (mem_evrefs m) - let m' = m { mem_cvrefs = cvrefs', mem_evrefs = evrefs' } - return $! m' + modifyMVarMasked_ (vcache_cvrefs vc) $ \ cvrefs -> do + case takeVREph (vref_addr v) (vref_type v) cvrefs of + Nothing -> return cvrefs -- was not cached + Just ( _ , cvrefs') -> return cvrefs' writeIORef (vref_cache v) NotCached {-# NOINLINE clearVRefCache #-} diff --git a/hsrc_lib/Database/VCache/Clean.hs b/hsrc_lib/Database/VCache/Clean.hs --- a/hsrc_lib/Database/VCache/Clean.hs +++ b/hsrc_lib/Database/VCache/Clean.hs @@ -36,6 +36,7 @@ import Data.Bits import qualified Data.Traversable as TR import qualified Data.Map.Strict as Map +import qualified Data.List as L import Data.IORef import qualified System.Mem.Weak as Weak import qualified System.Random as Random @@ -46,25 +47,25 @@ -- | Cache cleanup, and signal writer for old content. cleanStep :: VSpace -> IO () cleanStep vc = do - bsig <- shouldSignalWriter vc - when bsig (signalWriter vc) - wtgt <- readIORef (vcache_climit vc) w0 <- estCacheSize vc let hitRate = if ((100 * w0) < ( 80 * wtgt)) then 0.00 else if ((100 * w0) < (100 * wtgt)) then 0.01 else - if ((100 * w0) < (130 * wtgt)) then 0.02 else - if ((100 * w0) < (170 * wtgt)) then 0.03 else - if ((100 * w0) < (220 * wtgt)) then 0.04 else - if ((100 * w0) < (280 * wtgt)) then 0.05 else + if ((100 * w0) < (120 * wtgt)) then 0.02 else + if ((100 * w0) < (150 * wtgt)) then 0.03 else + if ((100 * w0) < (190 * wtgt)) then 0.04 else + if ((100 * w0) < (240 * wtgt)) then 0.05 else 0.06 xcln vc hitRate - updateCacheSizeEst vc + updateCacheSizeEst vc 10 0.01 wf <- estCacheSize vc + bsig <- shouldSignalWriter vc + when bsig (signalWriter vc) + let bSatisfied = (max w0 wf) < wtgt - let dtSleep = if bSatisfied then 295000 else 95000 + let dtSleep = if bSatisfied then 270000 else 135000 usleep dtSleep -- ~10Hz, slower when steady -- sleep for a number of microseconds @@ -86,46 +87,33 @@ readCacheAddrCt :: VSpace -> IO Int readCacheAddrCt vc = do - m <- readMVar (vcache_memory vc) - return $! Map.size (mem_cvrefs m) + cvrefs <- readMVar (vcache_cvrefs vc) + return $! Map.size cvrefs --- sample the cache at a few random addresses, use this to update the --- cache size by a small factor. Over the course of many seconds, the --- estimated average size per address should approach the actual size --- assuming the average itself is stable. Even if average size isn't --- stable, this is good enough to help guide the cache manager. --- --- The assumption here is that the cvrefs map is usually large. If it --- is small, we'll still use the same algorithm, even if it's a bit --- redundant, to simplify reasoning and testing. A constant number of --- samples are taken in each round. Probabilistically -updateCacheSizeEst :: VSpace -> IO () -updateCacheSizeEst vc = - readMVar (vcache_memory vc) >>= \ m -> - let cvrefs = mem_cvrefs m in +-- sample the cache at random addresses, and update using an +-- exponential running average. +updateCacheSizeEst :: VSpace -> Int -> Double -> IO () +updateCacheSizeEst vc !n !alpha = + readMVar (vcache_cvrefs vc) >>= \ cvrefs -> if Map.null cvrefs then return () else - let nextIx = Random.randomR (0, Map.size cvrefs - 1) in - let loop !n !r !sz !sqsz = - if (0 == n) then return (sz,sqsz) else - let (ix,r') = nextIx r in - let (_, tym) = Map.elemAt ix cvrefs in - let (_, e) = Map.findMin tym in -- safe; address elements non-empty + Random.newStdGen >>= \ rgen -> + let ixs = L.take n $ Random.randomRs (0, Map.size cvrefs - 1) rgen in + let readAddrSize ix = + let (_addr, tym) = Map.elemAt ix cvrefs in + let (_ty, e) = Map.findMin tym in readVREphSize e >>= \ esz -> - let addrsz = fromIntegral $ esz * Map.size tym in - let sz' = sz + addrsz in - let sqsz' = sqsz + (addrsz * addrsz) in - loop (n-1) r' sz' sqsz' + return (esz * fromIntegral (Map.size tym)) in - let nSamples = 15 :: Int in - Random.newStdGen >>= \ r -> - loop nSamples r 0 0 >>= \ (totalSize, totalSqSize) -> - let sampleAvg = totalSize / fromIntegral nSamples in - let sampleAvgSq = totalSqSize / fromIntegral nSamples in - readIORef (vcache_csize vc) >>= \ (CacheSizeEst oldAvg oldAvgSq) -> - let alpha = 0.015 :: Double in - let newAvg = alpha * sampleAvg + ((1.0 - alpha) * oldAvg) in - let newAvgSq = alpha * sampleAvgSq + ((1.0 - alpha) * oldAvgSq) in - writeIORef (vcache_csize vc) $! (CacheSizeEst newAvg newAvgSq) + mapM readAddrSize ixs >>= \ lSizes -> + let szTotal = L.foldl' (+) 0 lSizes in + let sqszTotal = L.foldl' (\ ssq x -> ssq + (x*x)) 0 lSizes in + let szAvgSamp = fromIntegral (szTotal `div` n) in + let sqszAvgSamp = fromIntegral (sqszTotal `div` n) in + readIORef (vcache_csize vc) >>= \ (CacheSizeEst szAvgEst sqszAvgEst) -> + let upd new old = (alpha * new) + ((1.0 - alpha) * old) in + let szAvg' = upd szAvgSamp szAvgEst in + let sqszAvg' = upd sqszAvgSamp sqszAvgEst in + writeIORef (vcache_csize vc) $! (CacheSizeEst szAvg' sqszAvg') readVREphSize :: VREph -> IO Int readVREphSize (VREph { vreph_cache = wk }) = @@ -133,7 +121,7 @@ Nothing -> return 2048 -- GC'd recently; high estimate Just cache -> readIORef cache >>= \ c -> case c of NotCached -> - let eMsg = "VCache bug: NotCached element found in mem_cvrefs" in + let eMsg = "VCache bug: NotCached element found in vcache_cvrefs" in fail eMsg Cached _ bf -> let lgSz = 6 + fromIntegral (0x1f .&. bf) in @@ -158,33 +146,27 @@ xclnStrike vc r >>= xclnLoop vc (n-1) xclnStrike :: VSpace -> Random.StdGen -> IO Random.StdGen -xclnStrike !vc !r = modifyMVarMasked (vcache_memory vc) $ \ m -> - if Map.null (mem_cvrefs m) then return (m,r) else do - let cvrefs = mem_cvrefs m - let evrefs = mem_evrefs m +xclnStrike !vc !r = modifyMVarMasked (vcache_cvrefs vc) $ \ cvrefs -> + if Map.null cvrefs then return (cvrefs, r) else do let (ix,r') = Random.randomR (0, Map.size cvrefs - 1) r let (addr, tym) = Map.elemAt ix cvrefs - (tymc, tyme) <- Map.mapEither id <$> TR.traverse strikeVREph tym - let cvrefs' = if Map.null tymc then Map.delete addr cvrefs else - if Map.null tyme then cvrefs else - Map.insert addr tymc cvrefs - let evrefs' = if Map.null tyme then evrefs else - Map.insertWith (Map.union) addr tyme evrefs - let m' = m { mem_cvrefs = cvrefs', mem_evrefs = evrefs' } - return (m', m' `seq` r') + tym' <- Map.mapMaybe id <$> TR.traverse strikeVREph tym + let cvrefs' = if Map.null tym' then Map.delete addr cvrefs + else Map.insert addr tym' cvrefs + return (cvrefs', r') -- strikeVREph will reduce the CacheMode for a cached element or --- remove it from the cache (in right) for CacheMode0. -strikeVREph :: VREph -> IO (Either VREph VREph) +-- remove it from the cache for CacheMode0. +strikeVREph :: VREph -> IO (Maybe VREph) strikeVREph vreph@(VREph { vreph_cache = wk }) = Weak.deRefWeak wk >>= \ mbCache -> case mbCache of - Nothing -> return (Right vreph) -- + Nothing -> return Nothing Just cache -> atomicModifyIORef cache $ \ c -> case c of Cached r bf | (0 /= bf .&. 0x60) -> let bf' = (0x80 .|. (bf - 0x20)) in let c' = Cached r bf' in - (c', c' `seq` (Left vreph)) - _ -> (NotCached, Right vreph) + (c', c' `seq` (Just vreph)) + _ -> (NotCached, Nothing) -- If the writer has obvious work it could be doing, signal it. This -- won't significantly affect a busy writer, but an idle writer may @@ -196,7 +178,7 @@ let bHoldingAllocs = not (emptyAllocation (mem_alloc m)) in if bHoldingAllocs then return True else readZeroesCt vc >>= \ ctZeroes -> - let ctEphAddrs = Map.size (mem_cvrefs m) + Map.size (mem_evrefs m) + Map.size (mem_pvars m) in + let ctEphAddrs = Map.size (mem_vrefs m) + Map.size (mem_pvars m) in if (ctEphAddrs < ctZeroes) then return True else return False diff --git a/hsrc_lib/Database/VCache/Open.hs b/hsrc_lib/Database/VCache/Open.hs --- a/hsrc_lib/Database/VCache/Open.hs +++ b/hsrc_lib/Database/VCache/Open.hs @@ -142,6 +142,7 @@ mvSignal <- newMVar () cLimit <- newIORef vcDefaultCacheLimit cSize <- newIORef vcInitCacheSizeEst + cVRefs <- newMVar Map.empty ctWrites <- newIORef $ WriteCt 0 0 0 gcStart <- newIORef Nothing gcCount <- newIORef 0 @@ -168,6 +169,7 @@ , vcache_rwlock = rwLock , vcache_climit = cLimit , vcache_csize = cSize + , vcache_cvrefs = cVRefs , vcache_signal_writes = updWriteCt ctWrites , vcache_ct_writes = ctWrites , vcache_alloc_init = allocStart @@ -204,7 +206,7 @@ ac = Allocator addr af af af gcf = GCFrame Map.empty gc = GC gcf gcf - m0 = Memory Map.empty Map.empty Map.empty gc ac + m0 = Memory Map.empty Map.empty gc ac -- Update write counts. updWriteCt :: IORef WriteCt -> Writes -> IO () diff --git a/hsrc_lib/Database/VCache/Stats.hs b/hsrc_lib/Database/VCache/Stats.hs --- a/hsrc_lib/Database/VCache/Stats.hs +++ b/hsrc_lib/Database/VCache/Stats.hs @@ -20,14 +20,12 @@ , vcstat_vref_count :: {-# UNPACK #-} !Int -- ^ number of immutable values in the database , vcstat_pvar_count :: {-# UNPACK #-} !Int -- ^ number of mutable PVars in the database , vcstat_root_count :: {-# UNPACK #-} !Int -- ^ number of named roots (a subset of PVars) - , vcstat_mem_vrefs :: {-# UNPACK #-} !Int -- ^ number of VRefs in Haskell process memory (some may share address) + , vcstat_mem_vrefs :: {-# UNPACK #-} !Int -- ^ number of VRefs in Haskell process memory , vcstat_mem_pvars :: {-# UNPACK #-} !Int -- ^ number of PVars in Haskell process memory - , vcstat_mem_addrs :: {-# UNPACK #-} !Int -- ^ number of addresses held by Haskell process memory , vcstat_eph_count :: {-# UNPACK #-} !Int -- ^ number of addresses with zero references , vcstat_alloc_pos :: {-# UNPACK #-} !Address -- ^ address to next be used by allocator , vcstat_alloc_count :: {-# UNPACK #-} !Int -- ^ number of allocations by this process - , vcstat_cache_count :: {-# UNPACK #-} !Int -- ^ number of VRefs with cached values - , vcstat_cache_limit :: {-# UNPACK #-} !Int -- ^ target cache size + , vcstat_cache_limit :: {-# UNPACK #-} !Int -- ^ target cache size in bytes , vcstat_cache_size :: {-# UNPACK #-} !Int -- ^ estimated cache size in bytes , vcstat_gc_count :: {-# UNPACK #-} !Int -- ^ number of addresses GC'd by this process , vcstat_write_pvars :: {-# UNPACK #-} !Int -- ^ number of PVar updates to disk (after batching) @@ -52,6 +50,7 @@ wct <- readIORef (vcache_ct_writes vc) cLimit <- readIORef (vcache_climit vc) cSizeEst <- readIORef (vcache_csize vc) + cvrefs <- readMVar (vcache_cvrefs vc) let fileSize = (1 + (fromIntegral $ me_last_pgno envInfo)) * (fromIntegral $ ms_psize envStat) @@ -59,15 +58,10 @@ let pvarCount = (fromIntegral $ ms_entries dbMemStat) - vrefCount let ephCount = (fromIntegral $ ms_entries ephStat) let rootCount = (fromIntegral $ ms_entries rootStat) - let cvrefsCount = Map.foldl' (\ a b -> a + Map.size b) 0 (mem_cvrefs memory) - let evrefsCount = Map.foldl' (\ a b -> a + Map.size b) 0 (mem_evrefs memory) - let cacheSizeBytes = ceiling $ fromIntegral (Map.size (mem_cvrefs memory)) - * csze_addr_size cSizeEst - let memVRefsCount = cvrefsCount + evrefsCount + let cacheSizeBytes = ceiling $ fromIntegral (Map.size cvrefs) + * sqrt (csze_addr_sqsz cSizeEst) + let memVRefsCount = Map.foldl' (\ a b -> a + Map.size b) 0 (mem_vrefs memory) let memPVarsCount = Map.size (mem_pvars memory) - let memAddrsCount = Map.size (mem_pvars memory) - + Map.size (mem_cvrefs memory) - + Map.size (mem_evrefs memory) let allocPos = alloc_new_addr (mem_alloc memory) let allocDiff = allocPos - vcache_alloc_init vc let allocCount = fromIntegral $ allocDiff `div` 2 @@ -78,11 +72,9 @@ , vcstat_root_count = rootCount , vcstat_mem_vrefs = memVRefsCount , vcstat_mem_pvars = memPVarsCount - , vcstat_mem_addrs = memAddrsCount , vcstat_eph_count = ephCount , vcstat_alloc_pos = allocPos , vcstat_alloc_count = allocCount - , vcstat_cache_count = cvrefsCount , vcstat_cache_limit = cLimit , vcstat_cache_size = cacheSizeBytes , vcstat_write_sync = wct_sync wct diff --git a/hsrc_lib/Database/VCache/Types.hs b/hsrc_lib/Database/VCache/Types.hs --- a/hsrc_lib/Database/VCache/Types.hs +++ b/hsrc_lib/Database/VCache/Types.hs @@ -329,6 +329,7 @@ , vcache_climit :: !(IORef Int) -- targeted max cache size in bytes , vcache_csize :: !(IORef CacheSizeEst) -- estimated cache sizes + , vcache_cvrefs :: !(MVar VREphMap) -- track just the cached VRefs -- share persistent variables for safe STM @@ -421,13 +422,11 @@ -- operations on them are atomic... and STM isn't permitted -- because vref constructors are used with unsafePerformIO. data Memory = Memory - { mem_evrefs :: !VREphMap -- ^ VRefs with empty cache. - , mem_cvrefs :: !VREphMap -- ^ VRefs with full cache. + { mem_vrefs :: !VREphMap -- ^ In-memory VRefs , mem_pvars :: !PVEphMap -- ^ In-memory PVars , mem_gc :: !GC -- ^ recently GC'd addresses (two frames) , mem_alloc :: !Allocator -- ^ recent or pending allocations (three frames) } - -- simple read-only operations diff --git a/hsrc_lib/Database/VCache/Write.hs b/hsrc_lib/Database/VCache/Write.hs --- a/hsrc_lib/Database/VCache/Write.hs +++ b/hsrc_lib/Database/VCache/Write.hs @@ -443,9 +443,8 @@ gcSelectFrame :: VSpace -> GCBatch -> IO GCBatch gcSelectFrame vc gcb = modifyMVarMasked (vcache_memory vc) $ \ m -> do - let gcb' = (((gcb `Map.difference` mem_evrefs m) - `Map.difference` mem_cvrefs m) - `Map.difference` mem_pvars m) + let gcb' = ((gcb `Map.difference` mem_vrefs m) + `Map.difference` mem_pvars m) let gc' = GC { gc_frm_curr = GCFrame gcb' , gc_frm_prev = gc_frm_curr (mem_gc m) } let m' = m { mem_gc = gc' } diff --git a/vcache.cabal b/vcache.cabal --- a/vcache.cabal +++ b/vcache.cabal @@ -1,5 +1,5 @@ Name: vcache -Version: 0.1 +Version: 0.1.1 Synopsis: large, persistent, memcached values and structure sharing for Haskell Category: Database Description: