haxl 2.1.2.0 → 2.3.0.0

raw patch · 40 files changed

+2267/−881 lines, 40 filesdep +hashtablesdep +optparse-applicativedep −semigroupsdep ~aesondep ~basedep ~hashablePVP ok

version bump matches the API change (PVP)

Dependencies added: hashtables, optparse-applicative

Dependencies removed: semigroups

Dependency ranges changed: aeson, base, hashable, time

API changes (from Hackage documentation)

Files

• Haxl/Core.hs +14/−7
• Haxl/Core/DataCache.hs +114/−61
• Haxl/Core/DataSource.hs +179/−23
• Haxl/Core/Exception.hs +2/−6
• Haxl/Core/Fetch.hs +208/−127
• Haxl/Core/Flags.hs +3/−13
• Haxl/Core/Memo.hs +74/−37
• Haxl/Core/Monad.hs +251/−100
• Haxl/Core/Parallel.hs +104/−37
• Haxl/Core/Profile.hs +117/−43
• Haxl/Core/RequestStore.hs +26/−10
• Haxl/Core/Run.hs +78/−34
• Haxl/Core/StateStore.hs +0/−4
• Haxl/Core/Stats.hs +125/−50
• Haxl/Core/Util.hs +19/−2
• Haxl/DataSource/ConcurrentIO.hs +1/−3
• Haxl/Prelude.hs +20/−7
• changelog.md +14/−1
• haxl.cabal +20/−24
• readme.md +8/−0
• tests/AllTests.hs +7/−9
• tests/BadDataSource.hs +53/−16
• tests/BatchTests.hs +2/−0
• tests/Bench.hs +12/−8
• tests/DataCacheTest.hs +8/−12
• tests/ExampleDataSource.hs +8/−0
• tests/FullyAsyncTest.hs +0/−3
• tests/MemoizationTests.hs +0/−4
• tests/MockTAO.hs +14/−7
• tests/MonadAsyncTest.hs +128/−0
• tests/MonadBench.hs +100/−79
• tests/OutgoneFetchesTests.hs +38/−94
• tests/ParallelTests.hs +72/−0
• tests/ProfileTests.hs +108/−19
• tests/StatsTests.hs +191/−0
• tests/TestBadDataSource.hs +50/−14
• tests/TestExampleDataSource.hs +1/−1
• tests/TestUtils.hs +3/−1
• tests/WorkDataSource.hs +54/−13
• tests/WriteTests.hs +41/−12

@@ -17,19 +17,19 @@     -- *** Operations in the monad   , env, withEnv, withLabel     -- *** Building the Env-  , initEnvWithData, initEnv, emptyEnv+  , initEnvWithData, initEnv, emptyEnv, sanitizeEnv     -- *** Building the StateStore   , StateStore, stateGet, stateSet, stateEmpty      -- ** Writes inside the monad-  , tellWrite+  , tellWrite, tellWriteNoMemo      -- ** Exceptions   , throw, catch, catchIf, try, tryToHaxlException      -- ** Data fetching and caching   , dataFetch, uncachedRequest-  , cacheRequest, cacheResult, cacheResultWithShow+  , cacheRequest, dupableCacheRequest, cacheResult, cacheResultWithShow   , cachedComputation, preCacheComputation   , dumpCacheAsHaskell @@ -39,25 +39,29 @@   , memoFingerprint, MemoFingerprintKey(..)      -- ** Conditionals-  , pAnd, pOr+  , pAnd, pOr, unsafeChooseFirst      -- ** Statistics   , Stats(..)   , FetchStats(..)+  , CallId   , Microseconds   , Timestamp   , emptyStats   , numFetches   , ppStats   , ppFetchStats-  , Profile+  , aggregateFetchBatches+  , Profile(..)+  , ProfileMemo(..)+  , ProfileFetch(..)   , emptyProfile-  , profile   , ProfileLabel+  , ProfileKey   , ProfileData(..)   , emptyProfileData   , AllocCount-  , MemoHitCount+  , LabelHitCount      -- ** Tracing flags   , Flags(..)@@ -75,6 +79,7 @@   , PerformFetch(..)   , StateKey(..)   , SchedulerHint(..)+  , FailureClassification(..)      -- ** Result variables   , ResultVar(..)@@ -86,6 +91,8 @@      -- ** Default fetch implementations   , asyncFetch, asyncFetchWithDispatch, asyncFetchAcquireRelease+  , backgroundFetchSeq, backgroundFetchPar+  , backgroundFetchAcquireRelease, backgroundFetchAcquireReleaseMVar   , stubFetch   , syncFetch 

@@ -4,11 +4,10 @@ -- This source code is distributed under the terms of a BSD license, -- found in the LICENSE file. -{-# LANGUAGE CPP #-} {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-}-{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# LANGUAGE TypeFamilies #-}  -- | -- A cache mapping data requests to their results.  This module is@@ -19,24 +18,21 @@   ( DataCache(..)   , SubCache(..)   , emptyDataCache+  , filter   , insert   , insertNotShowable   , insertWithShow   , lookup   , showCache+  , readCache   ) where +import Prelude hiding (lookup, filter) import Control.Exception-import Data.Hashable-import Prelude hiding (lookup) import Unsafe.Coerce-import Data.HashMap.Strict (HashMap)-import qualified Data.HashMap.Strict as HashMap import Data.Typeable-import Data.Maybe-#if __GLASGOW_HASKELL__ < 710-import Control.Applicative ((<$>))-#endif+import Data.Hashable+import qualified Data.HashTable.IO as H  -- --------------------------------------------------------------------------- -- DataCache@@ -45,7 +41,7 @@ -- any @req@ and @a@ provided @req a@ is an instance of 'Typeable'. In -- practice @req a@ will be a request type parameterised by its result. ---newtype DataCache res = DataCache (HashMap TypeRep (SubCache res))+newtype DataCache res = DataCache (HashTable TypeRep (SubCache res))  -- | The implementation is a two-level map: the outer level maps the -- types of requests to 'SubCache', which maps actual requests to their@@ -55,14 +51,15 @@ -- data SubCache res =   forall req a . (Hashable (req a), Eq (req a), Typeable (req a)) =>-       SubCache (req a -> String) (a -> String) ! (HashMap (req a) (res a))+       SubCache (req a -> String) (a -> String) ! (HashTable (req a) (res a))        -- NB. the inner HashMap is strict, to avoid building up        -- a chain of thunks during repeated insertions. --- | A new, empty 'DataCache'.-emptyDataCache :: DataCache res-emptyDataCache = DataCache HashMap.empty+type HashTable k v = H.BasicHashTable k v +-- | A new, empty 'DataCache'.+emptyDataCache :: IO (DataCache res)+emptyDataCache = DataCache <$> H.new  -- | Inserts a request-result pair into the 'DataCache'. insert@@ -72,10 +69,24 @@   -> res a   -- ^ Result   -> DataCache res-  -> DataCache res+  -> IO ()  insert = insertWithShow show show +-- | Inserts a request-result pair into the 'DataCache', without+-- requiring Show instances of the request or the result.  The cache+-- cannot be subsequently used with `showCache`.+insertNotShowable+  :: (Hashable (req a), Typeable (req a), Eq (req a))+  => req a+  -- ^ Request+  -> res a+  -- ^ Result+  -> DataCache res+  -> IO ()++insertNotShowable = insertWithShow notShowable notShowable+ -- | Inserts a request-result pair into the 'DataCache', with the given -- functions used to show the request and result. insertWithShow@@ -89,71 +100,113 @@   -> res a   -- ^ Result   -> DataCache res-  -> DataCache res--insertWithShow showRequest showResult req result (DataCache m) =-  DataCache $-    HashMap.insertWith fn (typeOf req)-       (SubCache showRequest showResult (HashMap.singleton req result)) m-  where-    fn (SubCache _ _ new) (SubCache showReq showRes old) =-      SubCache showReq showRes (unsafeCoerce new `HashMap.union` old)---- | Inserts a request-result pair into the 'DataCache', without--- requiring Show instances of the request or the result.  The cache--- cannot be subsequently used with `showCache`.-insertNotShowable-  :: (Hashable (req a), Typeable (req a), Eq (req a))-  => req a-  -- ^ Request-  -> res a-  -- ^ Result-  -> DataCache res-  -> DataCache res+  -> IO () -insertNotShowable = insertWithShow notShowable notShowable+insertWithShow showRequest showResult request result (DataCache m) =+  H.mutateIO m (typeOf request) (mutate showRequest showResult request result)  notShowable :: a notShowable = error "insertNotShowable" +-- | A mutation function for mutateIO. If the key doesn't exist in the top-level+-- cache, creates a new hashtable and inserts the request and result.+-- If the key exists, insert the request and result into the existing subcache,+-- replacing any existing mapping.+mutate :: (Hashable (req a), Typeable (req a), Eq (req a))+  => (req a -> String)+  -> (a -> String)+  -> req a+  -> res a+  -> Maybe (SubCache res)+  -> IO (Maybe (SubCache res), ())+mutate showRequest showResult request result Nothing = do+  newTable <- H.new+  H.insert newTable request result+  return (Just (SubCache showRequest showResult newTable), ())+mutate _ _ request result (Just sc@(SubCache _ _ oldTable)) = do+    H.insert oldTable (unsafeCoerce request) (unsafeCoerce result)+    return (Just sc, ())+ -- | Looks up the cached result of a request. lookup   :: Typeable (req a)   => req a   -- ^ Request   -> DataCache res-  -> Maybe (res a)+  -> IO (Maybe (res a)) -lookup req (DataCache m) =-      case HashMap.lookup (typeOf req) m of-        Nothing -> Nothing-        Just (SubCache _ _ sc) ->-           unsafeCoerce (HashMap.lookup (unsafeCoerce req) sc)+lookup req (DataCache m) = do+  mbRes <- H.lookup m (typeOf req)+  case mbRes of+    Nothing -> return Nothing+    Just (SubCache _ _ sc) ->+      unsafeCoerce (H.lookup sc (unsafeCoerce req)) +filter+  :: forall res+  . (forall a. res a -> IO Bool)+  -> DataCache res+  -> IO (DataCache res)+filter pred (DataCache cache) = do+  cacheList <- H.toList cache+  filteredCache <- filterSubCache `mapM` cacheList+  DataCache <$> H.fromList filteredCache+  where+    filterSubCache+      :: (TypeRep, SubCache res)+      -> IO (TypeRep, SubCache res)+    filterSubCache (ty, SubCache showReq showRes hm) = do+      filteredList <- H.foldM go [] hm+      filteredSC <- H.fromList filteredList+      return (ty, SubCache showReq showRes filteredSC)+      where+        go res (request, rvar) = do+          predRes <- pred rvar+          return $ if predRes then (request, rvar):res else res+ -- | Dumps the contents of the cache, with requests and responses -- converted to 'String's using the supplied show functions.  The -- entries are grouped by 'TypeRep'.  Note that this will fail if -- 'insertNotShowable' has been used to insert any entries.--- showCache   :: forall res   .  DataCache res   -> (forall a . res a -> IO (Maybe (Either SomeException a)))   -> IO [(TypeRep, [(String, Either SomeException String)])]+showCache (DataCache cache) readRes = H.foldM goSubCache [] cache+  where+    goSubCache+      :: [(TypeRep, [(String, Either SomeException String)])]+      -> (TypeRep, SubCache res)+      -> IO [(TypeRep, [(String, Either SomeException String)])]+    goSubCache res (ty, SubCache showReq showRes hm) = do+      subCacheResult <- H.foldM go [] hm+      return $ (ty, subCacheResult):res+      where+        go res (request, rvar) = do+          maybe_r <- readRes rvar+          return $ case maybe_r of+            Nothing -> res+            Just (Left e) -> (showReq request, Left e) : res+            Just (Right result) ->+              (showReq request, Right (showRes result)) : res -showCache (DataCache cache) readRes = mapM goSubCache (HashMap.toList cache)- where-  goSubCache-    :: (TypeRep, SubCache res)-    -> IO (TypeRep,[(String, Either SomeException String)])-  goSubCache (ty, SubCache showReq showRes hmap) = do-    elems <- catMaybes <$> mapM go (HashMap.toList hmap)-    return (ty, elems)-   where-    go  (req, rvar) = do-      maybe_r <- readRes rvar-      case maybe_r of-        Nothing -> return Nothing-        Just (Left e) -> return (Just (showReq req, Left e))-        Just (Right result) ->-          return (Just (showReq req, Right (showRes result)))+-- | Dumps the contents of the cache responses to list+readCache+  :: forall res ret+  .  DataCache res+  -> (forall a . res a -> IO ret)+  -> IO [(TypeRep, [Either SomeException ret])]+readCache (DataCache cache) readRes = H.foldM goSubCache [] cache+  where+    goSubCache+      :: [(TypeRep, [Either SomeException ret])]+      -> (TypeRep, SubCache res)+      -> IO [(TypeRep, [Either SomeException ret])]+    goSubCache res (ty, SubCache _showReq _showRes hm) = do+      subCacheResult <- H.foldM go [] hm+      return $ (ty, subCacheResult):res+      where+        go res (_request, rvar) = do+          r <- try $ readRes rvar+          return $ r : res

@@ -4,7 +4,6 @@ -- This source code is distributed under the terms of a BSD license, -- found in the LICENSE file. -{-# LANGUAGE CPP #-} {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE GADTs #-}@@ -28,6 +27,7 @@   , BlockedFetch(..)   , PerformFetch(..)   , SchedulerHint(..)+  , FailureClassification(..)    -- * Result variables   , ResultVar(..)@@ -40,6 +40,9 @@   -- * Default fetch implementations   , asyncFetch, asyncFetchWithDispatch   , asyncFetchAcquireRelease+  , backgroundFetchSeq, backgroundFetchPar+  , backgroundFetchAcquireRelease+  , backgroundFetchAcquireReleaseMVar   , stubFetch   , syncFetch @@ -48,17 +51,11 @@   , setError   ) where -#if __GLASGOW_HASKELL__ < 710-import Control.Applicative-#endif import Control.Exception+import Control.Monad import Data.Hashable import Data.Text (Text)-#if __GLASGOW_HASKELL__ >= 802 import Data.Typeable-#else-import Data.Typeable.Internal-#endif  import Haxl.Core.Exception import Haxl.Core.Flags@@ -66,6 +63,15 @@ import Haxl.Core.StateStore  +import GHC.Conc ( newStablePtrPrimMVar+                , PrimMVar)+import Control.Concurrent ( threadCapability+                          , forkOn+                          , myThreadId )+import Control.Concurrent.MVar+import Foreign.StablePtr+import System.Mem (setAllocationCounter)+ -- --------------------------------------------------------------------------- -- DataSource class @@ -98,6 +104,9 @@   schedulerHint :: u -> SchedulerHint req   schedulerHint _ = TryToBatch +  classifyFailure :: u -> req a -> SomeException -> FailureClassification+  classifyFailure _ _ _ = StandardFailure+ class DataSourceName (req :: * -> *) where   -- | The name of this 'DataSource', used in tracing and stats. Must   -- take a dummy request.@@ -129,6 +138,11 @@     -- returns BackgroundFetch, otherwise requests to this data source will     -- be performed synchronously, one at a time. +-- | Hints to the stats module about how to deal with these failures+data FailureClassification+  = StandardFailure+  | IgnoredForStatsFailure+ -- | A data source can fetch data in one of four ways. -- data PerformFetch req@@ -143,20 +157,6 @@     -- ^ Fetches the data in the background, calling 'putResult' at     -- any time in the future.  This is the best kind of fetch,     -- because it provides the most concurrency.-  | FutureFetch ([BlockedFetch req] -> IO (IO ()))-    -- ^ Returns an IO action that, when performed, waits for the data-    -- to be received.  This is the second-best type of fetch, because-    -- the scheduler still has to perform the blocking wait at some-    -- point in the future, and when it has multiple blocking waits to-    -- perform, it can't know which one will return first.-    ---    -- Why not just forkIO the IO action to make a FutureFetch into a-    -- BackgroundFetch?  The blocking wait will probably do a safe FFI-    -- call, which means it needs its own OS thread.  If we don't want-    -- to create an arbitrary number of OS threads, then FutureFetch-    -- enables all the blocking waits to be done on a single thread.-    -- Also, you might have a data source that requires all calls to-    -- be made in the same OS thread.   -- | A 'BlockedFetch' is a pair of@@ -198,7 +198,7 @@ putSuccess r = putResult r . Right  putResult :: ResultVar a -> Either SomeException a -> IO ()-putResult (ResultVar io) res =  io res False+putResult (ResultVar io) res = io res False  -- | Like `putResult`, but used to get correct accounting when work is -- being done in child threads.  This is particularly important for@@ -311,7 +311,46 @@   dispatch service   sequence_ getResults +backgroundFetchSeq, backgroundFetchPar+  :: (forall a. request a -> IO (Either SomeException a))+  -- ^ Run one request, will be run in a background thread +  -> State request+  -- ^ Currently unused.++  -> Flags+  -- ^ Currently unused.++  -> u+  -- ^ Currently unused.++  -> PerformFetch request++backgroundFetchSeq run _state _flags _si =+  BackgroundFetch $ \requests -> do+    (cap, _) <- threadCapability =<< myThreadId+    mask $ \restore -> void $ forkOn cap $ do+      let rethrow = rethrowFromBg requests+      restore (mapM_ runOne requests) `catch` rethrow+      where+        runOne (BlockedFetch request result) = do+          res <- run request+          putResultFromBg result res++backgroundFetchPar run _state _flags _si =+  BackgroundFetch $ \requests -> do+    (cap, _) <- threadCapability =<< myThreadId+    mapM_ (runOneInThread cap) requests+  where+    runOneInThread cap request = do+      mask $ \restore -> void $ forkOn cap $ do+        let rethrow = rethrowFromBg [request]+        restore (runOne request) `catch` rethrow+    runOne (BlockedFetch request result) = do+      res <- run request+      putResultFromBg result res++ {- | A version of 'asyncFetch' (actually 'asyncFetchWithDispatch') that handles exceptions correctly.  You should use this instead of@@ -388,3 +427,120 @@   -> IO (IO ()) submitFetch service fetchFn (BlockedFetch request result)   = (putResult result =<<) <$> fetchFn service request++putResultFromBg :: ResultVar a -> Either SomeException a -> IO ()+putResultFromBg result r = do+  -- See comment on putResultFromChildThread+  -- We must set the allocation counter to 0 here in case there are more+  -- results in the batch.+  -- This is safe as we own this thread, and know that there+  -- is no allocation limits set.+  putResultFromChildThread result r+  setAllocationCounter 0++rethrowFromBg :: [BlockedFetch req] -> SomeException -> IO ()+rethrowFromBg requests e = do+  mapM_ (rethrow1bg e) requests+  rethrowAsyncExceptions e+  where+    rethrow1bg e (BlockedFetch _ result) =+      putResultFromBg result (Left e)+++backgroundFetchAcquireReleaseMVar+  :: IO service+  -- ^ Resource acquisition for this datasource++  -> (service -> IO ())+  -- ^ Resource release++  -> (service -> Int -> MVar () -> IO ())+  -- ^ Dispatch all the pending requests and when ready trigger the given mvar++  -> (service -> IO ())+  -- ^ Process all requests++  -> (forall a. service -> request a -> IO (IO (Either SomeException a)))+  -- ^ Submits an individual request to the service.++  -> State request+  -- ^ Currently unused.++  -> Flags+  -- ^ Currently unused.++  -> u+  -- ^ Currently unused.++  -> PerformFetch request++backgroundFetchAcquireReleaseMVar+  acquire release dispatch process enqueue _state _flags _si =+  BackgroundFetch $ \requests -> do+    mvar <- newEmptyMVar+    mask $ \restore -> do+      (cap, _) <- threadCapability =<< myThreadId+      service <- acquire+      getResults <- (do+        results <- restore $ mapM (submit service) requests+        -- dispatch takes ownership of mvar, so we call it under `mask` to+        -- ensure that it can safely manage that resource.+        dispatch service cap mvar+        return (sequence_ results)) `onException` release service+      -- now spawn off a background thread to wait on the dispatch to finish+      _tid <- forkOn cap $ do+        takeMVar mvar+          -- todo: it is possible that we would want to do+          -- this processResults on the main scheduler thread for performance+          -- which might reduce thread switching, especially for large batches+          -- but for now this seems to work just fine+        let rethrow = rethrowFromBg requests+        _ <- finally+          (restore (process service >> getResults) `catch` rethrow)+          (release service `catch` rethrow)+        return ()+      return ()+  where+    submit service (BlockedFetch request result) =+      (putResultFromBg result =<<) <$> enqueue service request+++{- |+A version of 'backgroundFetchAcquireReleaseMVar' where the dispatch function+is given a 'StablePtr PrimMVar' which is more useful for C based APIs.+-}++backgroundFetchAcquireRelease+  :: IO service+  -- ^ Resource acquisition for this datasource++  -> (service -> IO ())+  -- ^ Resource release++  -> (service -> Int -> StablePtr PrimMVar -> IO ())+  -- ^ Dispatch all the pending requests and when ready trigger the given mvar++  -> (service -> IO ())+  -- ^ Process all requests++  -> (forall a. service -> request a -> IO (IO (Either SomeException a)))+  -- ^ Submits an individual request to the service.++  -> State request+  -- ^ Currently unused.++  -> Flags+  -- ^ Currently unused.++  -> u+  -- ^ Currently unused.++  -> PerformFetch request++backgroundFetchAcquireRelease+  a r dispatch = backgroundFetchAcquireReleaseMVar+                 a+                 r+                 (\s c mvar -> do+                     sp <- newStablePtrPrimMVar mvar+                     dispatch s c sp)

@@ -73,8 +73,8 @@   tryWithRethrow,   ) where -#if __GLASGOW_HASKELL__ <= 708-import Control.Applicative ((<$>))+#if __GLASGOW_HASKELL__ >= 808+import Prelude hiding (MonadFail) #endif import Control.Exception as Exception import Data.Aeson@@ -361,14 +361,10 @@ -- rethrowAsyncExceptions :: SomeException -> IO () rethrowAsyncExceptions e-#if __GLASGOW_HASKELL__ >= 708   | Just SomeAsyncException{} <- fromException e = Exception.throw e-#endif-#if __GLASGOW_HASKELL__ >= 710   | Just AllocationLimitExceeded{} <- fromException e = Exception.throw e     -- AllocationLimitExceeded is not a child of SomeAsyncException,     -- but it should be.-#endif   | otherwise = return ()  tryWithRethrow :: IO a -> IO (Either SomeException a)

@@ -14,6 +14,8 @@ {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE KindSignatures #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE DeriveDataTypeable #-}  -- | Implementation of data-fetching operations.  Most users should -- import "Haxl.Core" instead.@@ -23,6 +25,7 @@   , dataFetchWithShow   , uncachedRequest   , cacheResult+  , dupableCacheRequest   , cacheResultWithShow   , cacheRequest   , performFetches@@ -71,15 +74,18 @@   = Uncached        (ResultVar a)        {-# UNPACK #-} !(IVar u w a)+       {-# UNPACK #-} !CallId    -- | The request has been seen before, but its result has not yet been   -- fetched.   | CachedNotFetched       {-# UNPACK #-} !(IVar u w a)+       {-# UNPACK #-} !CallId    -- | The request has been seen before, and its result has already been   -- fetched.   | Cached (ResultVal a w)+           {-# UNPACK #-} !CallId   -- | Show functions for request and its result.@@ -98,29 +104,32 @@   :: forall r a u w.      (DataSource u r, Typeable (r a))   => (r a -> String)    -- See Note [showFn]-  -> (r a -> IVar u w a -> DataCache (IVar u w) -> DataCache (IVar u w))+  -> (r a -> DataCacheItem u w a -> DataCache (DataCacheItem u w) -> IO ())   -> Env u w -> r a -> IO (CacheResult u w a)-cachedWithInsert showFn insertFn Env{..} req = do-  cache <- readIORef cacheRef+cachedWithInsert showFn insertFn env@Env{..} req = do   let     doFetch = do       ivar <- newIVar+      k <- nextCallId env       let !rvar = stdResultVar ivar completions submittedReqsRef flags             (Proxy :: Proxy r)-      writeIORef cacheRef $! insertFn req ivar cache-      return (Uncached rvar ivar)-  case DataCache.lookup req cache of+      insertFn req (DataCacheItem ivar k) dataCache+      return (Uncached rvar ivar k)+  mbRes <- DataCache.lookup req dataCache+  case mbRes of     Nothing -> doFetch-    Just (IVar cr) -> do+    Just (DataCacheItem i@IVar{ivarRef = cr} k) -> do       e <- readIORef cr       case e of-        IVarEmpty _ -> return (CachedNotFetched (IVar cr))+        IVarEmpty _ -> do+          ivar <- withCurrentCCS i+          return (CachedNotFetched ivar k)         IVarFull r -> do           ifTrace flags 3 $ putStrLn $ case r of             ThrowIO{} -> "Cached error: " ++ showFn req             ThrowHaxl{} -> "Cached error: " ++ showFn req             Ok{} -> "Cached request: " ++ showFn req-          return (Cached r)+          return (Cached r k)   -- | Make a ResultVar with the standard function for sending a CompletionReq@@ -143,26 +152,29 @@         getAllocationCounter       else         return 0-    -- Decrement the counter as request has finished-    ifReport flags 1 $-      atomicModifyIORef' ref (\m -> (subFromCountMap p 1 m, ()))-    atomically $ do+    atomicallyOnBlocking+      (LogicBug (ReadingCompletionsFailedFetch (dataSourceName p))) $ do       cs <- readTVar completions       writeTVar completions (CompleteReq r ivar allocs : cs)+    -- Decrement the counter as request has finished. Do this after updating the+    -- completions TVar so that if the scheduler is tracking what was being+    -- waited on it gets a consistent view.+    ifReport flags 1 $+      atomicModifyIORef' ref (\m -> (subFromCountMap p 1 m, ())) {-# INLINE stdResultVar #-}   -- | Record the call stack for a data fetch in the Stats.  Only useful -- when profiling.-logFetch :: Env u w -> (r a -> String) -> r a -> IO ()+logFetch :: Env u w -> (r a -> String) -> r a -> CallId -> IO () #ifdef PROFILING-logFetch env showFn req = do+logFetch env showFn req fid = do   ifReport (flags env) 5 $ do     stack <- currentCallStack     modifyIORef' (statsRef env) $ \(Stats s) ->-      Stats (FetchCall (showFn req) stack : s)+      Stats (FetchCall (showFn req) stack fid : s) #else-logFetch _ _ _ = return ()+logFetch _ _ _ _ = return () #endif  -- | Performs actual fetching of data for a 'Request' from a 'DataSource'.@@ -184,41 +196,44 @@   :: forall u w r a    . (DataSource u r, Eq (r a), Hashable (r a), Typeable (r a))   => (r a -> String)    -- See Note [showFn]-  -> (r a -> IVar u w a -> DataCache (IVar u w) -> DataCache (IVar u w))+  -> (r a -> DataCacheItem u w a -> DataCache (DataCacheItem u w) -> IO ())   -> r a   -> GenHaxl u w a dataFetchWithInsert showFn insertFn req =   GenHaxl $ \env@Env{..} -> do   -- First, check the cache   res <- cachedWithInsert showFn insertFn env req-  ifProfiling flags $ addProfileFetch env req   case res of     -- This request has not been seen before-    Uncached rvar ivar -> do-      logFetch env showFn req+    Uncached rvar ivar fid -> do+      logFetch env showFn req fid+      ifProfiling flags $ addProfileFetch env req fid False       --       -- Check whether the data source wants to submit requests       -- eagerly, or batch them up.       --+      let blockedFetch = BlockedFetch req rvar+      let blockedFetchI = BlockedFetchInternal fid       case schedulerHint userEnv :: SchedulerHint r of-        SubmitImmediately -> do-          (_,ios) <- performFetches 0 env-            [BlockedFetches [BlockedFetch req rvar]]-          when (not (null ios)) $-            error "bad data source:SubmitImmediately but returns FutureFetch"+        SubmitImmediately ->+          performFetches env [BlockedFetches [blockedFetch] [blockedFetchI]]         TryToBatch ->           -- add the request to the RequestStore and continue           modifyIORef' reqStoreRef $ \bs ->-            addRequest (BlockedFetch req rvar) bs+            addRequest blockedFetch blockedFetchI bs       ---      return $ Blocked ivar (Cont (getIVar ivar))+      return $ Blocked ivar (Return ivar)      -- Seen before but not fetched yet.  We're blocked, but we don't have     -- to add the request to the RequestStore.-    CachedNotFetched ivar -> return $ Blocked ivar (Cont (getIVar ivar))+    CachedNotFetched ivar fid -> do+      ifProfiling flags $ addProfileFetch env req fid True+      return $ Blocked ivar (Return ivar)      -- Cached: either a result, or an exception-    Cached r -> done r+    Cached r fid -> do+      ifProfiling flags $ addProfileFetch env req fid True+      done r  -- | A data request that is not cached.  This is not what you want for -- normal read requests, because then multiple identical requests may@@ -243,12 +258,13 @@   subRef <- env submittedReqsRef   if recording flg /= 0     then dataFetch req-    else GenHaxl $ \Env{..} -> do+    else GenHaxl $ \e@Env{..} -> do       ivar <- newIVar+      k <- nextCallId e       let !rvar = stdResultVar ivar completions subRef flg (Proxy :: Proxy r)       modifyIORef' reqStoreRef $ \bs ->-        addRequest (BlockedFetch req rvar) bs-      return $ Blocked ivar (Cont (getIVar ivar))+        addRequest (BlockedFetch req rvar) (BlockedFetchInternal k) bs+      return $ Blocked ivar (Return ivar)   -- | Transparently provides caching. Useful for datasources that can@@ -272,12 +288,13 @@ cacheResultWithInsert   :: Typeable (r a)   => (r a -> String)    -- See Note [showFn]-  -> (r a -> IVar u w a -> DataCache (IVar u w) -> DataCache (IVar u w)) -> r a-  -> IO a -> GenHaxl u w a-cacheResultWithInsert showFn insertFn req val = GenHaxl $ \env -> do-  let !ref = cacheRef env-  cache <- readIORef ref-  case DataCache.lookup req cache of+  -> (r a -> DataCacheItem u w a -> DataCache (DataCacheItem u w) -> IO ())+  -> r a+  -> IO a+  -> GenHaxl u w a+cacheResultWithInsert showFn insertFn req val = GenHaxl $ \e@Env{..} -> do+  mbRes <- DataCache.lookup req dataCache+  case mbRes of     Nothing -> do       eitherResult <- Exception.try val       case eitherResult of@@ -285,9 +302,10 @@         _ -> return ()       let result = eitherToResultThrowIO eitherResult       ivar <- newFullIVar result-      writeIORef ref $! insertFn req ivar cache+      k <- nextCallId e+      insertFn req (DataCacheItem ivar k) dataCache       done result-    Just (IVar cr) -> do+    Just (DataCacheItem IVar{ivarRef = cr} _) -> do       e <- readIORef cr       case e of         IVarEmpty _ -> corruptCache@@ -311,51 +329,53 @@ -- cacheRequest   :: Request req a => req a -> Either SomeException a -> GenHaxl u w ()-cacheRequest request result = GenHaxl $ \env -> do-  cache <- readIORef (cacheRef env)-  case DataCache.lookup request cache of+cacheRequest request result = GenHaxl $ \e@Env{..} -> do+  mbRes <- DataCache.lookup request dataCache+  case mbRes of     Nothing -> do       cr <- newFullIVar (eitherToResult result)-      writeIORef (cacheRef env) $! DataCache.insert request cr cache+      k <- nextCallId e+      DataCache.insert request (DataCacheItem cr k) dataCache       return (Done ()) -    -- It is an error if the request is already in the cache.  We can't test-    -- whether the cached result is the same without adding an Eq constraint,-    -- and we don't necessarily have Eq for all results.+    -- It is an error if the request is already in the cache.+    -- We can't test whether the cached result is the same without adding an+    -- Eq constraint, and we don't necessarily have Eq for all results.     _other -> raise $       DataSourceError "cacheRequest: request is already in the cache" +-- | Similar to @cacheRequest@ but doesn't throw an exception if the key+-- already exists in the cache.+-- If this function is called twice to cache the same Haxl request, the first+-- value will be discarded and overwritten with the second value.+-- Useful e.g. for unit tests+dupableCacheRequest+  :: Request req a => req a -> Either SomeException a -> GenHaxl u w ()+dupableCacheRequest request result = GenHaxl $ \e@Env{..} -> do+  cr <- newFullIVar (eitherToResult result)+  k <- nextCallId e+  DataCache.insert request (DataCacheItem cr k) dataCache+  return (Done ())+ performRequestStore-   :: forall u w. Int -> Env u w -> RequestStore u -> IO (Int, [IO ()])-performRequestStore n env reqStore =-  performFetches n env (contents reqStore)+   :: forall u w. Env u w -> RequestStore u -> IO ()+performRequestStore env reqStore =+  performFetches env (contents reqStore)  -- | Issues a batch of fetches in a 'RequestStore'. After -- 'performFetches', all the requests in the 'RequestStore' are -- complete, and all of the 'ResultVar's are full. performFetches-  :: forall u w. Int -> Env u w -> [BlockedFetches u] -> IO (Int, [IO ()])-performFetches n env@Env{flags=f, statsRef=sref} jobs = do-  let !n' = n + length jobs-+  :: forall u w. Env u w -> [BlockedFetches u] -> IO ()+performFetches env@Env{flags=f, statsRef=sref, statsBatchIdRef=sbref} jobs = do   t0 <- getTimestamp -  let-    roundstats =-      [ (dataSourceName (Proxy :: Proxy r), length reqs)-      | BlockedFetches (reqs :: [BlockedFetch r]) <- jobs ]--  ifTrace f 1 $-    printf "Batch data fetch (%s)\n" $-      intercalate (", "::String) $-        map (\(name,num) -> printf "%d %s" num (Text.unpack name)) roundstats-   ifTrace f 3 $-    forM_ jobs $ \(BlockedFetches reqs) ->+    forM_ jobs $ \(BlockedFetches reqs _) ->       forM_ reqs $ \(BlockedFetch r _) -> putStrLn (showp r)    let-    applyFetch i (BlockedFetches (reqs :: [BlockedFetch r])) =+    applyFetch i bfs@(BlockedFetches (reqs :: [BlockedFetch r]) _) =       case stateGet (states env) of         Nothing ->           return (FetchToDo reqs (SyncFetch (mapM_ (setError e))))@@ -365,10 +385,15 @@                   <> ": "                   <> Text.pack (showp req)         Just state ->-          return-            $ FetchToDo reqs+          return $ FetchToDo reqs             $ (if report f >= 2-                then wrapFetchInStats sref dsName (length reqs)+                then wrapFetchInStats+                        (userEnv env)+                        sref+                        sbref+                        dsName+                        (length reqs)+                        bfs                 else id)             $ wrapFetchInTrace i (length reqs) dsName             $ wrapFetchInCatch reqs@@ -376,17 +401,15 @@       where         dsName = dataSourceName (Proxy :: Proxy r) -  fetches <- zipWithM applyFetch [n..] jobs+  fetches <- zipWithM applyFetch [0..] jobs -  waits <- scheduleFetches fetches (submittedReqsRef env) (flags env)+  scheduleFetches fetches (submittedReqsRef env) (flags env)    t1 <- getTimestamp   let roundtime = fromIntegral (t1 - t0) / 1000000 :: Double    ifTrace f 1 $-    printf "Batch data fetch done (%.2fs)\n" (realToFrac roundtime :: Double)--  return (n', waits)+    printf "Batch data fetch done (%.4fs)\n" (realToFrac roundtime :: Double)  data FetchToDo where   FetchToDo@@ -415,9 +438,6 @@       -- sources themselves to catch (synchronous) exceptions.  Async       -- exceptions aren't a problem because we're going to rethrow       -- them all the way to runHaxl anyway.-    FutureFetch f ->-      FutureFetch $ \reqs -> f reqs `Exception.catch` (-                     \e -> handler e >> return (return ()))     BackgroundFetch f ->       BackgroundFetch $ \reqs -> f reqs `Exception.catch` handler   where@@ -432,86 +452,129 @@       putResult rvar (except e)  +data FailureCount = FailureCount+  { failureCountStandard :: {-# UNPACK #-} !Int+  , failureCountIgnored :: {-# UNPACK #-} !Int+  }++#if __GLASGOW_HASKELL__ >= 804+instance Semigroup FailureCount where+  (<>) = mappend+#endif++instance Monoid FailureCount where+  mempty = FailureCount 0 0+  mappend (FailureCount s1 i1) (FailureCount s2 i2)+    = FailureCount (s1+s2) (i1+i2)+ wrapFetchInStats-  :: IORef Stats+  :: DataSource u req+  => u+  -> IORef Stats+  -> IORef Int   -> Text   -> Int+  -> BlockedFetches u   -> PerformFetch req   -> PerformFetch req--wrapFetchInStats !statsRef dataSource batchSize perform = do+wrapFetchInStats+  u+  !statsRef+  !batchIdRef+  dataSource+  batchSize+  (BlockedFetches _reqs reqsI)+  perform = do   case perform of     SyncFetch f ->       SyncFetch $ \reqs -> do-        fail_ref <- newIORef 0-        (t0,t,alloc,_) <- statsForIO (f (map (addFailureCount fail_ref) reqs))+        bid <- newBatchId+        fail_ref <- newIORef mempty+        (t0,t,alloc,_) <- statsForIO (f (map (addFailureCount u fail_ref) reqs))         failures <- readIORef fail_ref-        updateFetchStats t0 t alloc batchSize failures+        updateFetchStats bid allFids t0 t alloc batchSize failures     AsyncFetch f -> do       AsyncFetch $ \reqs inner -> do+        bid <- newBatchId         inner_r <- newIORef (0, 0)-        fail_ref <- newIORef 0+        fail_ref <- newIORef mempty         let inner' = do               (_,t,alloc,_) <- statsForIO inner               writeIORef inner_r (t,alloc)-            reqs' = map (addFailureCount fail_ref) reqs+            reqs' = map (addFailureCount u fail_ref) reqs         (t0, totalTime, totalAlloc, _) <- statsForIO (f reqs' inner')         (innerTime, innerAlloc) <- readIORef inner_r         failures <- readIORef fail_ref-        updateFetchStats t0 (totalTime - innerTime) (totalAlloc - innerAlloc)-          batchSize failures-    FutureFetch submit ->-      FutureFetch $ \reqs -> do-        fail_ref <- newIORef 0-        let reqs' = map (addFailureCount fail_ref) reqs-        (t0, submitTime, submitAlloc, wait) <- statsForIO (submit reqs')-        return $ do-          (_, waitTime, waitAlloc, _) <- statsForIO wait-          failures <- readIORef fail_ref-          updateFetchStats t0 (submitTime + waitTime) (submitAlloc + waitAlloc)-            batchSize failures+        updateFetchStats bid allFids t0 (totalTime - innerTime)+          (totalAlloc - innerAlloc) batchSize failures     BackgroundFetch io -> do       BackgroundFetch $ \reqs -> do+        bid <- newBatchId         startTime <- getTimestamp-        io (map (addTimer startTime) reqs)+        io (zipWith (addTimer u bid startTime) reqs reqsI)   where+    allFids = map (\(BlockedFetchInternal k) -> k) reqsI+    newBatchId = atomicModifyIORef' batchIdRef $ \x -> (x+1,x+1)     statsForIO io = do       prevAlloc <- getAllocationCounter       (t0,t,a) <- time io       postAlloc <- getAllocationCounter       return (t0,t, fromIntegral $ prevAlloc - postAlloc, a) -    addTimer t0 (BlockedFetch req (ResultVar fn)) =-      BlockedFetch req $ ResultVar $ \result isChildThread -> do-        t1 <- getTimestamp-        -- We cannot measure allocation easily for BackgroundFetch. Here we-        -- just attribute all allocation to the last `putResultFromChildThread`-        -- and use 0 for the others. While the individual allocations may not-        -- be correct, the total sum and amortized allocation are still-        -- meaningful.-        -- see Note [tracking allocation in child threads]-        allocs <- if isChildThread then getAllocationCounter else return 0-        updateFetchStats t0 (t1 - t0)-          (negate allocs)-          1 -- batch size: we don't know if this is a batch or not-          (if isLeft result then 1 else 0) -- failures-        fn result isChildThread+    calcFailure _u _r (Right _) = mempty+    calcFailure u r (Left e) = case classifyFailure u r e of+      StandardFailure -> mempty { failureCountStandard = 1 }+      IgnoredForStatsFailure -> mempty { failureCountIgnored = 1 } ++    addTimer+      u+      bid+      t0+      (BlockedFetch req (ResultVar fn))+      (BlockedFetchInternal fid) =+        BlockedFetch req $ ResultVar $ \result isChildThread -> do+          t1 <- getTimestamp+          -- We cannot measure allocation easily for BackgroundFetch. Here we+          -- just attribute all allocation to the last+          -- `putResultFromChildThread` and use 0 for the others.+          -- While the individual allocations may not be correct,+          -- the total sum and amortized allocation are still meaningful.+          -- see Note [tracking allocation in child threads]+          allocs <- if isChildThread then getAllocationCounter else return 0+          updateFetchStats bid [fid] t0 (t1 - t0)+            (negate allocs)+            1 -- batch size: we don't know if this is a batch or not+            (calcFailure u req result) -- failures+          fn result isChildThread+     updateFetchStats-      :: Timestamp -> Microseconds -> Int64 -> Int -> Int -> IO ()-    updateFetchStats start time space batch failures = do+      :: Int+      -> [CallId]+      -> Timestamp+      -> Microseconds+      -> Int64+      -> Int+      -> FailureCount+      -> IO ()+    updateFetchStats bid fids start time space batch FailureCount{..} = do       let this = FetchStats { fetchDataSource = dataSource                             , fetchBatchSize = batch                             , fetchStart = start                             , fetchDuration = time                             , fetchSpace = space-                            , fetchFailures = failures }+                            , fetchFailures = failureCountStandard+                            , fetchIgnoredFailures = failureCountIgnored+                            , fetchBatchId = bid+                            , fetchIds = fids }       atomicModifyIORef' statsRef $ \(Stats fs) -> (Stats (this : fs), ()) -    addFailureCount :: IORef Int -> BlockedFetch r -> BlockedFetch r-    addFailureCount ref (BlockedFetch req (ResultVar fn)) =+    addFailureCount :: DataSource u r+      => u -> IORef FailureCount -> BlockedFetch r -> BlockedFetch r+    addFailureCount u ref (BlockedFetch req (ResultVar fn)) =       BlockedFetch req $ ResultVar $ \result isChildThread -> do-        when (isLeft result) $ atomicModifyIORef' ref (\r -> (r+1,()))+        let addFailures r = (r <> calcFailure u req result, ())+        when (isLeft result) $ atomicModifyIORef' ref addFailures         fn result isChildThread  wrapFetchInTrace@@ -546,7 +609,7 @@  -- | Start all the async fetches first, then perform the sync fetches before -- getting the results of the async fetches.-scheduleFetches :: [FetchToDo] -> IORef ReqCountMap -> Flags -> IO [IO ()]+scheduleFetches :: [FetchToDo] -> IORef ReqCountMap -> Flags -> IO () scheduleFetches fetches ref flags = do   -- update ReqCountmap for these fetches   ifReport flags 1 $ sequence_@@ -554,19 +617,30 @@         \m -> (addToCountMap (Proxy :: Proxy r) (length reqs) m, ())     | FetchToDo (reqs :: [BlockedFetch r]) _f <- fetches     ]++  ifTrace flags 1 $ printf "Batch data fetch round: %s\n" $+     intercalate (", "::String) $+        map (\(c, n, ds) -> printf "%s %s %d" n ds c) stats+   fully_async_fetches-  waits <- future_fetches   async_fetches sync_fetches-  return waits  where++  fetchName :: forall req . PerformFetch req -> String+  fetchName (BackgroundFetch _) = "background"+  fetchName (AsyncFetch _) = "async"+  fetchName (SyncFetch _) = "sync"++  srcName :: forall req . (DataSourceName req) => [BlockedFetch req] -> String+  srcName _ = Text.unpack $ dataSourceName (Proxy :: Proxy req)++  stats = [(length reqs, fetchName f, srcName reqs)+          | FetchToDo reqs f <- fetches]+   fully_async_fetches :: IO ()   fully_async_fetches = sequence_     [f reqs | FetchToDo reqs (BackgroundFetch f) <- fetches] -  future_fetches :: IO [IO ()]-  future_fetches = sequence-    [f reqs | FetchToDo reqs (FutureFetch f) <- fetches]-   async_fetches :: IO () -> IO ()   async_fetches = compose     [f reqs | FetchToDo reqs (AsyncFetch f) <- fetches]@@ -574,3 +648,10 @@   sync_fetches :: IO ()   sync_fetches = sequence_     [f reqs | FetchToDo reqs (SyncFetch f) <- fetches]+++-- | An exception thrown when reading from datasources fails+data ReadingCompletionsFailedFetch = ReadingCompletionsFailedFetch Text+  deriving Show++instance Exception ReadingCompletionsFailedFetch

@@ -4,8 +4,6 @@ -- This source code is distributed under the terms of a BSD license, -- found in the LICENSE file. -{-# LANGUAGE CPP #-}- -- | -- The 'Flags' type and related functions.  This module is provided -- for access to Haxl internals only; most users should import@@ -55,21 +53,13 @@   , recording = 0   } -#if __GLASGOW_HASKELL__ >= 710-#define FUNMONAD Monad m-#else-#define FUNMONAD (Functor m, Monad m)-#endif- -- | Runs an action if the tracing level is above the given threshold.-ifTrace :: FUNMONAD => Flags -> Int -> m a -> m ()+ifTrace :: Monad m => Flags -> Int -> m a -> m () ifTrace flags i = when (trace flags >= i) . void  -- | Runs an action if the report level is above the given threshold.-ifReport :: FUNMONAD => Flags -> Int -> m a -> m ()+ifReport :: Monad m => Flags -> Int -> m a -> m () ifReport flags i = when (report flags >= i) . void -ifProfiling :: FUNMONAD => Flags -> m a -> m ()+ifProfiling :: Monad m => Flags -> m a -> m () ifProfiling flags = when (report flags >= 4) . void--#undef FUNMONAD

@@ -5,7 +5,6 @@ -- found in the LICENSE file.  {-# LANGUAGE BangPatterns #-}-{-# LANGUAGE CPP #-} {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE MagicHash #-}@@ -40,15 +39,13 @@   , runMemo   ) where -#if __GLASGOW_HASKELL__ < 710-import Control.Applicative-#endif import Control.Exception as Exception hiding (throw) import Data.IORef import qualified Data.HashMap.Strict as HashMap import Data.Text (Text) import Data.Typeable import Data.Hashable+import Data.Int import Data.Word  import GHC.Prim (Addr#)@@ -57,7 +54,9 @@ import Haxl.Core.DataCache as DataCache import Haxl.Core.Flags import Haxl.Core.Monad+import Haxl.Core.Stats import Haxl.Core.Profile+import Haxl.Core.Util (trace_)  -- ----------------------------------------------------------------------------- -- Memoization@@ -76,15 +75,18 @@    => req a -> GenHaxl u w a -> GenHaxl u w a  cachedComputation req haxl = GenHaxl $ \env@Env{..} -> do-  cache <- readIORef memoRef-  ifProfiling flags $-     modifyIORef' profRef (incrementMemoHitCounterFor profLabel)-  case DataCache.lookup req cache of-    Just ivar -> unHaxl (getIVarWithWrites ivar) env+  mbRes <- DataCache.lookup req memoCache+  case mbRes of+    Just (DataCacheItem ivar k) -> do+      ifProfiling flags $ do+        incrementMemoHitCounterFor env k True+      unHaxl (getIVarWithWrites ivar) env     Nothing -> do       ivar <- newIVar-      writeIORef memoRef $! DataCache.insertNotShowable req ivar cache-      unHaxl (execMemoNow haxl ivar) env+      k <- nextCallId env+      -- no need to incremenetMemoHitCounter as execMemo will do it+      DataCache.insertNotShowable req (DataCacheItem ivar k) memoCache+      execMemoNowProfiled env haxl ivar k   -- | Like 'cachedComputation', but fails if the cache is already@@ -102,16 +104,15 @@      , Typeable (req a))   => req a -> GenHaxl u w a -> GenHaxl u w a preCacheComputation req haxl = GenHaxl $ \env@Env{..} -> do-  cache <- readIORef memoRef-  ifProfiling flags $-     modifyIORef' profRef (incrementMemoHitCounterFor profLabel)-  case DataCache.lookup req cache of+  mbRes <- DataCache.lookup req memoCache+  case mbRes of     Just _ -> return $ Throw $ toException $ InvalidParameter       "preCacheComputation: key is already cached"     Nothing -> do       ivar <- newIVar-      writeIORef memoRef $! DataCache.insertNotShowable req ivar cache-      unHaxl (execMemoNow haxl ivar) env+      k <- nextCallId env+      DataCache.insertNotShowable req (DataCacheItem ivar k) memoCache+      execMemoNowProfiled env haxl ivar k  -- ----------------------------------------------------------------------------- -- Memoization@@ -120,8 +121,8 @@  data MemoStatus u w a   = MemoEmpty-  | MemoReady (GenHaxl u w a)-  | MemoRun {-# UNPACK #-} !(IVar u w a)+  | MemoReady (GenHaxl u w a) CallId+  | MemoRun {-# UNPACK #-} !(IVar u w a) {-# UNPACK #-} !CallId  -- | Create a new @MemoVar@ for storing a memoized computation. The created -- @MemoVar@ is initially empty, not tied to any specific computation. Running@@ -135,7 +136,10 @@ -- the supplied computation. A @MemoVar@ must be prepared before it is run. prepareMemo :: MemoVar u w a -> GenHaxl u w a -> GenHaxl u w () prepareMemo (MemoVar memoRef) memoCmp-  = unsafeLiftIO $ writeIORef memoRef (MemoReady memoCmp)+  = GenHaxl $ \env -> do+      k <- nextCallId env+      writeIORef memoRef (MemoReady memoCmp k)+      return (Done ())  -- | Convenience function, combines @newMemo@ and @prepareMemo@. newMemoWith :: GenHaxl u w a -> GenHaxl u w (MemoVar u w a)@@ -197,46 +201,79 @@   stored <- readIORef memoRef   case stored of     -- Memo was not prepared first; throw an exception.-    MemoEmpty -> raise $ CriticalError "Attempting to run empty memo."+    MemoEmpty -> trace_ "MemoEmpty " $+      raise $ CriticalError "Attempting to run empty memo."     -- Memo has been prepared but not run yet-    MemoReady cont -> do+    MemoReady cont k -> trace_ "MemoReady" $ do       ivar <- newIVar-      writeIORef memoRef (MemoRun ivar)-      unHaxl (execMemoNow cont ivar) env+      writeIORef memoRef (MemoRun ivar k)+      execMemoNowProfiled env cont ivar k     -- The memo has already been run, get (or wait for) for the result-    MemoRun ivar -> unHaxl (getIVarWithWrites ivar) env+    MemoRun ivar k -> trace_ "MemoRun" $ do+      ifProfiling (flags env) $ do+        incrementMemoHitCounterFor env k True+      unHaxl (getIVarWithWrites ivar) env +execMemoNowProfiled+  :: Env u w+  -> GenHaxl u w a+  -> IVar u w a+  -> CallId+  -> IO (Result u w a)+execMemoNowProfiled envOuter cont ivar cid = if report (flags envOuter) < 4+  then execMemoNow envOuter cont ivar+  else do+    incrementMemoHitCounterFor envOuter cid False+    unHaxl+      (collectMemoData 0 $ GenHaxl $ \e -> execMemoNow e cont ivar)+      envOuter+  where+    addStats :: Env u w -> Int64 -> IO ()+    addStats env acc = modifyIORef' (statsRef env) $ \(Stats s) ->+      Stats (MemoCall cid acc : s)+    collectMemoData :: Int64 -> GenHaxl u w a -> GenHaxl u w a+    collectMemoData acc f = GenHaxl $ \env -> do+      a0 <- getAllocationCounter+      r <- unHaxl f env{memoKey=cid}+      a1 <- getAllocationCounter+      let newTotal = acc + (a0 - a1)+      ret <- case r of+        Done a -> do addStats env newTotal; return (Done a)+        Throw e -> do addStats env newTotal; return (Throw e)+        Blocked ivar k ->+          return (Blocked ivar (Cont (collectMemoData newTotal (toHaxl k))))+      setAllocationCounter a1+      return ret -execMemoNow :: GenHaxl u w a -> IVar u w a -> GenHaxl u w a-execMemoNow cont ivar = GenHaxl $ \env -> do+execMemoNow :: Env u w -> GenHaxl u w a -> IVar u w a -> IO (Result u w a)+execMemoNow env cont ivar = do   wlogs <- newIORef NilWrites   let-    !ienv = imperative env { writeLogsRef = wlogs }-    -- don't speculate under memoized things-    -- also we won't an env with empty writes, so we can memoize the extra+    !menv = env { writeLogsRef = wlogs }+    -- use an env with empty writes, so we can memoize the extra     -- writes done as part of 'cont'-  r <- Exception.try $ unHaxl cont ienv+  r <- Exception.try $ unHaxl cont menv    case r of-    Left e -> do+    Left e -> trace_ ("execMemoNow: Left " ++ show e) $ do       rethrowAsyncExceptions e       putIVar ivar (ThrowIO e) env       throwIO e-    Right (Done a) -> do+    Right (Done a) -> trace_ "execMemoNow: Done" $ do       wt <- readIORef wlogs       putIVar ivar (Ok a wt) env       mbModifyWLRef wt (writeLogsRef env)       return (Done a)-    Right (Throw ex) -> do+    Right (Throw ex) -> trace_ ("execMemoNow: Throw" ++ show ex) $ do       wt <- readIORef wlogs       putIVar ivar (ThrowHaxl ex wt) env       mbModifyWLRef wt (writeLogsRef env)       return (Throw ex)-    Right (Blocked ivar' cont) -> do-      -- We "block" this memoized computation in the new environment 'ienv', so+    Right (Blocked ivar' cont) -> trace_ "execMemoNow: Blocked" $ do+      -- We "block" this memoized computation in the new environment 'menv', so       -- that when it finishes, we can store all the write logs from the env       -- in the IVar.-      addJob ienv (toHaxl cont) ivar ivar'+      addJob menv (toHaxl cont) ivar ivar'       -- Now we call @getIVarWithWrites@ to populate the writes in the original       -- environment 'env'.       return (Blocked ivar (Cont (getIVarWithWrites ivar)))

@@ -21,7 +21,6 @@ {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE GADTs #-}-{-# LANGUAGE MagicHash #-} {-# LANGUAGE MultiWayIf #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternGuards #-}@@ -29,7 +28,6 @@ {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE TupleSections #-}  -- | -- The implementation of the 'Haxl' monad.  Most users should@@ -44,7 +42,9 @@     -- * Writes (for debugging only)   , WriteTree(..)   , tellWrite+  , tellWriteNoMemo   , write+  , writeNoMemo   , flattenWT   , appendWTs   , mbModifyWLRef@@ -58,6 +58,7 @@   , IVarContents(..)   , newIVar   , newFullIVar+  , withCurrentCCS   , getIVar   , getIVarWithWrites   , putIVar@@ -73,15 +74,19 @@      -- * Env   , Env(..)+  , DataCacheItem(..)   , Caches   , caches   , initEnvWithData   , initEnv   , emptyEnv   , env, withEnv-  , speculate-  , imperative+  , nextCallId+  , sanitizeEnv +    -- * Profiling+  , ProfileCurrent(..)+     -- * JobList   , JobList(..)   , appendJobList@@ -115,22 +120,27 @@ import Haxl.Core.Exception import Haxl.Core.RequestStore as RequestStore import Haxl.Core.DataCache as DataCache+import Haxl.Core.Util (trace_) +import Control.Applicative (liftA2) import Control.Arrow (left) import Control.Concurrent.STM import qualified Data.Text as Text import qualified Control.Monad.Catch as Catch import Control.Exception (Exception(..), SomeException, throwIO)+#if __GLASGOW_HASKELL__ >= 808+import Control.Monad hiding (MonadFail)+import qualified Control.Monad as CTL+#else import Control.Monad-import qualified Control.Exception as Exception-#if __GLASGOW_HASKELL__ < 710-import Control.Applicative hiding (Const) #endif-#if __GLASGOW_HASKELL__ < 706-import Prelude hiding (catch)-#endif+import qualified Control.Exception as Exception import Data.IORef import Data.Int+import Data.Either (rights)+#if __GLASGOW_HASKELL__ < 804+import Data.Semigroup+#endif import GHC.Exts (IsString(..)) import Text.PrettyPrint hiding ((<>)) import Text.Printf@@ -143,26 +153,28 @@ import qualified Data.Map as Map import Data.Text (Text) import Data.Typeable+import Foreign.Ptr (Ptr) import GHC.Stack import Haxl.Core.CallGraph #endif --trace_ :: String -> a -> a-trace_ _ = id---trace_ = Debug.Trace.trace- -- ----------------------------------------------------------------------------- -- The environment  -- | The data we carry around in the Haxl monad.++data DataCacheItem u w a = DataCacheItem (IVar u w a) {-# UNPACK #-} !CallId+ data Env u w = Env-  { cacheRef     :: {-# UNPACK #-} !(IORef (DataCache (IVar u w)))+  { dataCache     :: {-# UNPACK #-} !(DataCache (DataCacheItem u w))       -- ^ cached data fetches -  , memoRef      :: {-# UNPACK #-} !(IORef (DataCache (IVar u w)))+  , memoCache      :: {-# UNPACK #-} !(DataCache (DataCacheItem u w))       -- ^ memoized computations +  , memoKey    :: {-# UNPACK #-} !CallId+      -- ^ current running memo key+   , flags        :: !Flags       -- conservatively not unpacking, because this is passed       -- to 'fetch' and would need to be rebuilt.@@ -173,9 +185,16 @@   , statsRef     :: {-# UNPACK #-} !(IORef Stats)       -- ^ statistics, collected according to the 'report' level in 'flags'. -  , profLabel    :: ProfileLabel-      -- ^ current profiling label, see 'withLabel'+  , statsBatchIdRef :: {-# UNPACK #-} !(IORef Int)+     -- ^ keeps track of a Unique ID for each batch dispatched with stats+     -- enabled, for aggregating after. +  , callIdRef :: {-# UNPACK #-} !(IORef CallId)+     -- ^ keeps track of a Unique ID for each fetch/memo.++  , profCurrent    :: ProfileCurrent+     -- ^ current profiling label, see 'withLabel'+   , profRef      :: {-# UNPACK #-} !(IORef Profile)       -- ^ profiling data, collected according to the 'report' level in 'flags'. @@ -205,17 +224,15 @@        -- become non-empty is how the scheduler blocks waiting for        -- data fetches to return. -  , pendingWaits :: [IO ()]-       -- ^ this is a list of IO actions returned by 'FutureFetch'-       -- data sources.  These do a blocking wait for the results of-       -- some data fetch.--  , speculative :: {-# UNPACK #-} !Int-   , writeLogsRef :: {-# UNPACK #-} !(IORef (WriteTree w))        -- ^ A log of all writes done as part of this haxl computation. Any        -- haxl computation that needs to be memoized runs in its own-       -- environment so+       -- environment so that we can get a hold of those writes and put them+       -- in the IVar associated with the compuatation.+  , writeLogsRefNoMemo :: {-# UNPACK #-} !(IORef (WriteTree w))+       -- ^ This is just a specialized version of @writeLogsRef@, where we put+       -- logs that user doesn't want memoized. This is a better alternative to+       -- doing arbitrary IO from a (memoized) Haxl computation. #ifdef PROFILING   , callGraphRef ::  Maybe (IORef CallGraph)        -- ^ An edge list representing the current function call graph. The type@@ -226,38 +243,60 @@ #endif   } -type Caches u w = (IORef (DataCache (IVar u w)), IORef (DataCache (IVar u w)))+data ProfileCurrent = ProfileCurrent+  { profCurrentKey ::  {-# UNPACK #-} !ProfileKey+  , profCurrentLabel :: {-# UNPACK #-} !ProfileLabel+  } +type Caches u w = (DataCache (DataCacheItem u w), DataCache (DataCacheItem u w))+ caches :: Env u w -> Caches u w-caches env = (cacheRef env, memoRef env)+caches env = (dataCache env, memoCache env) +getMaxCallId :: DataCache (DataCacheItem u w) -> IO (Maybe Int)+getMaxCallId c = do+  callIds  <- rights . concatMap snd <$>+              DataCache.readCache c (\(DataCacheItem _ i) -> return i)+  case callIds of+    [] -> return Nothing+    vals -> return $ Just (maximum vals)++ -- | Initialize an environment with a 'StateStore', an input map, a -- preexisting 'DataCache', and a seed for the random number generator. initEnvWithData :: StateStore -> u -> Caches u w -> IO (Env u w)-initEnvWithData states e (cref, mref) = do+initEnvWithData states e (dcache, mcache) = do+  newCid <- max <$>+    (maybe 0 ((+) 1) <$> getMaxCallId dcache) <*>+    (maybe 0 ((+) 1) <$> getMaxCallId mcache)+  ciref<- newIORef newCid   sref <- newIORef emptyStats+  sbref <- newIORef 0   pref <- newIORef emptyProfile   rs <- newIORef noRequests          -- RequestStore-  rq <- newIORef JobNil-  sr <- newIORef emptyReqCounts+  rq <- newIORef JobNil              -- RunQueue+  sr <- newIORef emptyReqCounts      -- SubmittedReqs   comps <- newTVarIO []              -- completion queue   wl <- newIORef NilWrites+  wlnm <- newIORef NilWrites   return Env-    { cacheRef = cref-    , memoRef = mref+    { dataCache = dcache+    , memoCache = mcache+    , memoKey = (-1)     , flags = defaultFlags     , userEnv = e     , states = states     , statsRef = sref-    , profLabel = "MAIN"+    , statsBatchIdRef = sbref+    , profCurrent = ProfileCurrent 0 "MAIN"+    , callIdRef = ciref     , profRef = pref     , reqStoreRef = rs     , runQueueRef = rq     , submittedReqsRef = sr     , completions = comps-    , pendingWaits = []-    , speculative = 0     , writeLogsRef = wl+    , writeLogsRefNoMemo = wlnm #ifdef PROFILING     , callGraphRef = Nothing     , currFunction = mainFunction@@ -267,23 +306,43 @@ -- | Initializes an environment with 'StateStore' and an input map. initEnv :: StateStore -> u -> IO (Env u w) initEnv states e = do-  cref <- newIORef emptyDataCache-  mref <- newIORef emptyDataCache-  initEnvWithData states e (cref,mref)+  dcache <- emptyDataCache+  mcache <- emptyDataCache+  initEnvWithData states e (dcache, mcache)  -- | A new, empty environment. emptyEnv :: u -> IO (Env u w) emptyEnv = initEnv stateEmpty -speculate :: Env u w -> Env u w-speculate env@Env{..}-  | speculative == 0 = env { speculative = 1 }-  | otherwise = env--imperative :: Env u w -> Env u w-imperative env@Env{..}-  | speculative == 1 = env { speculative = 0 }-  | otherwise = env+-- | If you're using the env from a failed Haxl computation in a second Haxl+-- computation, it is recommended to sanitize the Env to remove all empty+-- IVars - especially if it's possible the first Haxl computation could've+-- been interrupted via an async exception. This is because if the Haxl+-- computation was interrupted by an exception, it's possible that there are+-- entries in the cache which are still blocked, while the results from+-- outgone fetches have been discarded.+sanitizeEnv :: Env u w -> IO (Env u w)+sanitizeEnv env@Env{..} = do+  sanitizedDC <- DataCache.filter isIVarFull dataCache+  sanitizedMC <- DataCache.filter isIVarFull memoCache+  rs <- newIORef noRequests          -- RequestStore+  rq <- newIORef JobNil              -- RunQueue+  comps <- newTVarIO []              -- completion queue+  sr <- newIORef emptyReqCounts      -- SubmittedReqs+  return env+    { dataCache = sanitizedDC+    , memoCache = sanitizedMC+    , reqStoreRef = rs+    , runQueueRef = rq+    , completions = comps+    , submittedReqsRef = sr+    }+  where+  isIVarFull (DataCacheItem IVar{..} _) = do+    ivarContents <- readIORef ivarRef+    case ivarContents of+      IVarFull _ -> return True+      _ -> return False  -- ----------------------------------------------------------------------------- -- WriteTree@@ -307,6 +366,7 @@  -- This function must be called at the end of the Haxl computation to get -- a list of writes.+-- Haxl provides no guarantees on the order of the returned logs. flattenWT :: WriteTree w -> [w] flattenWT = go []   where@@ -327,10 +387,12 @@ --    of the scheduler, including unfetched requests and the run queue --    of computations. ----- * It is a writer monad for 'WriteTree'. We strongly advise these be---   used only for logs used for debugging. These are not memoized.---   Other relevant writes should be returned as function output,---   which is the more "functional" way.+--  * It is a writer monad for 'WriteTree'. These can be used to do+--    arbitrary "logs" from any Haxl computation. These are better than+--    doing arbitrary IO from a Haxl computation as these writes also get+--    memoized if the Haxl computation associated with them is memoized.+--    Now if this memoized computation is run again, you'll get the writes+--    twice.  --  * It is a concurrency, or resumption, monad. A computation may run --    partially and return 'Blocked', in which case the framework should@@ -354,6 +416,15 @@   mbModifyWLRef wt writeLogsRef   return $ Done () +tellWriteNoMemo :: w -> GenHaxl u w ()+tellWriteNoMemo = writeNoMemo . SomeWrite++writeNoMemo :: WriteTree w -> GenHaxl u w ()+writeNoMemo wt = GenHaxl $ \Env{..} -> do+  mbModifyWLRef wt writeLogsRefNoMemo+  return $ Done ()++ instance IsString a => IsString (GenHaxl u w a) where   fromString s = return (fromString s) @@ -399,7 +470,15 @@  -- | A synchronisation point.  It either contains a value, or a list -- of computations waiting for the value.-newtype IVar u w a = IVar (IORef (IVarContents u w a))+#ifdef PROFILING+data IVar u w a = IVar+  { ivarRef :: {-# UNPACK #-} !(IORef (IVarContents u w a))+  , ivarCCS :: {-# UNPACK #-} !(Ptr CostCentreStack)+#else+newtype IVar u w a = IVar+  { ivarRef :: IORef (IVarContents u w a)+#endif+  }  data IVarContents u w a   = IVarFull (ResultVal a w)@@ -410,34 +489,53 @@     -- faster (benchmarked with tests/MonadBench.hs).  newIVar :: IO (IVar u w a)-newIVar = IVar <$> newIORef (IVarEmpty JobNil)+newIVar = do+  ivarRef <- newIORef (IVarEmpty JobNil)+#ifdef PROFILING+  ivarCCS <- getCurrentCCS ivarRef+#endif+  return IVar{..}  newFullIVar :: ResultVal a w -> IO (IVar u w a)-newFullIVar r = IVar <$> newIORef (IVarFull r)+newFullIVar r = do+  ivarRef <- newIORef (IVarFull r)+#ifdef PROFILING+  ivarCCS <- getCurrentCCS ivarRef+#endif+  return IVar{..} +withCurrentCCS :: IVar u w a -> IO (IVar u w a)+#ifdef PROFILING+withCurrentCCS ivar = do+  ccs <- getCurrentCCS ivar+  return ivar{ivarCCS = ccs}+#else+withCurrentCCS = return+#endif+ getIVar :: IVar u w a -> GenHaxl u w a-getIVar (IVar !ref) = GenHaxl $ \Env{..} -> do+getIVar i@IVar{ivarRef = !ref} = GenHaxl $ \Env{..} -> do   e <- readIORef ref   case e of     IVarFull (Ok a _wt) -> return (Done a)-    IVarFull (ThrowHaxl e _wt) -> return (Throw e)+    IVarFull (ThrowHaxl e _wt) -> raiseFromIVar i e     IVarFull (ThrowIO e) -> throwIO e-    IVarEmpty _ -> return (Blocked (IVar ref) (Cont (getIVar (IVar ref))))+    IVarEmpty _ -> return (Blocked i (Return i))  -- Just a specialised version of getIVar, for efficiency in <*> getIVarApply :: IVar u w (a -> b) -> a -> GenHaxl u w b-getIVarApply (IVar !ref) a = GenHaxl $ \Env{..} -> do+getIVarApply i@IVar{ivarRef = !ref} a = GenHaxl $ \Env{..} -> do   e <- readIORef ref   case e of     IVarFull (Ok f _wt) -> return (Done (f a))-    IVarFull (ThrowHaxl e _wt) -> return (Throw e)+    IVarFull (ThrowHaxl e _wt) -> raiseFromIVar i e     IVarFull (ThrowIO e) -> throwIO e     IVarEmpty _ ->-      return (Blocked (IVar ref) (Cont (getIVarApply (IVar ref) a)))+      return (Blocked i (Cont (getIVarApply i a)))  -- Another specialised version of getIVar, for efficiency in cachedComputation getIVarWithWrites :: IVar u w a -> GenHaxl u w a-getIVarWithWrites (IVar !ref) = GenHaxl $ \Env{..} -> do+getIVarWithWrites i@IVar{ivarRef = !ref} = GenHaxl $ \Env{..} -> do   e <- readIORef ref   case e of     IVarFull (Ok a wt) -> do@@ -445,23 +543,27 @@       return (Done a)     IVarFull (ThrowHaxl e wt) -> do       mbModifyWLRef wt writeLogsRef-      return  (Throw e)+      raiseFromIVar i e     IVarFull (ThrowIO e) -> throwIO e     IVarEmpty _ ->-      return (Blocked (IVar ref) (Cont (getIVarWithWrites (IVar ref))))+      return (Blocked i (Cont (getIVarWithWrites i)))  putIVar :: IVar u w a -> ResultVal a w -> Env u w -> IO ()-putIVar (IVar ref) a Env{..} = do+putIVar IVar{ivarRef = !ref} a Env{..} = do   e <- readIORef ref   case e of     IVarEmpty jobs -> do       writeIORef ref (IVarFull a)       modifyIORef' runQueueRef (appendJobList jobs)-    IVarFull{} -> error "putIVar: multiple put"+      -- An IVar is typically only meant to be written to once+      -- so it would make sense to throw an error here. But there+      -- are legitimate use-cases for writing several times.+      -- (See Haxl.Core.Parallel)+    IVarFull{} -> return ()  {-# INLINE addJob #-} addJob :: Env u w -> GenHaxl u w b -> IVar u w b -> IVar u w a -> IO ()-addJob env !haxl !resultIVar (IVar !ref) =+addJob env !haxl !resultIVar IVar{ivarRef = !ref} =   modifyIORef' ref $ \contents ->     case contents of       IVarEmpty list -> IVarEmpty (JobCons env haxl resultIVar list)@@ -486,7 +588,7 @@  done :: ResultVal a w -> IO (Result u w a) done (Ok a _) = return (Done a)-done (ThrowHaxl e _) = return (Throw e)+done (ThrowHaxl e _) = raise e done (ThrowIO e) = throwIO e  eitherToResultThrowIO :: Either SomeException a -> ResultVal a w@@ -530,10 +632,6 @@ putResult is that this is already a synchronisation method, because runHaxl is waiting for the result of the dataFetch. -(slight wrinkle here: runHaxl might not wait for the result of the-dataFetch in the case where we do some speculative execution in-pAnd/pOr)- We need a special version of putResult for child threads (putResultFromChildThread), because we don't want to propagate any allocation from the runHaxl thread back to itself and count it twice.@@ -630,22 +728,25 @@   = Cont (GenHaxl u w a)   | forall b. Cont u w b :>>= (b -> GenHaxl u w a)   | forall b. (b -> a) :<$> (Cont u w b)+  | Return (IVar u w a)  toHaxl :: Cont u w a -> GenHaxl u w a toHaxl (Cont haxl) = haxl toHaxl (m :>>= k) = toHaxlBind m k toHaxl (f :<$> x) = toHaxlFmap f x+toHaxl (Return i) = getIVar i  toHaxlBind :: Cont u w b -> (b -> GenHaxl u w a) -> GenHaxl u w a toHaxlBind (m :>>= k) k2 = toHaxlBind m (k >=> k2) toHaxlBind (Cont haxl) k = haxl >>= k toHaxlBind (f :<$> x) k = toHaxlBind x (k . f)+toHaxlBind (Return i) k = getIVar i >>= k  toHaxlFmap :: (a -> b) -> Cont u w a -> GenHaxl u w b toHaxlFmap f (m :>>= k) = toHaxlBind m (k >=> return . f) toHaxlFmap f (Cont haxl) = f <$> haxl toHaxlFmap f (g :<$> x) = toHaxlFmap (f . g) x-+toHaxlFmap f (Return i) = f <$> getIVar i  -- ----------------------------------------------------------------------------- -- Monad/Applicative instances@@ -659,12 +760,16 @@       Throw e -> return (Throw e)       Blocked ivar cont -> trace_ ">>= Blocked" $         return (Blocked ivar (cont :>>= k))-  fail msg = GenHaxl $ \_env ->-    return $ Throw $ toException $ MonadFail $ Text.pack msg    -- We really want the Applicative version of >>   (>>) = (*>) +#if __GLASGOW_HASKELL__ >= 808+instance CTL.MonadFail (GenHaxl u w) where+#endif+  fail msg = GenHaxl $ \_env ->+    return $ Throw $ toException $ MonadFail $ Text.pack msg+ instance Functor (GenHaxl u w) where   fmap f (GenHaxl m) = GenHaxl $ \env -> do     r <- m env@@ -688,24 +793,40 @@             return (Blocked ivar (f :<$> fcont))       Throw e -> trace_ "Throw" $ return (Throw e)       Blocked ivar1 fcont -> do-         ra <- aa env-         case ra of-           Done a -> trace_ "Blocked/Done" $-             return (Blocked ivar1 (($ a) :<$> fcont))-           Throw e -> trace_ "Blocked/Throw" $-             return (Blocked ivar1 (fcont :>>= (\_ -> throw e)))-           Blocked ivar2 acont -> trace_ "Blocked/Blocked" $ do+        ra <- aa env+        case ra of+          Done a -> trace_ "Blocked/Done" $+            return (Blocked ivar1 (($ a) :<$> fcont))+          Throw e -> trace_ "Blocked/Throw" $+            return (Blocked ivar1 (fcont :>>= (\_ -> throw e)))+          Blocked ivar2 acont -> trace_ "Blocked/Blocked" $+            blockedBlocked env ivar1 fcont ivar2 acont              -- Note [Blocked/Blocked]-              if speculative env /= 0-                then-                  return (Blocked ivar1-                            (Cont (toHaxl fcont <*> toHaxl acont)))-                else do-                  i <- newIVar-                  addJob env (toHaxl fcont) i ivar1-                  let cont = acont :>>= \a -> getIVarApply i a-                  return (Blocked ivar2 cont) +instance Semigroup a => Semigroup (GenHaxl u w a) where+  (<>) = liftA2 (<>)++instance Monoid a => Monoid (GenHaxl u w a) where+  mempty = pure mempty+  mappend = liftA2 mappend++blockedBlocked+  :: Env u w+  -> IVar u w c+  -> Cont u w (a -> b)+  -> IVar u w d+  -> Cont u w a+  -> IO (Result u w b)+blockedBlocked _ _ (Return i) ivar2 acont =+  return (Blocked ivar2 (acont :>>= getIVarApply i))+blockedBlocked _ _ (g :<$> Return i) ivar2 acont =+  return (Blocked ivar2 (acont :>>= \ a -> (\f -> g f a) <$> getIVar i))+blockedBlocked env ivar1 fcont ivar2 acont = do+  i <- newIVar+  addJob env (toHaxl fcont) i ivar1+  let cont = acont :>>= \a -> getIVarApply i a+  return (Blocked ivar2 cont)+ -- Note [Blocked/Blocked] -- -- This is the tricky case: we're blocked on both sides of the <*>.@@ -750,6 +871,9 @@     Blocked ivar k ->       return (Blocked ivar (Cont (withEnv newEnv (toHaxl k)))) +nextCallId :: Env u w -> IO CallId+nextCallId env = atomicModifyIORef' (callIdRef env) $ \x -> (x+1,x+1)+ #ifdef PROFILING -- ----------------------------------------------------------------------------- -- CallGraph recording@@ -800,6 +924,20 @@   where     somex = toException e +raiseFromIVar :: Exception e => IVar u w a -> e -> IO (Result u w b)+#ifdef PROFILING+raiseFromIVar ivar e+  | Just (HaxlException Nothing h) <- fromException somex = do+    stk <- ccsToStrings (ivarCCS ivar)+    return (Throw (toException (HaxlException (Just stk) h)))+  | otherwise+#else+raiseFromIVar _ivar e+#endif+    = return (Throw somex)+  where+    somex = toException e+ -- | Catch an exception in the Haxl monad catch :: Exception e => GenHaxl u w a -> (e -> GenHaxl u w a) -> GenHaxl u w a catch (GenHaxl m) h = GenHaxl $ \env -> do@@ -840,9 +978,20 @@ -- the Haxl monad.  This is morally unsafe, because you could then -- catch those exceptions in Haxl and observe the underlying execution -- order.  Not to be exposed to user code.+--+-- Note: this function does not catch async exceptions. This is a flaw in Haxl+-- where it can sometimes leave the environment in a bad state when async+-- exceptions are thrown (for example the cache may think a fetch is happening+-- but the exception has stopped it). TODO would be to make Haxl async exception+-- safe and then remove the rethrowAsyncExceptions below, but for now this is+-- safer to avoid bugs. Additionally this would not protect you from async+-- exceptions thrown while executing code in the scheduler, and so relying on+-- this function to catch all async exceptions would be ambitious at best. unsafeToHaxlException :: GenHaxl u w a -> GenHaxl u w a unsafeToHaxlException (GenHaxl m) = GenHaxl $ \env -> do-  r <- m env `Exception.catch` \e -> return (Throw e)+  r <- m env `Exception.catch` \e -> do+    rethrowAsyncExceptions e+    return (Throw e)   case r of     Blocked cvar c ->       return (Blocked cvar (Cont (unsafeToHaxlException (toHaxl c))))@@ -868,19 +1017,22 @@ -- >   cacheRequest (CountAardvarks "abcabc") (Right (2)) -- dumpCacheAsHaskell :: GenHaxl u w String-dumpCacheAsHaskell = dumpCacheAsHaskellFn "loadCache" "GenHaxl u w ()"+dumpCacheAsHaskell =+    dumpCacheAsHaskellFn "loadCache" "GenHaxl u w ()" "cacheRequest"  -- | Dump the contents of the cache as Haskell code that, when -- compiled and run, will recreate the same cache contents. -- Does not take into account the writes done as part of the computation. -- -- Takes the name and type for the resulting function as arguments.-dumpCacheAsHaskellFn :: String -> String -> GenHaxl u w String-dumpCacheAsHaskellFn fnName fnType = do-  ref <- env cacheRef  -- NB. cacheRef, not memoRef.  We ignore memoized+-- Also takes the cacheFn to use, we can use either @cacheRequest@ or+-- @dupableCacheRequest@.+dumpCacheAsHaskellFn :: String -> String -> String -> GenHaxl u w String+dumpCacheAsHaskellFn fnName fnType cacheFn = do+  cache <- env dataCache  -- NB. dataCache, not memoCache.  We ignore memoized                        -- results when dumping the cache.   let-    readIVar (IVar ref) = do+    readIVar (DataCacheItem IVar{ivarRef = !ref} _) = do       r <- readIORef ref       case r of         IVarFull (Ok a _) -> return (Just (Right a))@@ -889,12 +1041,11 @@         IVarEmpty _ -> return Nothing      mk_cr (req, res) =-      text "cacheRequest" <+> parens (text req) <+> parens (result res)+      text cacheFn <+> parens (text req) <+> parens (result res)     result (Left e) = text "except" <+> parens (text (show e))     result (Right s) = text "Right" <+> parens (text s)    entries <- unsafeLiftIO $ do-    cache <- readIORef ref     showCache cache readIVar    let

@@ -6,6 +6,7 @@  {-# LANGUAGE CPP #-} {-# LANGUAGE BangPatterns #-}+{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-}  -- | Psuedo-parallel operations.  Most users should import "Haxl.Core"@@ -13,12 +14,17 @@ -- module Haxl.Core.Parallel   ( -- * Parallel operations-    pAnd+    biselect+  , pAnd   , pOr+  , unsafeChooseFirst   ) where -import Haxl.Core.Monad+import Haxl.Core.Monad hiding (catch, throw)+import Haxl.Core.Exception +import Control.Exception (throw)+ -- ----------------------------------------------------------------------------- -- Parallel operations @@ -26,6 +32,71 @@ infixr 5 `pAnd` infixr 4 `pOr` ++biselect :: GenHaxl u w (Either a b)+         -> GenHaxl u w (Either a c)+         -> GenHaxl u w (Either a (b,c))+biselect haxla haxlb = biselect_opt id id Left Right haxla haxlb++{-# INLINE biselect_opt #-}+biselect_opt :: (l -> Either a b)+             -> (r -> Either a c)+             -> (a -> t)+             -> ((b,c) -> t)+             -> GenHaxl u w l+             -> GenHaxl u w r+             -> GenHaxl u w t+biselect_opt discrimA discrimB left right haxla haxlb =+  let go (GenHaxl haxla) (GenHaxl haxlb) = GenHaxl $ \env@Env{..} -> do+        ra <- haxla env+        case ra of+          Done ea ->+            case discrimA ea of+              Left a -> return (Done (left a))+              Right b -> do+                  rb <- haxlb env+                  case rb of+                    Done eb ->+                      case discrimB eb of+                        Left a -> return (Done (left a))+                        Right c -> return (Done (right (b,c)))+                    Throw e -> return (Throw e)+                    Blocked ib haxlb' ->+                      return (Blocked ib+                              (haxlb' :>>= \b' -> go_right b b'))+          Throw e -> return (Throw e)+          Blocked ia haxla' -> do+            rb <- haxlb env+            case rb of+              Done eb ->+                case discrimB eb of+                  Left a -> return (Done (left a))+                  Right c ->+                     return (Blocked ia+                             (haxla' :>>= \a' -> go_left a' c))+              Throw e -> return (Throw e)+              Blocked ib haxlb' -> do+                i <- newIVar+                addJob env (return ()) i ia+                addJob env (return ()) i ib+                return (Blocked i (Cont (go (toHaxl haxla') (toHaxl haxlb'))))+                -- The code above makes sure that the computation+                -- wakes up whenever either 'ia' or 'ib' is filled.+                -- The ivar 'i' is used as a synchronisation point+                -- for the whole computation, and we make sure that+                -- whenever 'ia' or 'ib' are filled in then 'i' will+                -- also be filled.++      go_right b eb =+        case discrimB eb of+          Left a -> return (left a)+          Right c -> return (right (b,c))+      go_left ea c =+        case discrimA ea of+          Left a -> return (left a)+          Right b -> return (right (b,c))+  in go haxla haxlb+ -- | Parallel version of '(.||)'.  Both arguments are evaluated in -- parallel, and if either returns 'True' then the other is -- not evaluated any further.@@ -36,26 +107,12 @@ -- the other argument may have produced if it had been allowed to -- complete. pOr :: GenHaxl u w Bool -> GenHaxl u w Bool -> GenHaxl u w Bool-GenHaxl a `pOr` GenHaxl b = GenHaxl $ \env@Env{..} -> do-  let !senv = speculate env-  ra <- a senv-  case ra of-    Done True -> return (Done True)-    Done False -> b env  -- not speculative-    Throw _ -> return ra-    Blocked ia a' -> do-      rb <- b senv-      case rb of-        Done True -> return rb-        Done False -> return ra-        Throw _ -> return rb-        Blocked _ b' -> return (Blocked ia (Cont (toHaxl a' `pOr` toHaxl b')))-          -- Note [pOr Blocked/Blocked]-          -- This will only wake up when ia is filled, which-          -- is whatever the left side was waiting for.  This is-          -- suboptimal because the right side might wake up first,-          -- but handling this non-determinism would involve a much-          -- more complicated implementation here.+pOr x y = biselect_opt discrim discrim left right x y+  where+    discrim True = Left ()+    discrim False = Right ()+    left _ = True+    right _ = False  -- | Parallel version of '(.&&)'.  Both arguments are evaluated in -- parallel, and if either returns 'False' then the other is@@ -67,18 +124,28 @@ -- the other argument may have produced if it had been allowed to -- complete. pAnd :: GenHaxl u w Bool -> GenHaxl u w Bool -> GenHaxl u w Bool-GenHaxl a `pAnd` GenHaxl b = GenHaxl $ \env@Env{..} -> do-  let !senv = speculate env-  ra <- a senv-  case ra of-    Done False -> return (Done False)-    Done True -> b env-    Throw _ -> return ra-    Blocked ia a' -> do-      rb <- b senv-      case rb of-        Done False -> return rb-        Done True -> return ra-        Throw _ -> return rb-        Blocked _ b' -> return (Blocked ia (Cont (toHaxl a' `pAnd` toHaxl b')))-         -- See Note [pOr Blocked/Blocked]+pAnd x y = biselect_opt discrim discrim left right x y+  where+    discrim False = Left ()+    discrim True = Right ()+    left _ = False+    right _ = True++-- | This function takes two haxl computations as input, and returns the+-- output of whichever computation finished first. This is clearly+-- non-deterministic in its output and exception behavior, be careful when+-- using it.+unsafeChooseFirst+  :: GenHaxl u w a+  -> GenHaxl u w b+  -> GenHaxl u w (Either a b)+unsafeChooseFirst x y = biselect_opt discrimx discrimy id right x y+  where+    discrimx :: a -> Either (Either a b) ()+    discrimx a = Left (Left a)++    discrimy :: b -> Either (Either a b) ()+    discrimy b = Left (Right b)++    right _ = throw $ CriticalError+      "unsafeChooseFirst: We should never have a 'Right ()'"

@@ -9,6 +9,7 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}  -- | Implementation of lightweight profiling.  Most users should -- import "Haxl.Core" instead.@@ -27,13 +28,10 @@ #if __GLASGOW_HASKELL__ < 804 import Data.Monoid #endif-import Data.Text (Text) import Data.Typeable import qualified Data.HashMap.Strict as HashMap-import qualified Data.HashSet as HashSet import GHC.Exts import qualified Data.Text as Text- import Haxl.Core.DataSource import Haxl.Core.Flags import Haxl.Core.Stats@@ -66,40 +64,98 @@   -> Env u w   -> IO (Result u w a) collectProfileData l m env = do-   a0 <- getAllocationCounter-   r <- m env{profLabel=l} -- what if it throws?-   a1 <- getAllocationCounter-   modifyProfileData env l (a0 - a1)-   -- So we do not count the allocation overhead of modifyProfileData-   setAllocationCounter a1-   case r of-     Done a -> return (Done a)-     Throw e -> return (Throw e)-     Blocked ivar k -> return (Blocked ivar (Cont (withLabel l (toHaxl k))))+  let (ProfileCurrent prevProfKey prevProfLabel) = profCurrent env+  if prevProfLabel == l+  then+    -- do not add a new label if we are recursing+    m env+  else do+    key <- atomicModifyIORef' (profRef env) $ \p ->+      case HashMap.lookup (l, prevProfKey) (profileTree p) of+        Just k -> (p, k)+        Nothing -> (p+          { profileTree = HashMap.insert+            (l, prevProfKey)+            (profileNextKey p)+            (profileTree p)+          , profileNextKey = profileNextKey p + 1 }, profileNextKey p)+    runProfileData l key m False env {-# INLINE collectProfileData #-} -modifyProfileData :: Env u w -> ProfileLabel -> AllocCount -> IO ()-modifyProfileData env label allocs =+runProfileData+  :: ProfileLabel+  -> ProfileKey+  -> (Env u w -> IO (Result u w a))+  -> Bool+  -> Env u w+  -> IO (Result u w a)+runProfileData l key m isCont env = do+  t0 <- getTimestamp+  a0 <- getAllocationCounter+  let+    nextCurrent = ProfileCurrent+                  { profCurrentKey = key+                  , profCurrentLabel = l }+    caller = profCurrentKey (profCurrent env)++  r <- m env{profCurrent=nextCurrent} -- what if it throws?+  a1 <- getAllocationCounter+  t1 <- getTimestamp++  -- caller might not be the actual caller of this function+  -- for example MAIN may be continuing a function from the middle of the stack.+  -- But this is what we want as we need to account for allocations.+  -- So do not be tempted to pass through prevProfKey (from collectProfileData)+  -- which is the original caller+  modifyProfileData env key caller (a0 - a1) (t1-t0) (if isCont then 0 else 1)++  -- So we do not count the allocation overhead of modifyProfileData+  setAllocationCounter a1+  case r of+    Done a -> return (Done a)+    Throw e -> return (Throw e)+    Blocked ivar k -> return (Blocked ivar (Cont $ runCont (toHaxl k)))+  where+    runCont (GenHaxl h) = GenHaxl $ runProfileData l key h True+{-# INLINE runProfileData #-}++modifyProfileData+  :: Env u w+  -> ProfileKey+  -> ProfileKey+  -> AllocCount+  -> Microseconds+  -> LabelHitCount+  -> IO ()+modifyProfileData env key caller allocs t labelIncrement = do   modifyIORef' (profRef env) $ \ p ->     p { profile =-          HashMap.insertWith updEntry label newEntry .+          HashMap.insertWith updEntry key newEntry .           HashMap.insertWith updCaller caller newCaller $           profile p }-  where caller = profLabel env-        newEntry =+  where newEntry =           emptyProfileData             { profileAllocs = allocs-            , profileDeps = HashSet.singleton caller }+            , profileLabelHits = labelIncrement+            , profileTime = t+            }         updEntry _ old =-          old { profileAllocs = profileAllocs old + allocs-              , profileDeps = HashSet.insert caller (profileDeps old) }-        -- subtract allocs from caller, so they are not double counted+          old+            { profileAllocs = profileAllocs old + allocs+            , profileLabelHits = profileLabelHits old + labelIncrement+            , profileTime = profileTime old + t+            }+        -- subtract allocs/time from caller, so they are not double counted         -- we don't know the caller's caller, but it will get set on         -- the way back out, so an empty hashset is fine for now         newCaller =-          emptyProfileData { profileAllocs = -allocs }+          emptyProfileData { profileAllocs = -allocs+                           , profileTime = -t+                           }         updCaller _ old =-          old { profileAllocs = profileAllocs old - allocs }+          old { profileAllocs = profileAllocs old - allocs+              , profileTime = profileTime old - t+              }   -- Like collectProfileData, but intended to be run from the scheduler.@@ -117,45 +173,63 @@   -> Env u w   -> IO (Result u w a) profileCont m env = do+  t0 <- getTimestamp   a0 <- getAllocationCounter   r <- m env   a1 <- getAllocationCounter+  t1 <- getTimestamp   let     allocs = a0 - a1-    newEntry = emptyProfileData { profileAllocs = allocs }-    updEntry _ old = old { profileAllocs = profileAllocs old + allocs }+    t = t0 - t1+    newEntry = emptyProfileData+      { profileAllocs = allocs+      , profileTime = t+      }+    updEntry _ old = old+      { profileAllocs = profileAllocs old + allocs+      , profileTime = profileTime old + t+      }+    profKey = profCurrentKey (profCurrent env)   modifyIORef' (profRef env) $ \ p ->     p { profile =-         HashMap.insertWith updEntry (profLabel env) newEntry $+         HashMap.insertWith updEntry profKey newEntry $          profile p }   -- So we do not count the allocation overhead of modifyProfileData   setAllocationCounter a1   return r {-# INLINE profileCont #-} --incrementMemoHitCounterFor :: ProfileLabel -> Profile -> Profile-incrementMemoHitCounterFor lbl p =-  p { profile = HashMap.adjust incrementMemoHitCounter lbl (profile p) }--incrementMemoHitCounter :: ProfileData -> ProfileData-incrementMemoHitCounter pd = pd { profileMemoHits = succ (profileMemoHits pd) }+incrementMemoHitCounterFor :: Env u w -> CallId -> Bool -> IO ()+incrementMemoHitCounterFor env callId wasCached = do+  modifyIORef' (profRef env) $ \p ->  p {+    profile = HashMap.insertWith+                upd+                (profCurrentKey $ profCurrent env)+                (emptyProfileData { profileMemos = [val] })+                (profile p)+    }+  where+    val = ProfileMemo callId wasCached+    upd _ old = old { profileMemos = val : profileMemos old }  {-# NOINLINE addProfileFetch #-} addProfileFetch   :: forall r u w a . (DataSourceName r, Eq (r a), Hashable (r a), Typeable (r a))-  => Env u w -> r a -> IO ()-addProfileFetch env _req = do+  => Env u w -> r a -> CallId -> Bool -> IO ()+addProfileFetch env _req cid wasCached = do   c <- getAllocationCounter+  let (ProfileCurrent profKey _) = profCurrent env   modifyIORef' (profRef env) $ \ p ->     let-      dsName :: Text-      dsName = dataSourceName (Proxy :: Proxy r)--      upd :: ProfileData -> ProfileData-      upd d = d { profileFetches =-        HashMap.insertWith (+) dsName 1 (profileFetches d) }+      val = ProfileFetch cid (memoKey env) wasCached+      upd _ old = old { profileFetches = val : profileFetches old } -    in p { profile = HashMap.adjust upd (profLabel env) (profile p) }+    in p { profile =+           HashMap.insertWith+             upd+             profKey+             (emptyProfileData { profileFetches = [val] })+             (profile p)+         }   -- So we do not count the allocation overhead of addProfileFetch   setAllocationCounter c

@@ -9,7 +9,6 @@ {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE KindSignatures #-} {-# LANGUAGE BangPatterns #-}-{-# LANGUAGE CPP #-} -- | Bucketing requests by 'DataSource'. -- -- When a request is issued by the client via 'dataFetch', it is placed@@ -22,24 +21,26 @@ -- module Haxl.Core.RequestStore   ( BlockedFetches(..)+  , BlockedFetchInternal(..)   , RequestStore   , isEmpty   , noRequests   , addRequest   , contents+  , getSize   , ReqCountMap(..)   , emptyReqCounts   , filterRCMap   , getMapFromRCMap+  , getSummaryMapFromRCMap   , addToCountMap   , subFromCountMap   ) where -#if __GLASGOW_HASKELL__ <= 708-import Control.Applicative ((<$>))-#endif import Haxl.Core.DataSource+import Haxl.Core.Stats import Data.Map (Map)+import qualified Data.HashMap.Strict as HashMap import qualified Data.Map.Strict as Map import Data.Proxy import Data.Text (Text)@@ -52,9 +53,12 @@   -- is dynamically-typed.  It maps the TypeRep of the request to the   -- 'BlockedFetches' for that 'DataSource'. +newtype BlockedFetchInternal = BlockedFetchInternal CallId+ -- | A batch of 'BlockedFetch' objects for a single 'DataSource' data BlockedFetches u =-  forall r. (DataSource u r) => BlockedFetches [BlockedFetch r]+  forall r. (DataSource u r) =>+        BlockedFetches [BlockedFetch r] [BlockedFetchInternal]  isEmpty :: RequestStore u -> Bool isEmpty (RequestStore m) = Map.null m@@ -66,13 +70,13 @@ -- | Adds a 'BlockedFetch' to a 'RequestStore'. addRequest   :: forall u r. (DataSource u r)-  => BlockedFetch r -> RequestStore u -> RequestStore u-addRequest bf (RequestStore m) =-  RequestStore $ Map.insertWith combine ty (BlockedFetches [bf]) m+  => BlockedFetch r -> BlockedFetchInternal -> RequestStore u -> RequestStore u+addRequest bf bfi (RequestStore m) =+  RequestStore $ Map.insertWith combine ty (BlockedFetches [bf] [bfi]) m  where   combine :: BlockedFetches u -> BlockedFetches u -> BlockedFetches u-  combine _ (BlockedFetches bfs)-    | typeOf1 (getR bfs) == ty = BlockedFetches (unsafeCoerce bf:bfs)+  combine _ (BlockedFetches bfs bfis)+    | typeOf1 (getR bfs) == ty = BlockedFetches (unsafeCoerce bf:bfs) (bfi:bfis)     | otherwise                = error "RequestStore.insert"          -- the dynamic type check here should be unnecessary, but if          -- there are bugs in `Typeable` or `Map` then we'll get an@@ -91,6 +95,9 @@ contents :: RequestStore u -> [BlockedFetches u] contents (RequestStore m) = Map.elems m +getSize :: RequestStore u -> Int+getSize (RequestStore m) = Map.size m+ -- A counter to keep track of outgone requests. Entries are added to this -- map as we send requests to datasources, and removed as these fetches -- are completed.@@ -149,3 +156,12 @@ getMapFromRCMap :: ReqCountMap -> Map Text (Map TypeRep Int) getMapFromRCMap r   | ReqCountMap m <- filterRCMap r = m++getSummaryMapFromRCMap :: ReqCountMap -> HashMap.HashMap Text Int+getSummaryMapFromRCMap (ReqCountMap m) = HashMap.fromList+  [ (k, s)+  | (k, v) <- Map.toList m+  , not $ Map.null v+  , let s = sum $ Map.elems v+  , s > 0+  ]

@@ -4,11 +4,12 @@ -- This source code is distributed under the terms of a BSD license, -- found in the LICENSE file. -{-# LANGUAGE CPP #-} {-# LANGUAGE BangPatterns #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE DeriveDataTypeable #-}  -- | Defines 'runHaxl'.  Most users should import "Haxl.Core" instead. --@@ -17,9 +18,6 @@   , runHaxlWithWrites   ) where -#if __GLASGOW_HASKELL__ <= 708-import Control.Applicative ((<$>))-#endif import Control.Concurrent.STM import Control.Exception as Exception import Control.Monad@@ -35,7 +33,9 @@ import Haxl.Core.Profile import Haxl.Core.RequestStore as RequestStore import Haxl.Core.Stats+import Haxl.Core.Util +import qualified Data.HashTable.IO as H  -- ----------------------------------------------------------------------------- -- runHaxl@@ -54,17 +54,31 @@  runHaxlWithWrites :: forall u w a. Env u w -> GenHaxl u w a -> IO (a, [w]) runHaxlWithWrites env@Env{..} haxl = do-  result@(IVar resultRef) <- newIVar -- where to put the final result+  result@IVar{ivarRef = resultRef} <- newIVar -- where to put the final result+  ifTraceLog <- do+    if trace flags < 3+    then return $ \_ -> return ()+    else do+      start <- getTimestamp+      return $ \s -> do+        now <- getTimestamp+        let t = fromIntegral (now - start) / 1000.0 :: Double+        printf "%.1fms: %s" t (s :: String)   let     -- Run a job, and put its result in the given IVar     schedule :: Env u w -> JobList u w -> GenHaxl u w b -> IVar u w b -> IO ()-    schedule env@Env{..} rq (GenHaxl run) (IVar !ref) = do-      ifTrace flags 3 $ printf "schedule: %d\n" (1 + lengthJobList rq)+    schedule env@Env{..} rq (GenHaxl run) ivar@IVar{ivarRef = !ref} = do+      ifTraceLog $ printf "schedule: %d\n" (1 + lengthJobList rq)       let {-# INLINE result #-}           result r = do             e <- readIORef ref             case e of-              IVarFull _ -> error "multiple put"+              IVarFull _ ->+                -- An IVar is typically only meant to be written to once+                -- so it would make sense to throw an error here. But there+                -- are legitimate use-cases for writing several times.+                -- (See Haxl.Core.Parallel)+                reschedule env rq               IVarEmpty haxls -> do                 writeIORef ref (IVarFull r)                 -- Have we got the final result now?@@ -95,8 +109,8 @@         Right (Throw ex) -> do           wt <- readIORef writeLogsRef           result (ThrowHaxl ex wt)-        Right (Blocked ivar fn) -> do-          addJob env (toHaxl fn) (IVar ref) ivar+        Right (Blocked i fn) -> do+          addJob env (toHaxl fn) ivar i           reschedule env rq      -- Here we have a choice:@@ -129,54 +143,50 @@      emptyRunQueue :: Env u w -> IO ()     emptyRunQueue env@Env{..} = do-      ifTrace flags 3 $ printf "emptyRunQueue\n"+      ifTraceLog $ printf "emptyRunQueue\n"       haxls <- checkCompletions env       case haxls of-        JobNil -> do-          case pendingWaits of-            [] -> checkRequestStore env-            wait:waits -> do-              ifTrace flags 3 $ printf "invoking wait\n"-              wait-              emptyRunQueue env { pendingWaits = waits } -- check completions+        JobNil -> checkRequestStore env         _ -> reschedule env haxls      checkRequestStore :: Env u w -> IO ()     checkRequestStore env@Env{..} = do+      ifTraceLog $ printf "checkRequestStore\n"       reqStore <- readIORef reqStoreRef       if RequestStore.isEmpty reqStore         then waitCompletions env         else do+          ifTraceLog $ printf "performFetches %d\n" (RequestStore.getSize reqStore)           writeIORef reqStoreRef noRequests-          (_, waits) <- performRequestStore 0 env reqStore-          ifTrace flags 3 $ printf "performFetches: %d waits\n" (length waits)+          performRequestStore env reqStore           -- empty the cache if we're not caching.  Is this the best           -- place to do it?  We do get to de-duplicate requests that           -- happen simultaneously.-          when (caching flags == 0) $-            writeIORef cacheRef emptyDataCache-          emptyRunQueue env{ pendingWaits = waits ++ pendingWaits }+          when (caching flags == 0) $ do+            let DataCache dc = dataCache+            H.foldM (\_ (k, _) -> H.delete dc k) () dc+          emptyRunQueue env      checkCompletions :: Env u w -> IO (JobList u w)     checkCompletions Env{..} = do-      ifTrace flags 3 $ printf "checkCompletions\n"-      comps <- atomically $ do+      ifTraceLog $ printf "checkCompletions\n"+      comps <- atomicallyOnBlocking (LogicBug ReadingCompletionsFailedRun) $ do         c <- readTVar completions         writeTVar completions []         return c       case comps of         [] -> return JobNil         _ -> do-          ifTrace flags 3 $ printf "%d complete\n" (length comps)+          ifTraceLog $ printf "%d complete\n" (length comps)           let-              getComplete (CompleteReq a (IVar cr) allocs) = do+              getComplete (CompleteReq a IVar{ivarRef = !cr} allocs) = do                 when (allocs < 0) $ do                   cur <- getAllocationCounter                   setAllocationCounter (cur + allocs)                 r <- readIORef cr                 case r of                   IVarFull _ -> do-                    ifTrace flags 3 $ printf "existing result\n"+                    ifTraceLog $ printf "existing result\n"                     return JobNil                     -- this happens if a data source reports a result,                     -- and then throws an exception.  We call putResult@@ -192,10 +202,36 @@      waitCompletions :: Env u w -> IO ()     waitCompletions env@Env{..} = do-      ifTrace flags 3 $ printf "waitCompletions\n"-      atomically $ do-        c <- readTVar completions-        when (null c) retry+      ifTraceLog $ printf "waitCompletions\n"+      let+        wrapped = atomicallyOnBlocking (LogicBug ReadingCompletionsFailedRun)+        doWait = wrapped $ do+          c <- readTVar completions+          when (null c) retry+        doWaitProfiled = do+          queueEmpty <- null <$> wrapped (readTVar completions)+          when queueEmpty $ do+            -- Double check the queue as we want to make sure that+            -- submittedReqsRef is copied before waiting on the queue but as a+            -- fast path do not want to copy it if the queue is empty.+            -- There is still a race oppoortunity as submittedReqsRef is+            -- decremented in whatever thread the completion happens, and so it+            -- is possible for waitingOn to be empty while queueEmpty2 is True.+            waitingOn <- readIORef submittedReqsRef+            queueEmpty2 <- null <$> wrapped (readTVar completions)+            when queueEmpty2 $ do+              start <- getTimestamp+              doWait+              end <- getTimestamp+              let fw = FetchWait+                        { fetchWaitReqs = getSummaryMapFromRCMap waitingOn+                        , fetchWaitStart = start+                        , fetchWaitDuration = (end-start)+                        }+              modifyIORef' statsRef $ \(Stats s) -> Stats (fw:s)+      if report flags >= 2+        then doWaitProfiled+        else doWait       emptyRunQueue env    --@@ -203,7 +239,9 @@   r <- readIORef resultRef   case r of     IVarEmpty _ -> throwIO (CriticalError "runHaxl: missing result")-    IVarFull (Ok a wt)  -> return (a, flattenWT wt)+    IVarFull (Ok a wt) -> do+      wtNoMemo <- readIORef writeLogsRefNoMemo+      return (a, flattenWT (wt `appendWTs` wtNoMemo))     IVarFull (ThrowHaxl e _wt)  -> throwIO e       -- The written logs are discarded when there's a Haxl exception. We       -- can change this behavior if we need to get access to partial logs.@@ -213,7 +251,7 @@ {- Note [runHaxl and unfinished requests]  runHaxl returns immediately when the supplied computation has returned-a result.  This doesn't necessarily mean that the whole computaiton+a result.  This doesn't necessarily mean that the whole computation graph has completed, however.  In particular, when using pAnd and pOr, we might have created some data fetches that have not completed, but weren't required, because the other branch of the pAnd/pOr subsumed@@ -239,3 +277,9 @@   | WaitAtLeast Int{-ms-}   | WaitForAllPendingRequests -}++-- | An exception thrown when reading from datasources fails+data ReadingCompletionsFailedRun = ReadingCompletionsFailedRun+  deriving Show++instance Exception ReadingCompletionsFailedRun

@@ -34,11 +34,7 @@ -- instance of 'StateKey' can store and retrieve information from a -- 'StateStore'. ---#if __GLASGOW_HASKELL__ >= 708 class Typeable f => StateKey (f :: * -> *) where-#else-class Typeable1 f => StateKey (f :: * -> *) where-#endif   data State f    -- | We default this to typeOf1, but if f is itself a complex type that is

@@ -4,10 +4,11 @@ -- This source code is distributed under the terms of a BSD license, -- found in the LICENSE file. -{-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE CPP #-}  -- | -- Types and operations for statistics and profiling.  Most users@@ -18,6 +19,7 @@   (   -- * Data-source stats     Stats(..)+  , CallId   , FetchStats(..)   , Microseconds   , Timestamp@@ -26,16 +28,19 @@   , numFetches   , ppStats   , ppFetchStats+  , aggregateFetchBatches    -- * Profiling-  , Profile+  , Profile(..)+  , ProfileMemo(..)+  , ProfileFetch(..)   , emptyProfile-  , profile+  , ProfileKey   , ProfileLabel   , ProfileData(..)   , emptyProfileData   , AllocCount-  , MemoHitCount+  , LabelHitCount    -- * Allocation   , getAllocationCounter@@ -43,25 +48,22 @@   ) where  import Data.Aeson+import Data.Function (on)+import Data.Maybe (mapMaybe) import Data.HashMap.Strict (HashMap)-import Data.HashSet (HashSet) import Data.Int-import Data.List (intercalate, maximumBy, minimumBy)-#if __GLASGOW_HASKELL__ < 710-import Data.Monoid-#endif+import Data.List (intercalate, sortOn, groupBy)+#if __GLASGOW_HASKELL__ < 804 import Data.Semigroup (Semigroup)-import Data.Ord (comparing)+#endif+import Data.Ord (Down(..)) import Data.Text (Text) import Data.Time.Clock.POSIX import Text.Printf import qualified Data.HashMap.Strict as HashMap-import qualified Data.HashSet as HashSet import qualified Data.Text as Text -#if __GLASGOW_HASKELL__ >= 710 import GHC.Conc (getAllocationCounter, setAllocationCounter)-#endif  -- --------------------------------------------------------------------------- -- Measuring time@@ -90,7 +92,7 @@       if fetchWasRunning rs           (minStartTime + (t - 1) * usPerDash)           (minStartTime + t * usPerDash)-        then '*'+        then fetchSymbol rs         else '-'       | t <- [1..numDashes]       ]@@ -98,18 +100,30 @@     | (i, rs) <- zip [(1::Int)..] validFetchStats ]   where     isFetchStats FetchStats{} = True+    isFetchStats FetchWait{} = True     isFetchStats _ = False     validFetchStats = filter isFetchStats (reverse rss)     numDashes = 50-    minStartTime = fetchStart $ minimumBy (comparing fetchStart) validFetchStats-    lastFs = maximumBy (comparing (\fs -> fetchStart fs + fetchDuration fs))-      validFetchStats-    usPerDash = (fetchStart lastFs + fetchDuration lastFs - minStartTime)-      `div` numDashes+    getStart FetchStats{..} = Just fetchStart+    getStart FetchWait{..} = Just fetchWaitStart+    getStart _ = Nothing+    getEnd FetchStats{..} = Just $ fetchStart + fetchDuration+    getEnd FetchWait{..} = Just $ fetchWaitStart + fetchWaitDuration+    getEnd _ = Nothing+    minStartTime = minimum $ mapMaybe getStart validFetchStats+    endTime = maximum $ mapMaybe getEnd validFetchStats+    usPerDash = (endTime - minStartTime) `div` numDashes+    fetchSymbol FetchStats{} = '*'+    fetchSymbol FetchWait{} = '.'+    fetchSymbol _ = '?'     fetchWasRunning :: FetchStats -> Timestamp -> Timestamp -> Bool-    fetchWasRunning fs t1 t2 =+    fetchWasRunning fs@FetchStats{} t1 t2 =       (fetchStart fs + fetchDuration fs) >= t1 && fetchStart fs < t2+    fetchWasRunning fw@FetchWait{} t1 t2 =+      (fetchWaitStart fw + fetchWaitDuration fw) >= t1 && fetchWaitStart fw < t2+    fetchWasRunning _ _ _ = False +type CallId = Int  -- | Maps data source name to the number of requests made in that round. -- The map only contains entries for sources that made requests in that@@ -119,10 +133,13 @@   = FetchStats     { fetchDataSource :: Text     , fetchBatchSize :: {-# UNPACK #-} !Int-    , fetchStart :: !Timestamp          -- TODO should be something else+    , fetchStart :: {-# UNPACK #-} !Timestamp     , fetchDuration :: {-# UNPACK #-} !Microseconds     , fetchSpace :: {-# UNPACK #-} !Int64     , fetchFailures :: {-# UNPACK #-} !Int+    , fetchIgnoredFailures :: {-# UNPACK #-} !Int+    , fetchBatchId :: {-# UNPACK #-} !Int+    , fetchIds :: [CallId]     }      -- | The stack trace of a call to 'dataFetch'.  These are collected@@ -130,8 +147,19 @@   | FetchCall     { fetchReq :: String     , fetchStack :: [String]+    , fetchStatId :: {-# UNPACK #-} !CallId     }-  deriving (Show)+  | MemoCall+    { memoStatId :: {-# UNPACK #-} !CallId+    , memoSpace :: {-# UNPACK #-} !Int64+    }+  | FetchWait+    { fetchWaitReqs :: HashMap Text Int+       -- ^ What DataSources had requests that were being waited for+    , fetchWaitStart :: {-# UNPACK #-} !Timestamp+    , fetchWaitDuration :: {-# UNPACK #-} !Microseconds+    }+  deriving (Eq, Show)  -- | Pretty-print RoundStats. ppFetchStats :: FetchStats -> String@@ -139,21 +167,59 @@   printf "%s: %d fetches (%.2fms, %d bytes, %d failures)"     (Text.unpack fetchDataSource) fetchBatchSize     (fromIntegral fetchDuration / 1000 :: Double)  fetchSpace fetchFailures-ppFetchStats (FetchCall r ss) = show r ++ '\n':show ss+ppFetchStats (FetchCall r ss _) = show r ++ '\n':show ss+ppFetchStats MemoCall{} = ""+ppFetchStats FetchWait{..}+  | HashMap.size fetchWaitReqs == 0 = msg "unexpected: Blocked on nothing"+  | HashMap.size fetchWaitReqs <= 2 =+    msg $ printf "Blocked on %s"+      (intercalate "," [printf "%s (%d reqs)" ds c+                       | (ds,c) <- HashMap.toList fetchWaitReqs])+  | otherwise = msg $ printf "Blocked on %d sources (%d reqs)"+                        (HashMap.size fetchWaitReqs)+                        (sum $ HashMap.elems fetchWaitReqs)+  where+    msg :: String -> String+    msg x = printf "%s (%.2fms)"+                x+                (fromIntegral fetchWaitDuration / 1000 :: Double) +-- | Aggregate stats merging FetchStats from the same dispatched batch into one.+aggregateFetchBatches :: ([FetchStats] -> a) -> Stats -> [a]+aggregateFetchBatches agg (Stats fetches) =+      map agg $+      groupBy ((==) `on` fetchBatchId) $+      sortOn (Down . fetchBatchId)+      [f | f@FetchStats{} <- fetches]+ instance ToJSON FetchStats where   toJSON FetchStats{..} = object-    [ "datasource" .= fetchDataSource+    [ "type" .= ("FetchStats" :: Text)+    , "datasource" .= fetchDataSource     , "fetches" .= fetchBatchSize     , "start" .= fetchStart     , "duration" .= fetchDuration     , "allocation" .= fetchSpace     , "failures" .= fetchFailures+    , "ignoredFailures" .= fetchIgnoredFailures+    , "batchid" .= fetchBatchId+    , "fetchids" .= fetchIds     ]-  toJSON (FetchCall req strs) = object-    [ "request" .= req+  toJSON (FetchCall req strs fid) = object+    [ "type" .= ("FetchCall" :: Text)+    , "request" .= req     , "stack" .= strs+    , "fetchid" .= fid     ]+  toJSON (MemoCall cid allocs) = object+    [ "type" .= ("MemoCall" :: Text)+    , "callid" .= cid+    , "allocation" .= allocs+    ]+  toJSON FetchWait{..} = object+    [ "type" .= ("FetchWait" :: Text)+    , "duration" .= fetchWaitDuration+    ]  emptyStats :: Stats emptyStats = Stats []@@ -167,39 +233,48 @@  type ProfileLabel = Text type AllocCount = Int64-type MemoHitCount = Int64+type LabelHitCount = Int64+type ProfileKey = Int64 -newtype Profile = Profile-  { profile      :: HashMap ProfileLabel ProfileData-     -- ^ Data on individual labels.+data ProfileFetch = ProfileFetch+  { profileFetchFetchId :: {-# UNPACK #-} !CallId+  , profileFetchMemoId ::  {-# UNPACK #-} !CallId+  , profileFetchWasCached :: !Bool   }+  deriving (Show, Eq) +data ProfileMemo = ProfileMemo+  { profileMemoId :: {-# UNPACK #-} !CallId+  , profileMemoWasCached :: !Bool+  }+  deriving (Show, Eq)++data Profile = Profile+  { profile      :: HashMap ProfileKey ProfileData+     -- ^ Data per key (essentially per call stack)+  , profileTree :: HashMap (ProfileLabel, ProfileKey) ProfileKey+     -- ^ (label, parent) -> current. The exception is the root which will have+     -- ("MAIN", 0) -> 0+  , profileNextKey :: ProfileKey+     -- ^ Provides a unique key per callstack+  }+ emptyProfile :: Profile-emptyProfile = Profile HashMap.empty+emptyProfile = Profile HashMap.empty (HashMap.singleton ("MAIN", 0) 0) 1  data ProfileData = ProfileData   { profileAllocs :: {-# UNPACK #-} !AllocCount      -- ^ allocations made by this label-  , profileDeps :: HashSet ProfileLabel-     -- ^ labels that this label depends on-  , profileFetches :: HashMap Text Int-     -- ^ map from datasource name => fetch count-  , profileMemoHits :: {-# UNPACK #-} !MemoHitCount-    -- ^ number of hits to memoized computation at this label+  , profileFetches :: [ProfileFetch]+     -- ^ fetches made in this label+  , profileLabelHits :: {-# UNPACK #-} !LabelHitCount+     -- ^ number of hits at this label+  , profileMemos :: [ProfileMemo]+     -- ^ memo and a boolean representing if it was cached at the time+  , profileTime :: {-# UNPACK #-} !Microseconds+     -- ^ amount of time spent in computation at this label   }   deriving Show  emptyProfileData :: ProfileData-emptyProfileData = ProfileData 0 HashSet.empty HashMap.empty 0----- -------------------------------------------------------------------------------- Allocation accounting--#if __GLASGOW_HASKELL__ < 710-getAllocationCounter :: IO Int64-getAllocationCounter = return 0--setAllocationCounter :: Int64 -> IO ()-setAllocationCounter _ = return ()-#endif+emptyProfileData = ProfileData 0 [] 0 [] 0

@@ -6,18 +6,35 @@  -- | Internal utilities only. --+{-# OPTIONS_GHC -fno-warn-unused-imports #-} module Haxl.Core.Util-  ( compose+  ( atomicallyOnBlocking+  , compose   , textShow+  , trace_   ) where  import Data.Text (Text)-+import Debug.Trace (trace) import qualified Data.Text as Text +import Control.Concurrent.STM+import Control.Exception++atomicallyOnBlocking :: Exception e => e -> STM a -> IO a+atomicallyOnBlocking e stm =+  catch (atomically stm)+        (\BlockedIndefinitelyOnSTM -> throw e)+ -- | Composes a list of endofunctions. compose :: [a -> a] -> a -> a compose = foldr (.) id  textShow :: (Show a) => a -> Text textShow = Text.pack . show++-- | This function can be used to trace a bunch of lines to stdout when+-- debugging haxl core.+trace_ :: String -> a -> a+trace_ _ = id+--trace_ = Debug.Trace.trace

@@ -49,7 +49,6 @@   ) where  import Control.Concurrent-import Control.Exception as Exception import Control.Monad import qualified Data.Text as Text import Data.Typeable@@ -79,5 +78,4 @@  where   fetch _state _flags _u = BackgroundFetch $ \bfs -> do     forM_ bfs $ \(BlockedFetch req rv) ->-      mask $ \unmask ->-        forkFinally (unmask (performIO req)) (putResultFromChildThread rv)+      forkFinally (performIO req) (putResultFromChildThread rv)

@@ -4,7 +4,6 @@ -- This source code is distributed under the terms of a BSD license, -- found in the LICENSE file. -{-# LANGUAGE CPP #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE ScopedTypeVariables #-}@@ -36,14 +35,12 @@      -- * Extra Monad and Applicative things     Applicative(..),-#if __GLASGOW_HASKELL__ < 710-    (<$>),-#endif     mapM, mapM_, sequence, sequence_, filterM, foldM,     forM, forM_,     foldl', sort,     Monoid(..),     join,+    andThen,      -- * Lifted operations     IfThenElse(..),@@ -79,9 +76,6 @@ import Data.Traversable hiding (forM, mapM, sequence) import GHC.Exts (IsString(..)) import Prelude hiding (mapM, mapM_, sequence, sequence_)-#if __GLASGOW_HASKELL__ < 710-import Data.Monoid-#endif import Data.Maybe import Control.Exception (fromException) @@ -196,6 +190,25 @@     filt <$> mapM predicate xs   where     filt bools = [ x | (x,True) <- zip xs bools ]++-- | In somes cases, we do want the monadic version of @('>>')@ to disable+-- concurrency and start one computation only after the other finishes, e.g.:+--+-- @+-- deferedFetch x = do+--   sleep 5+--   fetch x  -- fetch will actually run concurrently with sleep+-- @+--+-- But we have defined @('>>') = ('*>')@ with the applicative behavior as this+-- is desired in most cases, so instead we define 'andThen' as the monadic+-- version of @('>>')@:+--+-- @+-- deferedFetch x = sleep 5 `andThen` fetch x+-- @+andThen :: Monad m => m a -> m b -> m b+andThen a b = a >>= \_ -> b  -------------------------------------------------------------------------------- 

@@ -1,3 +1,16 @@+# Changes in version <next>+  * Added fetchBatchId to FetchStats+  * Profiling now tracks full stacks and links each label to memos/fetches++# Changes in version 2.3.0.0+  * Removed `FutureFetch`++# Changes in version 2.2.0.0++  * Use BasicHashTable for the Haxl DataCache instead of HashMap+  * API Changes in: Haxl.Core.DataCache, Haxl.Core.Fetch+  * Removed support for GHC < 8.2+ # Changes in version 2.1.2.0    * Add a callgraph reference to 'Env' to record the function callgraph during a@@ -13,7 +26,7 @@    * Add a new 'w' parameter to 'GenHaxl' to allow arbitrary writes during     a computation. These writes are stored as a running log in the Env,-    and are not memoized. This allows users to extract information from+    and are memoized. This allows users to extract information from     a Haxl computation which throws. Our advise is to limit these writes to     monitoring and debugging logs. 

@@ -1,5 +1,5 @@ name:                haxl-version:             2.1.2.0+version:             2.3.0.0 synopsis:            A Haskell library for efficient, concurrent,                      and concise data access. homepage:            https://github.com/facebook/Haxl@@ -14,11 +14,11 @@ stability:           alpha cabal-version:       >= 1.10 tested-with:-  GHC==7.8.4,-  GHC==7.10.3,-  GHC==8.0.2,   GHC==8.2.2,-  GHC==8.4.3+  GHC==8.4.4,+  GHC==8.6.5,+  GHC==8.8.3,+  GHC==8.10.1  description:   Haxl is a library and EDSL for efficient scheduling of concurrent data@@ -43,8 +43,8 @@ library    build-depends:-    aeson >= 0.6 && < 1.5,-    base >= 4.7 && < 5,+    aeson >= 0.6 && < 1.6,+    base >= 4.10 && < 5,     binary >= 0.7 && < 0.10,     bytestring >= 0.9 && < 0.11,     containers >= 0.5 && < 0.7,@@ -52,20 +52,17 @@     exceptions >=0.8 && <0.11,     filepath >= 1.3 && < 1.5,     ghc-prim,-    hashable == 1.2.*,+    hashable >= 1.2 && < 1.4,+    hashtables >= 1.2.3.1,     pretty == 1.1.*,     -- text 1.2.1.0 required for instance Binary Text     text >= 1.2.1.0 && < 1.3,-    time >= 1.4 && < 1.9,+    time >= 1.4 && < 1.10,     stm >= 2.4 && < 2.6,     transformers,     unordered-containers == 0.2.*,     vector >= 0.10 && <0.13 -  if !impl(ghc >= 8.0)-    build-depends:-      semigroups >= 0.18.3 && <0.19-   exposed-modules:     Haxl.Core,     Haxl.Core.CallGraph,@@ -94,11 +91,7 @@   ghc-options:     -O2 -fprof-auto     -Wall--  if impl(ghc >= 8.0)-     ghc-options: -Wno-name-shadowing-  else-     ghc-options: -fno-warn-name-shadowing+    -Wno-name-shadowing  test-suite test @@ -112,6 +105,7 @@     deepseq,     filepath,     hashable,+    hashtables,     haxl,     test-framework,     test-framework-hunit,@@ -130,12 +124,8 @@   main-is:     TestMain.hs -  if impl(ghc >= 8.0)-    other-modules:-      AdoTests-      OutgoneFetchesTests-   other-modules:+    AdoTests     AllTests     BadDataSource     BatchTests@@ -146,8 +136,12 @@     LoadCache     MemoizationTests     MockTAO+    MonadAsyncTest+    OutgoneFetchesTests+    ParallelTests     ProfileTests     SleepDataSource+    StatsTests     TestBadDataSource     TestExampleDataSource     TestTypes@@ -174,7 +168,8 @@     base,     haxl,     hashable,-    time+    time,+    optparse-applicative   main-is:     MonadBench.hs   other-modules:@@ -193,6 +188,7 @@     base,     haxl,     hashable,+    hashtables,     time   main-is:     Bench.hs

@@ -57,3 +57,11 @@    Hackage.   * [There is no Fork: An Abstraction for Efficient, Concurrent, and Concise Data Access](http://simonmar.github.io/bib/papers/haxl-icfp14.pdf), our paper on Haxl, accepted for publication at ICFP'14.++## Contributing++We welcome contributions! See [CONTRIBUTING](https://github.com/facebook/Haxl/blob/master/CONTRIBUTING.md) for details on how to get started, and our [Code of Conduct](https://github.com/facebook/Haxl/blob/master/CODE_OF_CONDUCT.md).++## License++Haxl uses the BSD 3-clause License, as found in the [LICENSE](https://github.com/facebook/Haxl/blob/master/LICENSE) file.

@@ -4,24 +4,23 @@ -- This source code is distributed under the terms of a BSD license, -- found in the LICENSE file. -{-# LANGUAGE CPP, OverloadedStrings #-}+{-# LANGUAGE OverloadedStrings #-} module AllTests (allTests) where  import TestExampleDataSource import BatchTests import CoreTests import DataCacheTest-#if __GLASGOW_HASKELL__ >= 801 import AdoTests import OutgoneFetchesTests-#endif-#if __GLASGOW_HASKELL__ >= 710 import ProfileTests-#endif import MemoizationTests+import MonadAsyncTest import TestBadDataSource import FullyAsyncTest import WriteTests+import ParallelTests+import StatsTests  import Test.HUnit @@ -32,15 +31,14 @@   , TestLabel "BatchTests-sync" $ BatchTests.tests False   , TestLabel "CoreTests" CoreTests.tests   , TestLabel "DataCacheTests" DataCacheTest.tests-#if __GLASGOW_HASKELL__ >= 801   , TestLabel "AdoTests" $ AdoTests.tests False   , TestLabel "OutgoneFetchesTest" OutgoneFetchesTests.tests-#endif-#if __GLASGOW_HASKELL__ >= 710   , TestLabel "ProfileTests" ProfileTests.tests-#endif   , TestLabel "MemoizationTests" MemoizationTests.tests+  , TestLabel "MonadAsyncTests" MonadAsyncTest.tests   , TestLabel "BadDataSourceTests" TestBadDataSource.tests   , TestLabel "FullyAsyncTest" FullyAsyncTest.tests   , TestLabel "WriteTest" WriteTests.tests+  , TestLabel "ParallelTest" ParallelTests.tests+  , TestLabel "StatsTests" StatsTests.tests   ]

@@ -18,10 +18,10 @@  module BadDataSource (     -- * initialise the state-    State(..), initGlobalState,+    State(..), initGlobalState, FetchImpl(..),      -- * requests for this data source-    FailAfter(..),+    FailAfter(..)   ) where  import Haxl.Prelude@@ -33,7 +33,22 @@ import Data.Typeable import Data.Hashable import Control.Concurrent+import Control.Monad (void) +import GHC.Conc ( PrimMVar )+import Foreign.StablePtr+import Foreign.C.Types ( CInt(..) )++foreign import ccall safe+  hs_try_putmvar :: CInt -> StablePtr PrimMVar -> IO ()++data FetchImpl =+  Async+  | Background+  | BackgroundMVar+  | BackgroundSeq+  | BackgroundPar+ data FailAfter a where   FailAfter :: Int -> FailAfter Int   deriving Typeable@@ -55,32 +70,53 @@          , failDispatch :: IO ()          , failWaitDelay :: Int          , failWait :: IO ()+         , failImpl :: FetchImpl         }  instance DataSourceName FailAfter where   dataSourceName _ = "BadDataSource" + instance DataSource u FailAfter where-  -- I'll define exampleFetch below-  fetch state@FailAfterState{..} = asyncFetchAcquireRelease-     (do threadDelay failAcquireDelay; failAcquire)-     (\_ -> do threadDelay failReleaseDelay; failRelease)-     (\_ -> do threadDelay failDispatchDelay; failDispatch)-     (\_ -> do threadDelay failWaitDelay; failWait)-     submit state+  fetch state@FailAfterState{..}+    | BackgroundSeq <- failImpl = backgroundFetchSeq runOne state+    | BackgroundPar <- failImpl = backgroundFetchPar runOne state+    | Background <- failImpl = backgroundFetchAcquireRelease+        acquire release dispatchbg wait+        submit state+    | BackgroundMVar <- failImpl = backgroundFetchAcquireReleaseMVar+        acquire release dispatchbgMVar wait+        submit state+    | Async <- failImpl = asyncFetchAcquireRelease+       acquire release dispatch wait+       submit state    where+     acquire = do threadDelay failAcquireDelay; failAcquire+     release _ = do threadDelay failReleaseDelay; failRelease+     dispatch _ = do threadDelay failDispatchDelay; failDispatch+     dispatchBase put = (do+                          failDispatch+                          _ <- mask_ $ forkIO $ finally+                            (threadDelay failDispatchDelay)+                            put+                          return ()) `onException` put+     dispatchbg _ c m = dispatchBase (hs_try_putmvar (fromIntegral c) m)+     dispatchbgMVar _ _ m = dispatchBase (void $ tryPutMVar m ())+     wait _ = do threadDelay failWaitDelay; failWait      submit :: () -> FailAfter a -> IO (IO (Either SomeException a))      submit _ (FailAfter t) = do        threadDelay t        return (return (Left (toException (FetchError "failed request"))))+     runOne :: FailAfter a -> IO (Either SomeException a)+     runOne r = do+       bracket acquire release $ \s -> do+         dispatch s+         getRes <- submit s r+         wait s+         getRes --- Every data source should define a function 'initGlobalState' that--- initialises the state for that data source.  The arguments to this--- function might vary depending on the data source - we might need to--- pass in resources from the environment, or parameters to set up the--- data source.-initGlobalState :: IO (State FailAfter)-initGlobalState = do+initGlobalState :: FetchImpl -> IO (State FailAfter)+initGlobalState impl = do   return FailAfterState     { failAcquireDelay = 0     , failAcquire = return ()@@ -90,4 +126,5 @@     , failDispatch = return ()     , failWaitDelay = 0     , failWait = return ()+    , failImpl = impl     }

@@ -141,6 +141,8 @@   tao <- MockTAO.initGlobalState future   let st = stateSet tao stateEmpty   env2 <- initEnvWithData st testinput (caches env)+  cid <- readIORef (callIdRef env2)+  assertBool "callId is unique" (cid > 0)    -- ensure no more data fetching rounds needed   expectResultWithEnv 12 batching7_ env2

@@ -39,14 +39,16 @@ main = do   [n] <- fmap (fmap read) getArgs   t0 <- getCurrentTime+  cache <- emptyDataCache   let-     f 0  !cache = return cache-     f !n !cache = do+     f 0 = return ()+     f !n = do        m <- newIORef 0-       f (n-1) (DataCache.insert (ReqInt n) m cache)+       DataCache.insert (ReqInt n) m cache+       f (n-1)   ---  cache <- f n emptyDataCache-  let m = DataCache.lookup (ReqInt (n `div` 2)) cache+  f n+  m <- DataCache.lookup (ReqInt (n `div` 2)) cache   print =<< mapM readIORef m   t1 <- getCurrentTime   printf "insert: %.2fs\n" (realToFrac (t1 `diffUTCTime` t0) :: Double)@@ -54,9 +56,11 @@   t0 <- getCurrentTime   let      f 0  !m = return m-     f !n !m = case DataCache.lookup (ReqInt n) cache of-                 Nothing -> f (n-1) m-                 Just _  -> f (n-1) (m+1)+     f !n !m = do+      mbRes <- DataCache.lookup (ReqInt n) cache+      case mbRes of+        Nothing -> f (n-1) m+        Just _  -> f (n-1) (m+1)   f n 0 >>= print   t1 <- getCurrentTime   printf "lookup: %.2fs\n" (realToFrac (t1 `diffUTCTime` t0) :: Double)

@@ -4,16 +4,13 @@ -- This source code is distributed under the terms of a BSD license, -- found in the LICENSE file. -{-# LANGUAGE CPP, StandaloneDeriving, GADTs, DeriveDataTypeable #-}+{-# LANGUAGE StandaloneDeriving, GADTs, DeriveDataTypeable #-} module DataCacheTest (tests, newResult, takeResult) where  import Haxl.Core.DataCache as DataCache import Haxl.Core.Monad import Haxl.Core -#if __GLASGOW_HASKELL__ < 710-import Control.Applicative-#endif import Control.Exception import Data.Hashable import Data.Traversable@@ -47,32 +44,31 @@ dcSoundnessTest = TestLabel "DataCache soundness" $ TestCase $ do   m1 <- newResult 1   m2 <- newResult "hello"-  let cache =-          DataCache.insert (Req 1 :: TestReq Int) m1 $-          DataCache.insert (Req 2 :: TestReq String) m2 $-          emptyDataCache+  cache <- emptyDataCache+  DataCache.insert (Req 2 :: TestReq String) m2 cache+  DataCache.insert (Req 1 :: TestReq Int) m1 cache    -- "Req 1" has a result of type Int, so if we try to look it up   -- with a result of type String, we should get Nothing, not a crash.-  r <- mapM takeResult $ DataCache.lookup (Req 1) cache+  r <- mapM takeResult =<< DataCache.lookup (Req 1) cache   assertBool "dcSoundness1" $     case r :: Maybe (ResultVal String ()) of      Nothing -> True      _something_else -> False -  r <- mapM takeResult $ DataCache.lookup (Req 1) cache+  r <- mapM takeResult =<< DataCache.lookup (Req 1) cache   assertBool "dcSoundness2" $     case r :: Maybe (ResultVal Int ()) of      Just (Ok 1 NilWrites) -> True      _something_else -> False -  r <- mapM takeResult $ DataCache.lookup (Req 2) cache+  r <- mapM takeResult =<< DataCache.lookup (Req 2) cache   assertBool "dcSoundness3" $     case r :: Maybe (ResultVal String ()) of       Just (Ok "hello" NilWrites) -> True       _something_else -> False -  r <- mapM takeResult $ DataCache.lookup (Req 2) cache+  r <- mapM takeResult =<< DataCache.lookup (Req 2) cache   assertBool "dcSoundness4" $     case r :: Maybe (ResultVal Int ()) of       Nothing -> True

@@ -31,6 +31,7 @@ import Data.Typeable import Data.Hashable import Control.Concurrent+import qualified Control.Exception as E import System.IO  -- Here is an example minimal data source.  Our data source will have@@ -95,7 +96,12 @@   -- I'll define exampleFetch below   fetch = exampleFetch +  -- we don't want to treat NotFound as an exception for stats purposes+  classifyFailure _ _ e+    | Just NotFound{} <- E.fromException e = IgnoredForStatsFailure+    | otherwise = StandardFailure + -- Every data source should define a function 'initGlobalState' that -- initialises the state for that data source.  The arguments to this -- function might vary depending on the data source - we might need to@@ -144,6 +150,8 @@ fetch1 (BlockedFetch (CountAardvarks "BANG3") _) = do   hPutStr stderr "BANG3"   killThread =<< myThreadId -- an asynchronous exception+fetch1 (BlockedFetch (CountAardvarks "BANG4") r) = do+  putFailure r $ NotFound "BANG4" fetch1 (BlockedFetch (CountAardvarks str) m) =   putSuccess m (length (filter (== 'a') str)) fetch1 (BlockedFetch (ListWombats a) r) =

@@ -48,9 +48,6 @@   print stats   assertEqual "FullyAsyncTest: stats" 5 (numFetches stats) -andThen :: GenHaxl u w a -> GenHaxl u w b -> GenHaxl u w b-andThen a b = a >>= \_ -> b- {-            A         B         C 50          |        |       tick 1

@@ -4,12 +4,8 @@ -- This source code is distributed under the terms of a BSD license, -- found in the LICENSE file. -{-# LANGUAGE CPP #-} module MemoizationTests (tests) where -#if __GLASGOW_HASKELL__ < 710-import Control.Applicative-#endif import Data.IORef  import Test.HUnit

@@ -28,7 +28,11 @@ import Prelude () import qualified Data.Map as Map import qualified Data.Text as Text+import Control.Concurrent+import Control.Exception+import Control.Monad (void) + import Haxl.Prelude import Haxl.Core @@ -57,17 +61,20 @@  instance DataSource UserEnv TAOReq where   fetch TAOState{..} _flags _user-    | future = FutureFetch $ return . mapM_ doFetch-    | otherwise = SyncFetch $ mapM_ doFetch+    | future = BackgroundFetch $ \f -> do+        mask_ $ void . forkIO $  mapM_ (doFetch True) f+    | otherwise = SyncFetch $ mapM_ (doFetch False)  initGlobalState :: Bool -> IO (State TAOReq) initGlobalState future = return TAOState { future=future } -doFetch :: BlockedFetch TAOReq -> IO ()-doFetch (BlockedFetch req@(AssocRangeId2s a b) r) =-  case Map.lookup (a, b) assocs of-    Nothing -> putFailure r . NotFound . Text.pack $ show req-    Just result -> putSuccess r result+doFetch ::  Bool -> BlockedFetch TAOReq -> IO ()+doFetch bg (BlockedFetch req@(AssocRangeId2s a b) r) = put result+  where put = if bg then putResultFromChildThread r else putResult r+        result = case Map.lookup (a, b) assocs of+          Nothing -> except . NotFound . Text.pack $ show req+          Just result -> Right result+  assocs :: Map (Id,Id) [Id] assocs = Map.fromList [

@@ -0,0 +1,128 @@+-- Copyright (c) 2014-present, Facebook, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a BSD license,+-- found in the LICENSE file.++{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE ApplicativeDo #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}++module MonadAsyncTest (tests) where+import Haxl.Core+import Test.HUnit hiding (State)+import Control.Concurrent+import Control.Exception as Exception+import Control.Monad+import Haxl.Core.Monad (unsafeLiftIO)+import System.IO.Unsafe+import Data.Hashable+import Data.IORef+import Data.Text (Text)++newtype SimpleWrite = SimpleWrite Text deriving (Eq, Show)++{-# NOINLINE shouldThrowRef #-}+shouldThrowRef :: IORef Bool+shouldThrowRef = unsafePerformIO (newIORef False)++-- | This datasource contains "bad" code which can throw at the wrong+-- moment.+data ThrowableSleep a where+  Sleep :: Int -> ThrowableSleep Int++deriving instance Eq (ThrowableSleep a)+deriving instance Show (ThrowableSleep a)++instance ShowP ThrowableSleep where showp = show++instance Hashable (ThrowableSleep a) where+  hashWithSalt s (Sleep n) = hashWithSalt s n++instance StateKey ThrowableSleep where+  data State ThrowableSleep = ThrowableSleepState++initDataSource :: IO (State ThrowableSleep)+initDataSource = return ThrowableSleepState++instance DataSourceName ThrowableSleep where+  dataSourceName _ = "ThrowableSleep"++instance DataSource u ThrowableSleep where+  fetch _state _flags _u = BackgroundFetch $ \bfs -> forM_ bfs fill+    where+    fill :: BlockedFetch ThrowableSleep -> IO ()+    fill (BlockedFetch (Sleep n) rv) = do+      _ <- forkFinally+        (do+          threadDelay (n*1000)+          return n+        )+        (\res -> do+          shouldThrow <- atomicModifyIORef' shouldThrowRef (\s -> (False, s))+          -- Simulate case when datasource throws before putting Result into+          -- completions queue.+          when shouldThrow $ do+            throwIO $ ErrorCall "datasource threw an exception"+          -- In case the datasource throws before this point, there'll be+          -- nothing to put the result to the queue of 'completions', and+          -- therefore Haxl would block indefinitely.+          --+          -- Note that Haxl tries to catch datasource exceptions and put the+          -- "exception result" into `completions` using `wrapFetchInCatch`+          -- function. However that doesn't work in this case because the+          -- datasource throws in a separate thread.+          putResultFromChildThread rv res+        )+      return ()++tests :: Test+tests = TestList+  [ TestLabel "exceptionTest" exceptionTest+  ]++mkTestEnv :: IO (Env () SimpleWrite)+mkTestEnv = do+  st <- initDataSource+  initEnv (stateSet st stateEmpty) ()++exceptionTest :: Test+exceptionTest = TestCase $ do+  e <- mkTestEnv++  let+      fet (n :: Int) (st :: Bool )= do+        x <- dataFetch (Sleep (fromIntegral n))+        unsafeLiftIO $ writeIORef shouldThrowRef st+        y <- dataFetch (Sleep (fromIntegral x*2))+        return (x+y)++  r1 :: (Either Exception.SomeException Int)+    <- Exception.try $ runHaxl e $ fet 10 True++  -- Datasources are responsible for putting the fetched result into the+  -- completions queue. If for some reason they fail to do so, Haxl throws a+  -- LogicBug since the scheduler is still expecting some request(s) to+  -- be completed.+  case r1 of+    Left ex | Just (LogicBug _) <- Exception.fromException ex -> return ()+    _ -> assertFailure "r1 computation did not fail with Logic Bug!"++  -- Sanitize the env to get rid of all empty IVars+  -- While this test examines the case when there's an exception in the Haxl+  -- datasource itself, a similar behavior will occur in case an async+  -- exception is thrown to the Haxl scheduler thread.+  e' <- sanitizeEnv e++  r2 :: (Either Exception.SomeException Int)+    <- Exception.try $ runHaxl e' $ fet 10 False+  case r2 of+    Right _ -> return ()+    Left ex | Just (LogicBug _) <- Exception.fromException ex -> do+                assertFailure $ "bad exception in r2: " ++ show ex+    Left _ -> return ()

@@ -5,13 +5,15 @@ -- found in the LICENSE file.  -- | Benchmarking tool for core performance characteristics of the Haxl monad.-{-# LANGUAGE CPP, OverloadedStrings #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ApplicativeDo, RecordWildCards #-} module MonadBench (main) where  import Control.Monad import Data.List as List+import Data.Maybe import Data.Time.Clock-import System.Environment+import Options.Applicative import System.Exit import System.IO import Text.Printf@@ -22,117 +24,136 @@ import Haxl.Core.Memo (newMemoWith, runMemo)  import Haxl.Core+import Haxl.Core.Util  import ExampleDataSource  newtype SimpleWrite = SimpleWrite Text deriving (Eq, Show) -testEnv :: IO (Env () SimpleWrite)-testEnv = do+testEnv :: Int -> IO (Env () SimpleWrite)+testEnv report = do   exstate <- ExampleDataSource.initGlobalState   let st = stateSet exstate stateEmpty-  initEnv st ()+  env <- initEnv st ()+  return env { flags = (flags env) { report = report } } -main :: IO ()-main = do-  [test,n_] <- getArgs-  let n = read n_-  env <- testEnv-  t0 <- getCurrentTime-  case test of+type Test = (String, Int, Int -> GenHaxl () SimpleWrite ())++testName :: Test -> String+testName (t,_,_) = t++allTests :: [Test]+allTests =     -- parallel, identical queries-    "par1" -> runHaxl env $-       Haxl.sequence_ (replicate n (listWombats 3))+  [ ("par1", large, \n -> Haxl.sequence_ (replicate n (listWombats 3)))     -- parallel, distinct queries-    "par2" -> runHaxl env $-       Haxl.sequence_ (map listWombats [1..fromIntegral n])+  , ("par2", medium, \n ->+        Haxl.sequence_ (map listWombats [1..fromIntegral n]))     -- sequential, identical queries-    "seqr" -> runHaxl env $-       foldr andThen (return ()) (replicate n (listWombats 3))+  , ("seqr", huge, \n ->+        foldr andThen (return ()) (replicate n (listWombats 3)))     -- sequential, left-associated, distinct queries-    "seql" -> runHaxl env $ do+  , ("seql", medium, \n -> do        _ <- foldl andThen (return []) (map listWombats [1.. fromIntegral n])-       return ()-    "tree" -> runHaxl env $ void $ tree n+       return ())     -- No memoization-    "memo0" -> runHaxl env $-      Haxl.sequence_ [unionWombats | _ <- [1..n]]+  , ("memo0", small, \n -> Haxl.sequence_ [unionWombats | _ <- [1..n]])     -- One put, N gets.-    "memo1" -> runHaxl env $-      Haxl.sequence_ [memo (42 :: Int) unionWombats | _ <- [1..n]]+  , ("memo1", medium_large, \n ->+        Haxl.sequence_ [memo (42 :: Int) unionWombats | _ <- [1..n]])     -- N puts, N gets.-    "memo2" -> runHaxl env $-      Haxl.sequence_ [memo (i :: Int) unionWombats | i <- [1..n]]-    "memo3" ->-      runHaxl env $ do+  , ("memo2", small, \n ->+        Haxl.sequence_ [memo (i :: Int) unionWombats | i <- [1..n]])+  , ("memo3", medium_large, \n -> do         ref <- newMemoWith unionWombats         let c = runMemo ref-        Haxl.sequence_ [c | _ <- [1..n]]-    "memo4" ->-      runHaxl env $ do+        Haxl.sequence_ [c | _ <- [1..n]])+  , ("memo4", small, \n -> do         let f = unionWombatsTo-        Haxl.sequence_ [f x | x <- take n $ cycle [100, 200 .. 1000]]-    "memo5" ->-      runHaxl env $ do+        Haxl.sequence_ [f x | x <- take n $ cycle [100, 200 .. 1000]])+  , ("memo5", medium_large, \n -> do         f <- memoize1 unionWombatsTo-        Haxl.sequence_ [f x | x <- take n $ cycle [100, 200 .. 1000]]-    "memo6" ->-      runHaxl env $ do+        Haxl.sequence_ [f x | x <- take n $ cycle [100, 200 .. 1000]])+  , ("memo6", small, \n -> do         let f = unionWombatsFromTo         Haxl.sequence_ [ f x y                        | x <- take n $ cycle [100, 200 .. 1000]                        , let y = x + 1000-                       ]-    "memo7" ->-      runHaxl env $ do+                       ])+  , ("memo7", medium_large, \n -> do         f <- memoize2 unionWombatsFromTo         Haxl.sequence_ [ f x y                        | x <- take n $ cycle [100, 200 .. 1000]                        , let y = x + 1000-                       ]+                       ])+  , ("cc1", medium_large, \n ->+        Haxl.sequence_ [ cachedComputation (ListWombats 1000) unionWombats+                       | _ <- [1..n]+                       ])+  , ("tree", 20, \n -> void $ tree n (\_ act -> act))+  , ("tree_labels", 20, \n -> void $+      tree n (\n act -> withLabel (textShow n) act))     -- parallel writes-    "write1" -> runHaxl env $-      Haxl.sequence_ (replicate n (tellWrite (SimpleWrite "haha")))-+  , ("write1", large, \n ->+        Haxl.sequence_ (replicate n (tellWrite (SimpleWrite "haha"))))     -- sequential writes-    "write2" -> runHaxl env $-      foldr andThen (return ()) (replicate n (tellWrite (SimpleWrite "haha")))-+  , ("write2", huge, \n -> foldr+        andThen+        (return ())+        (replicate n (tellWrite (SimpleWrite "haha"))))+  ]+  where+    huge = large * 10+    large =  medium * 10+    medium_large =  medium * 4+    medium = 200000+    small = 1000 -    "cc1" -> runHaxl env $-      Haxl.sequence_ [ cachedComputation (ListWombats 1000) unionWombats-                     | _ <- [1..n]-                     ]+data Options = Options+  { test :: String+  , nOverride :: Maybe Int+  , reportFlag :: Int+  } -    _ -> do-      hPutStrLn stderr $ "syntax: monadbench " ++ concat-        [ "par1"-        , "par2"-        , "seqr"-        , "seql"-        , "memo0"-        , "memo1"-        , "memo2"-        , "memo3"-        , "memo4"-        , "memo5"-        , "memo6"-        , "memo7"-        , "cc1"-        ]-      exitWith (ExitFailure 1)+runTest :: Options -> Test -> IO ()+runTest Options{..} (t, nDef, act) = do+  let n = fromMaybe nDef nOverride+  env <- testEnv reportFlag+  t0 <- getCurrentTime+  runHaxl env $ act n   t1 <- getCurrentTime-  printf "%10s: %10d reqs: %.2fs\n"-    test n (realToFrac (t1 `diffUTCTime` t0) :: Double)- where-  -- can't use >>, it is aliased to *> and we want the real bind here-  andThen x y = x >>= const y+  printf "%12s: %10d reqs: %.2fs\n"+    t n (realToFrac (t1 `diffUTCTime` t0) :: Double) -tree :: Int -> GenHaxl () SimpleWrite [Id]-tree 0 = listWombats 0-tree n = concat <$> Haxl.sequence-  [ tree (n-1)-  , listWombats (fromIntegral n), tree (n-1)+optionsParser :: Parser Options+optionsParser = do+  test <- argument str (metavar "TEST")+  reportFlag <- option auto (long "report" <> value 0 <> metavar "REPORT")+  nOverride <- optional $ argument auto (metavar "NUM")+  return Options{..}++main :: IO ()+main = do+  opts@Options{..} <- execParser $ info optionsParser mempty+  let tests = if test == "all"+        then allTests+        else filter ((==) test . testName) allTests+  when+    (null tests)+    (do+        hPutStrLn stderr $ "syntax: monadbench [all|" +++          intercalate "|" (map testName allTests) ++ "]"+        exitWith (ExitFailure 1))+  Control.Monad.mapM_ (runTest opts) tests++tree+  :: Int+  -> (Int -> GenHaxl () SimpleWrite [Id] -> GenHaxl () SimpleWrite [Id])+  -> GenHaxl () SimpleWrite [Id]+tree 0 wrap = wrap 0 $ listWombats 0+tree n wrap = wrap n $ concat <$> Haxl.sequence+  [ tree (n-1) wrap+  , listWombats (fromIntegral n), tree (n-1) wrap   ]  unionWombats :: GenHaxl () SimpleWrite [Id]

@@ -16,7 +16,6 @@ import Haxl.Core.RequestStore (getMapFromRCMap)  import Data.IORef-import Data.Map (Map) import qualified Data.Map as Map import Data.Proxy (Proxy(..)) import Data.Typeable@@ -26,6 +25,7 @@ import ExampleDataSource import SleepDataSource +testEnv :: IO (Env () ()) testEnv = do   exstate <- ExampleDataSource.initGlobalState   sleepState <- mkConcurrentIOState@@ -38,100 +38,44 @@ wombats :: GenHaxl () () Int wombats = length <$> listWombats 3 -getFetches :: Env () () -> IO (Map Text (Map TypeRep Int))-getFetches env = getMapFromRCMap <$> readIORef (submittedReqsRef env)--outgoneFetchesTest :: Test-outgoneFetchesTest = TestCase $ do-  let-    withTimeout env h = timeout 2000 $ runHaxl env h -- 2 ms--  -- test that a completed datasource fetch doesn't show up in Env-  env <- testEnv--  withTimeout env $ do-    _ <- sleep 1 -- 1 ms-    _ <- sleep 1 -- should be cached-    _ <- sleep 1-    wombats--  fetchesMap <- getFetches env-  assertEqual "outgoneFetches1" 0 (Map.size fetchesMap)--  -- test that unfinished datasource fetches shows up in Env-  env <- testEnv--  withTimeout env $ do-    _ <- sleep 4 -- 4 ms-    _ <- wombats-    _ <- sleep 5 -- 4 ms-    _  <- wombats-    return ()--  fetchesMap <- getFetches env-  assertEqual "outgoneFetches2" 1 (Map.size fetchesMap)-  assertEqual "outgoneFetches2"-    (Map.fromList-      [ ( dataSourceName (Proxy :: Proxy (ConcurrentIOReq Sleep))-        , Map.fromList [(typeOf1 (undefined :: ConcurrentIOReq Sleep a), 2)]-        )-      ])-    fetchesMap--  -- test for finished/unfinished fetches from the same datasource-  env <- testEnv--  withTimeout env $ do-    _ <- sleep 1 -- 1 ms-    _ <- sleep 4-    _ <- sleep 5-    return ()--  fetchesMap <- getFetches env-  assertEqual "outgoneFetches3"-    (Map.fromList-      [ ( dataSourceName (Proxy :: Proxy (ConcurrentIOReq Sleep))-        , Map.fromList [(typeOf1 (undefined :: ConcurrentIOReq Sleep a), 2)]-        )-      ])-    fetchesMap--  -- test for cached requests not showing up twice in ReqCountMap-  env <- testEnv--  withTimeout env $ do-    _ <- sleep 4 -- 3 ms-    _ <- sleep 4-    return ()--  fetchesMap <- getFetches env-  assertEqual "outgoneFetches4"-    (Map.fromList-      [ ( dataSourceName (Proxy :: Proxy (ConcurrentIOReq Sleep))-        , Map.fromList [(typeOf1 (undefined :: ConcurrentIOReq Sleep a), 1)]-        )-      ])-    fetchesMap--  -- test for unsent requests not showing up in ReqCountMap+outgoneFetchesTest :: String -> Int -> GenHaxl () () a -> Test+outgoneFetchesTest label unfinished haxl = TestLabel label $ TestCase $ do   env <- testEnv--  withTimeout env $ do-    _ <- sleep =<< sleep 4 -- second req should never be sent-    return ()--  fetchesMap <- getFetches env-  assertEqual "outgoneFetches5"-    (Map.fromList-      [ ( dataSourceName (Proxy :: Proxy (ConcurrentIOReq Sleep))-        , Map.fromList [(typeOf1 (undefined :: ConcurrentIOReq Sleep a), 1)]-        )-      ])-    fetchesMap---+  _ <- timeout (100*1000) $ runHaxl env haxl -- 100ms+  actual <- getMapFromRCMap <$> readIORef (submittedReqsRef env)+  assertEqual "fetchesMap" expected actual+  where+  expected = if unfinished == 0 then Map.empty else+    Map.singleton (dataSourceName (Proxy :: Proxy (ConcurrentIOReq Sleep))) $+      Map.singleton (typeOf1 (undefined :: ConcurrentIOReq Sleep a)) unfinished +tests :: Test tests = TestList-  [ TestLabel "outgoneFetchesTest" outgoneFetchesTest+  [ outgoneFetchesTest "finished" 0 $ do+      -- test that a completed datasource fetch doesn't show up in Env+      _ <- sleep 1  -- finished+      _ <- sleep 1  -- cached/finished+      _ <- sleep 1  -- cached/finished+      wombats+  , outgoneFetchesTest "unfinished" 2 $ do+      -- test that unfinished datasource fetches shows up in Env+      _ <- sleep 200 -- unfinished+      _ <- wombats+      _ <- sleep 300 -- unfinished+      _  <- wombats+      return ()+  , outgoneFetchesTest "mixed" 2 $ do+      -- test for finished/unfinished fetches from the same datasource+      _ <- sleep 1   -- finished+      _ <- sleep 200  -- unfinished+      _ <- sleep 300  -- unfinished+      return ()+  , outgoneFetchesTest "cached" 1 $ do+      -- test for cached requests not showing up twice in ReqCountMap+      _ <- sleep 200  -- unfinished+      _ <- sleep 200  -- cached/unfinished+      return ()+  , outgoneFetchesTest "unsent" 1 $+      -- test for unsent requests not showing up in ReqCountMap+      sleep 200 `andThen` sleep 300 -- second req should never be sent   ]

@@ -0,0 +1,72 @@+module ParallelTests where++import Haxl.Prelude+import Haxl.Core++import Haxl.DataSource.ConcurrentIO+import SleepDataSource++import Data.Time.Clock++import Test.HUnit++testEnv = do+  sleepState <- mkConcurrentIOState+  let st = stateSet sleepState stateEmpty+  initEnv st ()++sync_test :: IO ()+sync_test = do+  env <- testEnv+  -- This computation tests that the two arguments of the pOr can fire+  -- without causing an error. The reason we test for this is that the+  -- synchronization involved in this case is a little fragile.+  False <- runHaxl env $ do+    (fmap (const False) (sleep 50)+      `pOr` fmap (const False) (sleep 100))+      `pOr` fmap (const False) (sleep 200)+  return ()++semantics_when_computation_is_blocked_test :: IO ()+semantics_when_computation_is_blocked_test = do+  env <- testEnv+    -- Test semantics of blocking+  let sleepReturn bool t = do+        _ <- sleep t+        return bool+  r <- runHaxl env $ do+        -- All sleep times are different so that they're not cached+        a <- sleepReturn False 10 `pOr` sleepReturn False 11+        b <- sleepReturn False 12 `pOr` sleepReturn True 13+        c <- sleepReturn True 14 `pOr` sleepReturn False 15+        d <- sleepReturn True 16 `pOr` sleepReturn True 17+        return (not a && b && c && d)+  assertBool "pOr blocked semantics" r+++timing_test = do+  env <- testEnv+  t0 <- getCurrentTime+  True <- runHaxl env $+    fmap (const True) (sleep 200) `pOr` fmap (const True) (sleep 100)+  t1 <- getCurrentTime+  True <- runHaxl env $+    fmap (const True) (sleep 100) `pOr` fmap (const True) (sleep 200)+  t2 <- getCurrentTime+  False <- runHaxl env $+    fmap (const False) (sleep 200) `pOr` fmap (const False) (sleep 100)+  t3 <- getCurrentTime+  False <- runHaxl env $+    fmap (const False) (sleep 100) `pOr` fmap (const False) (sleep 200)+  t4 <- getCurrentTime+  -- diffUTCTime returns the difference in seconds,+  -- while sleep expects milliseconds+  assert (t4 `diffUTCTime` t3 < 0.2)+  assert (t3 `diffUTCTime` t2 < 0.2)+  assert (t2 `diffUTCTime` t1 < 0.2)+  assert (t1 `diffUTCTime` t0 < 0.2)++tests = TestList [TestLabel "sync_test" (TestCase sync_test)+                 ,TestLabel "timing_test" (TestCase timing_test)+                 ,TestLabel "semantics_when_computation_is_blocked_test" (TestCase semantics_when_computation_is_blocked_test)+                 ]

@@ -6,6 +6,7 @@  {-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}  module ProfileTests where @@ -14,7 +15,6 @@ import Haxl.Core import Haxl.Core.Monad import Haxl.Core.Stats-import Haxl.DataSource.ConcurrentIO  import Test.HUnit @@ -23,15 +23,26 @@ import Data.Aeson import Data.IORef import qualified Data.HashMap.Strict as HashMap-import qualified Data.HashSet as HashSet+import Data.Int  import TestUtils import WorkDataSource+import SleepDataSource  mkProfilingEnv = do   env <- makeTestEnv False   return env { flags = (flags env) { report = 4 } } +-- expects only one label to be shown+labelToDataMap :: Profile -> HashMap.HashMap ProfileLabel ProfileData+labelToDataMap Profile{..} = HashMap.fromList hashKeys+  where+    labelKeys = HashMap.fromList [+      (k, l) | ((l, _), k) <- HashMap.toList profileTree]+    hashKeys = [ (l, v)+      | (k, v) <- HashMap.toList profile+      , Just l <- [HashMap.lookup k labelKeys]]+ collectsdata :: Assertion collectsdata = do   e <- mkProfilingEnv@@ -39,22 +50,42 @@           withLabel "bar" $             withLabel "foo" $ do               u <- env userEnv+              slp <- sum <$> mapM (\x -> withLabel "baz" $ return x) [1..5]               -- do some non-trivial work that can't be lifted out-              case fromJSON <$> HashMap.lookup "A" u of+              -- first sleep though in order to force a Blocked result+              sleep slp `andThen` case fromJSON <$> HashMap.lookup "A" u of                 Just (Success n) | sum [n .. 1000::Integer] > 0 -> return 5                 _otherwise -> return (4::Int)-  profData <- profile <$> readIORef (profRef e)-  assertEqual "has data" 3 $ HashMap.size profData+  profCopy <- readIORef (profRef e)+  let+    profData = profile profCopy+    labelKeys = HashMap.fromList [+      (l, k) | ((l, _), k) <- HashMap.toList (profileTree profCopy)]+    getData k = do+      k2 <- HashMap.lookup k labelKeys+      HashMap.lookup k2 profData+  assertEqual "has data" 4 $ HashMap.size profData   assertBool "foo allocates" $-    case profileAllocs <$> HashMap.lookup "foo" profData of+    case profileAllocs <$> getData "foo" of       Just x -> x > 10000       Nothing -> False+  assertEqual "foo is only called once" (Just 1) $+    profileLabelHits <$> getData "foo"+  assertEqual "baz is called 5 times" (Just 5) $+    profileLabelHits <$> getData "baz"   assertBool "bar does not allocate (much)" $-    case profileAllocs <$> HashMap.lookup "bar" profData of+    case profileAllocs <$> getData "bar" of       Just n -> n < 5000  -- getAllocationCounter can be off by +/- 4K       _otherwise -> False-  assertEqual "foo's parent" (Just ["bar"]) $-    HashSet.toList . profileDeps <$> HashMap.lookup "foo" profData+  let fooParents = case HashMap.lookup "foo" labelKeys of+        Nothing -> []+        Just kfoo ->+          [ kparent+          | ((_, kparent), k) <- HashMap.toList (profileTree profCopy)+          , k == kfoo]+  assertEqual "foo's parent" 1 (length fooParents)+  assertEqual "foo's parent is bar" (Just (head fooParents)) $+    HashMap.lookup ("bar", 0) (profileTree profCopy)  exceptions :: Assertion exceptions = do@@ -63,7 +94,7 @@           withLabel "outer" $             tryToHaxlException $ withLabel "inner" $               unsafeLiftIO $ evaluate $ force (error "pure exception" :: Int)-  profData <- profile <$> readIORef (profRef env)+  profData <- labelToDataMap <$> readIORef (profRef env)   assertBool "inner label not added" $     not $ HashMap.member "inner" profData @@ -72,7 +103,7 @@           withLabel "outer" $             tryToHaxlException $ withLabel "inner" $               throw $ NotFound "haxl exception"-  profData <- profile <$> readIORef (profRef env2)+  profData <- labelToDataMap <$> readIORef (profRef env2)   assertBool "inner label added" $     HashMap.member "inner" profData @@ -80,20 +111,78 @@ -- Test that we correctly attribute work done in child threads when -- using BackgroundFetch to the caller of runHaxl. This is important -- for correct accounting when relying on allocation limits.-threadAlloc :: Assertion-threadAlloc = do-  st <- mkConcurrentIOState-  env <- initEnv (stateSet st stateEmpty) ()+threadAlloc :: Integer -> Assertion+threadAlloc batches = do+  env' <- initEnv (stateSet mkWorkState stateEmpty) ()+  let env = env'  { flags = (flags env') { report = 2 } }   a0 <- getAllocationCounter-  _x <- runHaxl env $ work 100000+  let+    wsize = 100000+    w = forM [wsize..(wsize+batches-1)] work+  _x <- runHaxl env $ sum <$> w   a1 <- getAllocationCounter-  assertBool "threadAlloc" $ (a0 - a1) > 1000000+  let+    lower = fromIntegral $ 1000000 * batches+    upper = fromIntegral $ 25000000 * batches+  assertBool "threadAlloc lower bound" $ (a0 - a1) > lower+  assertBool "threadAlloc upper bound" $ (a0 - a1) < upper     -- the result was 16MB on 64-bit, or around 25KB if we miss the allocs-    -- in the child thread.+    -- in the child thread. For batched it should be similarly scaled.+    -- When we do not reset the counter for each batch was+    -- scaled again by number of batches. +  stats <- readIORef (statsRef env)+  assertEqual+    "threadAlloc: batches"+    [fromIntegral batches]+    (aggregateFetchBatches length stats)+  -- if we actually do more than 1 batch then the above test is not useful +data MemoType = Global | Local++-- Test that we correctly attribute memo work+memos:: MemoType -> Assertion+memos memoType = do+  env <- mkProfilingEnv+  let+    memoAllocs = 10000000 :: Int64+    doWork = unsafeLiftIO $ do+      a0 <- getAllocationCounter+      setAllocationCounter $ a0 - memoAllocs+      return (5 :: Int)+    mkWork+      | Global <- memoType = return (memo (1 :: Int) doWork)+      | Local <- memoType = memoize doWork+  _ <- runHaxl env $ do+    work <- mkWork+    andThen+      (withLabel "do" work)+      (withLabel "cached" work)+  profData <- labelToDataMap <$> readIORef (profRef env)+  case HashMap.lookup "do" profData of+    Nothing -> assertFailure "do not in data"+    Just ProfileData{..} -> do+      assertEqual "has correct memo id" profileMemos [ProfileMemo 1 False]+      assertBool "allocs are included in 'do'" (profileAllocs >= memoAllocs)+  case HashMap.lookup "cached" profData of+    Nothing -> assertFailure "cached not in data"+    Just ProfileData{..} -> do+      assertEqual "has correct memo id" profileMemos [ProfileMemo 1 True]+      assertBool "allocs are *not* included in 'cached'" (profileAllocs < 50000)+  (Stats memoStats) <- readIORef (statsRef env)+  assertEqual "exactly 1 memo/fetch" 1 (length memoStats)+  let memoStat = head memoStats+  putStrLn $ "memoStat=" ++ show memoStat+  assertEqual "correct call id" 1 (memoStatId memoStat)+  assertBool "allocs are big enough" $ memoSpace memoStat >= memoAllocs+  assertBool "allocs are not too big" $ memoSpace memoStat < memoAllocs + 100000++ tests = TestList   [ TestLabel "collectsdata" $ TestCase collectsdata   , TestLabel "exceptions" $ TestCase exceptions-  , TestLabel "threads" $ TestCase threadAlloc+  , TestLabel "threads" $ TestCase (threadAlloc 1)+  , TestLabel "threads with batch" $ TestCase (threadAlloc 50)+  , TestLabel "memos - Global" $ TestCase (memos Global)+  , TestLabel "memos - Local" $ TestCase (memos Local)   ]

@@ -0,0 +1,191 @@+-- Copyright (c) 2014-present, Facebook, Inc.+-- All rights reserved.+--+-- This source code is distributed under the terms of a BSD license,+-- found in the LICENSE file.++{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}++module StatsTests (tests) where++import Test.HUnit+import Data.List+import Data.Maybe++import Haxl.Prelude+import Haxl.Core+import Prelude()++import ExampleDataSource+import SleepDataSource+import Haxl.DataSource.ConcurrentIO++import Control.Monad (void)+import Data.IORef+import qualified Data.HashMap.Strict as HashMap++aggregateBatches :: Test+aggregateBatches = TestCase $ do+  let+    statsNoBatches = [ FetchStats { fetchDataSource = "foo"+                                  , fetchBatchSize = 7+                                  , fetchStart = 0+                                  , fetchDuration = 10+                                  , fetchSpace = 1+                                  , fetchFailures = 2+                                  , fetchIgnoredFailures = 0+                                  , fetchBatchId = n+                                  , fetchIds = [1,2] }+                                  | n <- reverse [1..10] ++ [11..20] ]+                     ++ [ FetchCall "A" ["B"] 1, FetchCall "C" ["D"] 2 ]+    fetchBatch = [ FetchStats { fetchDataSource = "batch"+                              , fetchBatchSize = 1+                              , fetchStart = 100+                              , fetchDuration = 1000 * n+                              , fetchSpace = 3+                              , fetchFailures = if n <= 3 then 1 else 0+                              , fetchIgnoredFailures = 0+                              , fetchBatchId = 123+                              , fetchIds = [fromIntegral n] } | n <- [1..50] ]+    agg (sz,bids) FetchStats{..} = (sz + fetchBatchSize, fetchBatchId:bids)+    agg _ _ = error "unexpected"+    agg' = foldl' agg (0,[])+    aggNoBatch = aggregateFetchBatches agg' (Stats statsNoBatches)+    expectedNoBatch = [(7, [n]) | n <- reverse [1..20] :: [Int]]+    aggBatch = aggregateFetchBatches agg' (Stats fetchBatch)+    expectedResultBatch = (50, [123 | _ <- [1..50] :: [Int]])+    aggInterspersedBatch =+      aggregateFetchBatches agg'+      (Stats $ intersperse (head fetchBatch) statsNoBatches)+    expectedResultInterspersed =+      (21, [123 | _ <- [1..21] :: [Int]]) : expectedNoBatch+  assertEqual "No batch has no change" expectedNoBatch aggNoBatch+  assertEqual "Batch is combined" [expectedResultBatch] aggBatch+  assertEqual+    "Grouping works as expected" expectedResultInterspersed aggInterspersedBatch+++testEnv = do+  -- To use a data source, we need to initialize its state:+  exstate <- ExampleDataSource.initGlobalState+  sleepState <- mkConcurrentIOState++  -- And create a StateStore object containing the states we need:+  let st = stateSet exstate (stateSet sleepState stateEmpty)++  -- Create the Env:+  env <- initEnv st ()+  return env{ flags = (flags env){ report = 5 } }+++fetchIdsSync :: Test+fetchIdsSync = TestCase $ do+  env <- testEnv+  _ <- runHaxl  env $+       sequence_+       [ void $ countAardvarks "abcabc" + (length <$> listWombats 3)+       , void $ listWombats 100+       , void $ listWombats 99+       , void $ countAardvarks "BANG4" `catch` \NotFound{} -> return 123+       ]+  -- expect a single DS stat+  (Stats stats) <- readIORef (statsRef env)+  let+    fetchStats = [x | x@FetchStats{} <- stats]+  assertEqual "Only 1 batch" 1 (length fetchStats)+  let+    [stat] = fetchStats+  assertEqual "No real failures" 0 (fetchFailures stat)+  assertEqual "1 ignored failure" 1 (fetchIgnoredFailures stat)++fetchIdsBackground :: Test+fetchIdsBackground = TestCase $ do+  env <- testEnv+  _ <- runHaxl  env $+       sequence_+       [ withLabel "short" $ sleep 1+       , withLabel "long" $ sleep 500 ]++  -- make sure that with memo'ing we still preserve the stack+  _ <- runHaxl  env $ withLabel "base"+    (memo (1 :: Int) $ withLabel "child" $ sleep 102)++  _ <- runHaxl env $ withLabel "short_cached" $ sleep 1++  -- expect a single DS stat+  (Stats stats) <- readIORef (statsRef env)+  (Profile p pt _) <- readIORef (profRef env)+  let+    keyMap =+      HashMap.fromList [ (label, k) | ((label,_), k) <- HashMap.toList pt]+    revMap = HashMap.fromList [(v,k) | (k,v) <- HashMap.toList pt]+    parentMap =+      HashMap.fromList $+      catMaybes+      [ case HashMap.lookup kp revMap of+          Just (lp,_) -> Just (label, lp)+          Nothing -> Nothing+      | ((label,kp), _) <- HashMap.toList pt]+    fetchMap =  HashMap.fromList [ (fid, x) | x@FetchStats{} <- stats+                                   , fid <- fetchIds x]+    get l = [ (prof, wasCached, fetchStat)+            | Just key <- [HashMap.lookup l keyMap]+            , Just prof <- [HashMap.lookup key p]+            , ProfileFetch fid _ wasCached <- profileFetches prof+            , Just fetchStat <- [HashMap.lookup fid fetchMap]]+    [(short, shortWC, shortFetch)] = get "short"+    [(long, longWC, longFetch)] = get "long"+    [(shortCached, shortCachedWC, shortCachedFetch)] = get "short_cached"++  assertEqual "3 batches" 3 (HashMap.size fetchMap)+  assertEqual "6 labels (inc MAIN)" 6 (HashMap.size keyMap)++  assertEqual "child parent is base"+    (Just "base")+    (HashMap.lookup "child" parentMap)++  assertEqual "base parent is MAIN"+    (Just "MAIN")+    (HashMap.lookup "base" parentMap)++  assertEqual "long parent is MAIN"+    (Just "MAIN")+    (HashMap.lookup "long" parentMap)++  assertBool "original fetches not cached (short)" (not shortWC)+  assertBool "original fetches not cached (long)" (not longWC)+  assertBool "was cached short" shortCachedWC++  assertEqual "one fetch short" 1 (length $ profileFetches short)+  assertEqual "one fetch long" 1 (length $ profileFetches long)+  assertEqual "one fetch short_cached" 1 (length $ profileFetches shortCached)++  assertBool "short fetch mapped properly" (fetchDuration shortFetch < 100000)+  assertEqual+    "short cached fetch mapped properly"+    (fetchDuration shortFetch)+    (fetchDuration shortCachedFetch)+  assertBool "long fetch was mapped properly" (fetchDuration longFetch > 100000)+++ppStatsTest :: Test+ppStatsTest = TestCase $ do+  let+    r = ppStats (Stats [])+    mc = ppStats (Stats [MemoCall 0 0])+    fc = ppStats (Stats [FetchCall "" [] 0])+    fw = ppStats (Stats [FetchWait HashMap.empty 0 1])+    fs = ppStats (Stats [FetchStats "" 0 0 0 0 0 0 0 []])+  assertEqual "empty stats -> empty string" r ""+  assertEqual "memo call stats -> empty string" mc ""+  assertEqual "fetch call stats -> empty string" fc ""+  assertBool "fetch wait stats -> some data" (not $ null fw)+  assertBool "fetch stats -> some data" (not $ null fs)+++tests = TestList [ TestLabel "Aggregate Batches" aggregateBatches+                 , TestLabel "Fetch IDs Sync" fetchIdsSync+                 , TestLabel "Fetch IDs Background" fetchIdsBackground+                 , TestLabel "ppStats" ppStatsTest ]

@@ -15,29 +15,38 @@ import Data.IORef import Test.HUnit import Control.Exception+import System.Mem  import ExampleDataSource import BadDataSource -testEnv fn = do+testEnv impl fn = do+  -- Use allocation limits, just to make sure haxl properly behaves and+  -- doesn't reset this internally somewhere.+  -- `go` will disable this+  setAllocationCounter 5000000 -- 5 meg should be enough, uses ~100k atm+  enableAllocationLimit   exstate <- ExampleDataSource.initGlobalState-  badstate <- BadDataSource.initGlobalState+  badstate <- BadDataSource.initGlobalState impl   let st = stateSet exstate $ stateSet (fn badstate) stateEmpty   initEnv st ()  wombats :: GenHaxl () () Int wombats = length <$> listWombats 3 -badDataSourceTest :: Test-badDataSourceTest = TestCase $ do+wombatsMany :: GenHaxl () () Int+wombatsMany = length <$> listWombats 7++go :: FetchImpl -> Test+go impl = TestCase $ flip finally disableAllocationLimit $ do   -- test that a failed acquire doesn't fail the other requests   ref <- newIORef False-  env <- testEnv $ \st ->+  env <- testEnv impl $ \st ->     st { failAcquire = throwIO (DataSourceError "acquire")        , failRelease = writeIORef ref True }    x <- runHaxl env $-        (dataFetch (FailAfter 0) + wombats)+        (dataFetch (FailAfter 0) + wombatsMany)           `Haxl.catch` \DataSourceError{} -> wombats    assertEqual "badDataSourceTest1" 3 x@@ -47,12 +56,12 @@    -- test that a failed dispatch doesn't fail the other requests   ref <- newIORef False-  env <- testEnv $ \st ->+  env <- testEnv impl $ \st ->     st { failDispatch = throwIO (DataSourceError "dispatch")        , failRelease = writeIORef ref True }    x <- runHaxl env $-        (dataFetch (FailAfter 0) + wombats)+        (dataFetch (FailAfter 0) + wombatsMany)           `Haxl.catch` \DataSourceError{} -> wombats    assertEqual "badDataSourceTest3" x 3@@ -61,11 +70,11 @@   assertEqual "badDataSourceTest4" True =<< readIORef ref    -- test that a failed wait is a DataSourceError-  env <- testEnv $ \st ->+  env <- testEnv impl $ \st ->     st { failWait = throwIO (DataSourceError "wait") }    x <- runHaxl env $-        (dataFetch (FailAfter 0) + wombats)+        (dataFetch (FailAfter 0) + wombatsMany)           `Haxl.catch` \DataSourceError{} -> wombats    assertEqual "badDataSourceTest5" x 3@@ -75,18 +84,45 @@    -- test that a failed release is still a DataSourceError, even   -- though the request will have completed successfully-  env <- testEnv $ \st ->+  env <- testEnv impl $ \st ->     st { failRelease = throwIO (DataSourceError "release") } +  let+    -- In background fetches the scheduler might happen to process the data+    -- source result (FetchError in this case) before it processes the exception+    -- from release. So we have to allow both cases.+    isBg = case impl of+      Background -> True+      BackgroundMVar -> True+      _ -> False+    releaseCatcher e+      | Just DataSourceError{} <- fromException e = wombats+      | Just FetchError{} <- fromException e =+          if isBg then wombats else Haxl.throw e+      | otherwise = Haxl.throw e+   x <- runHaxl env $-        (dataFetch (FailAfter 0) + wombats)-          `Haxl.catch` \DataSourceError{} -> wombats+        (dataFetch (FailAfter 0) + wombatsMany)+          `Haxl.catch` releaseCatcher    assertEqual "badDataSourceTest7" x 3 +  -- test that if we don't throw anything we get the result+  -- (which is a fetch error for this source)+  env <- testEnv impl id+  x <- runHaxl env $+        (dataFetch (FailAfter 0) + wombatsMany)+          `Haxl.catch` \FetchError{} -> wombats +  assertEqual "badDataSourceTest8" x 3  ++ tests = TestList-  [ TestLabel "badDataSourceTest" badDataSourceTest+  [ TestLabel "badDataSourceTest async" (go Async)+  , TestLabel "badDataSourceTest background" (go Background)+  , TestLabel "badDataSourceTest backgroundMVar" (go BackgroundMVar)+  , TestLabel "badDataSourceTest backgroundFetchSeq" (go BackgroundSeq)+  , TestLabel "badDataSourceTest backgroundFetchPar" (go BackgroundPar)   ]

@@ -209,7 +209,7 @@   env <- testEnv   runHaxl env loadCache   str <- runHaxl env dumpCacheAsHaskell-  lcPath <- loadCachePath+  lcPath <-loadCachePath   loadcache <- readFile lcPath   -- The order of 'cacheRequest ...' calls is nondeterministic and   -- differs among GHC versions, so we sort the lines for comparison.

@@ -16,6 +16,7 @@  import TestTypes import MockTAO+import Haxl.DataSource.ConcurrentIO  import Data.IORef import Data.Aeson@@ -49,7 +50,8 @@ makeTestEnv :: Bool -> IO HaxlEnv makeTestEnv future = do   tao <- MockTAO.initGlobalState future-  let st = stateSet tao stateEmpty+  stio <- mkConcurrentIOState+  let st = stateSet stio $ stateSet tao stateEmpty   env <- initEnv st testinput   return env { flags = (flags env) { report = 2 } } 

@@ -10,8 +10,10 @@ {-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE TypeFamilies #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE OverloadedStrings #-}  module WorkDataSource (+    mkWorkState,     work,   ) where @@ -19,26 +21,65 @@ import Prelude ()  import Haxl.Core-import Haxl.DataSource.ConcurrentIO  import Control.Exception import Data.Hashable import Data.Typeable+import Control.Monad (void)+import Control.Concurrent.MVar -work :: Integer -> GenHaxl u w Integer-work n = dataFetch (Work n) -data Work deriving Typeable-instance ConcurrentIO Work where-  data ConcurrentIOReq Work a where-    Work :: Integer -> ConcurrentIOReq Work Integer+data Work a where+  Work :: Integer -> Work Integer+  deriving Typeable -  performIO (Work n) = evaluate (sum [1..n]) >> return n+deriving instance Eq (Work a)+deriving instance Show (Work a)+instance ShowP Work where showp = show -deriving instance Eq (ConcurrentIOReq Work a)-deriving instance Show (ConcurrentIOReq Work a)+instance Hashable (Work a) where+   hashWithSalt s (Work a) = hashWithSalt s (0::Int,a) -instance ShowP (ConcurrentIOReq Work) where showp = show+instance DataSourceName Work where+  dataSourceName _ = "Work" -instance Hashable (ConcurrentIOReq Work a) where-  hashWithSalt s (Work n) = hashWithSalt s n+instance StateKey Work where+  data State Work = WorkState++newtype Service = Service (MVar [IO ()])++run :: Work a -> IO a+run (Work n) = evaluate (sum [1..n]) >> return n++mkService :: IO Service+mkService = Service <$> newMVar []++process :: Service -> IO ()+process (Service q) = do+  r <- swapMVar q []+  sequence_ r++enqueue :: Service -> Work a -> IO (IO (Either SomeException a))+enqueue (Service q) w = do+  res <- newEmptyMVar+  let r = do+        v <- Control.Exception.try $ run w+        putMVar res v+  modifyMVar_ q (return . (:) r)+  return (takeMVar res)++instance DataSource u Work where+  fetch = backgroundFetchAcquireReleaseMVar+    mkService+    (\_ -> return ())+    -- pretend we are ready so that process does the work+    (\_ _ m -> void $ tryPutMVar m ())+    process+    enqueue+++mkWorkState :: State Work+mkWorkState = WorkState++work :: Integer -> GenHaxl u w Integer+work n = dataFetch (Work n)

@@ -4,13 +4,9 @@ -- This source code is distributed under the terms of a BSD license, -- found in the LICENSE file. -{-# LANGUAGE CPP, OverloadedStrings #-}+{-# LANGUAGE OverloadedStrings #-} module WriteTests (tests) where -#if __GLASGOW_HASKELL__ < 710-import Control.Applicative-#endif- import Test.HUnit  import Data.Foldable@@ -21,8 +17,8 @@ newtype SimpleWrite = SimpleWrite Text   deriving (Eq, Show) -doWrite :: GenHaxl u SimpleWrite Int-doWrite = do+doInnerWrite :: GenHaxl u SimpleWrite Int+doInnerWrite = do   tellWrite $ SimpleWrite "inner"   return 0 @@ -30,7 +26,7 @@ doOuterWrite = do   tellWrite $ SimpleWrite "outer1" -  doWriteMemo <- newMemoWith doWrite+  doWriteMemo <- newMemoWith doInnerWrite   let doMemoizedWrite = runMemo doWriteMemo   _ <- doMemoizedWrite   _ <- doMemoizedWrite@@ -39,6 +35,12 @@    return 1 +doNonMemoWrites :: GenHaxl u SimpleWrite Int+doNonMemoWrites = do+  tellWrite $ SimpleWrite "inner"+  tellWriteNoMemo $ SimpleWrite "inner not memo"+  return 0+ writeSoundness :: Test writeSoundness = TestCase $ do   let numReps = 4@@ -46,7 +48,7 @@   -- do writes without memoization   env1 <- emptyEnv ()   (allRes, allWrites) <- runHaxlWithWrites env1 $-    Haxl.sequence (replicate numReps doWrite)+    Haxl.sequence (replicate numReps doInnerWrite)    assertBool "Write Soundness 1" $     allWrites == replicate numReps (SimpleWrite "inner")@@ -56,7 +58,7 @@   env2 <- emptyEnv ()    (memoRes, memoWrites) <- runHaxlWithWrites env2 $ do-    doWriteMemo <- newMemoWith doWrite+    doWriteMemo <- newMemoWith doInnerWrite     let memoizedWrite = runMemo doWriteMemo      Haxl.sequence (replicate numReps memoizedWrite)@@ -69,10 +71,10 @@   env3 <- emptyEnv ()    (ilRes, ilWrites) <- runHaxlWithWrites env3 $ do-    doWriteMemo <- newMemoWith doWrite+    doWriteMemo <- newMemoWith doInnerWrite     let memoizedWrite = runMemo doWriteMemo -    Haxl.sequence $ replicate numReps (doWrite *> memoizedWrite)+    Haxl.sequence $ replicate numReps (doInnerWrite *> memoizedWrite)    assertBool "Write Soundness 5" $     ilWrites == replicate (2*numReps) (SimpleWrite "inner")@@ -96,5 +98,32 @@   assertBool "Write Soundness 7" $     nestWrites == fold (replicate numReps expWrites)   assertBool "Write Soundness 8" $ nestRes == replicate numReps 1++  -- do both kinds of writes without memoization+  env5 <- emptyEnv ()+  (allRes, allWrites) <- runHaxlWithWrites env5 $+    Haxl.sequence (replicate numReps doNonMemoWrites)++  assertBool "Write Soundness 9" $+    allWrites == replicate numReps (SimpleWrite "inner") +++      replicate numReps (SimpleWrite "inner not memo")+  assertBool "Write Soundness 10" $ allRes == replicate numReps 0++  -- do both kinds of writes with memoization+  env6 <- emptyEnv ()++  (memoRes, memoWrites) <- runHaxlWithWrites env6 $ do+    doWriteMemo <- newMemoWith doNonMemoWrites+    let memoizedWrite = runMemo doWriteMemo++    Haxl.sequence (replicate numReps memoizedWrite)++  -- "inner not memo" only appears once in this test+  assertBool "Write Soundness 11" $+    memoWrites == replicate numReps (SimpleWrite "inner") +++      [SimpleWrite "inner not memo"]+  assertBool "Write Soundness 12" $ memoRes == replicate numReps 0++  tests = TestList [TestLabel "Write Soundness" writeSoundness]