content: {-
    Copyright (c) Meta Platforms, Inc. and affiliates.
    All rights reserved.
  
    This source code is licensed under the BSD-style license found in the
    LICENSE file in the root directory of this source tree.
  -}
  
  {-# LANGUAGE CPP #-}
  {-# LANGUAGE ConstraintKinds #-}
  {-# LANGUAGE GADTs #-}
  {-# LANGUAGE MagicHash #-}
  {-# LANGUAGE RankNTypes #-}
  {-# LANGUAGE BangPatterns #-}
  {-# LANGUAGE RecordWildCards #-}
  {-# LANGUAGE ScopedTypeVariables #-}
  {-# LANGUAGE OverloadedStrings #-}
  {-# LANGUAGE KindSignatures #-}
  {-# LANGUAGE StandaloneDeriving #-}
  {-# LANGUAGE DeriveDataTypeable #-}
  
  -- | Implementation of data-fetching operations.  Most users should
  -- import "Haxl.Core" instead.
  --
  module Haxl.Core.Fetch
    ( dataFetch
    , dataFetchWithShow
    , dataFetchWithInsert
    , uncachedRequest
    , cacheResult
    , dupableCacheRequest
    , cacheResultWithShow
    , cacheRequest
    , performFetches
    , performRequestStore
    , ShowReq
    ) where
  
  import Control.Concurrent.STM
  import Control.Exception as Exception
  import Control.Monad
  import Data.Either
  import Data.Hashable
  import Data.IORef
  import Data.Int
  import Data.List
  #if __GLASGOW_HASKELL__ < 804
  import Data.Monoid
  #endif
  import Data.Proxy
  import Data.Typeable
  import Data.Text (Text)
  import Data.Kind (Type)
  import qualified Data.Text as Text
  import Text.Printf
  #ifdef PROFILING
  import GHC.Stack
  #endif
  
  import Haxl.Core.DataSource
  import Haxl.Core.DataCache as DataCache
  import Haxl.Core.Exception
  import Haxl.Core.Flags
  import Haxl.Core.Monad
  import Haxl.Core.Profile
  import Haxl.Core.RequestStore
  import Haxl.Core.ShowP
  import Haxl.Core.Stats
  import Haxl.Core.StateStore
  import Haxl.Core.Util
  
  -- -----------------------------------------------------------------------------
  -- Data fetching and caching
  
  -- | Possible responses when checking the cache.
  data CacheResult u w a
    -- | The request hadn't been seen until now.
    = 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.
  type ShowReq r a = (r a -> String, a -> String)
  
  -- Note [showFn]
  --
  -- Occasionally, for tracing purposes or generating exceptions, we need to
  -- call 'show' on the request in a place where we *cannot* have a Show
  -- dictionary. (Because the function is a worker which is called by one of
  -- the *WithShow variants that take explicit show functions via a ShowReq
  -- argument.) None of the functions that does this is exported, so this is
  -- hidden from the Haxl user.
  
  cachedWithInsert
    :: forall r a u w.
       (DataSource u r, Typeable (r a))
    => (r a -> String)    -- See Note [showFn]
    -> (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@Env{..} req = do
    let
      doFetch = do
        ivar <- newIVar
        k <- nextCallId env
        let !rvar = stdResultVar env ivar (Proxy :: Proxy r)
        insertFn req (DataCacheItem ivar k) dataCache
        return (Uncached rvar ivar k)
    mbRes <- DataCache.lookup req dataCache
    case mbRes of
      Nothing -> doFetch
      Just (DataCacheItem i@IVar{ivarRef = cr} k) -> do
        e <- readIORef cr
        case e of
          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 k)
  
  
  -- | Make a ResultVar with the standard function for sending a CompletionReq
  -- to the scheduler. This is the function will be executed when the fetch
  -- completes.
  stdResultVar
    :: forall r a u w. (DataSourceName r, Typeable r)
    => Env u w
    -> IVar u w a
    -> Proxy r
    -> ResultVar a
  stdResultVar Env{..} ivar p =
    mkResultVar $ \r isChildThread _ -> do
      allocs <- if isChildThread
        then
          -- In a child thread, return the current allocation counter too,
          -- for correct tracking of allocation.
          getAllocationCounter
        else
          return 0
      atomicallyOnBlocking
        (LogicBug (ReadingCompletionsFailedFetch (dataSourceName p))) $ do
        cs <- readTVar completions
        writeTVar completions (CompleteReq (eitherToResult 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 ReportOutgoneFetches $
        atomicModifyIORef' submittedReqsRef (\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 -> CallId -> IO ()
  #ifdef PROFILING
  logFetch env showFn req fid = do
    ifReport (flags env) ReportFetchStack $ do
      stack <- currentCallStack
      modifyIORef' (statsRef env) $ \(Stats s) ->
        Stats (FetchCall (showFn req) stack fid : s)
  #else
  logFetch _ _ _ _ = return ()
  #endif
  
  calcFailure
    :: forall u req a . DataSource u req
    => u
    -> req a
    -> Either SomeException a
    -> FailureCount
  calcFailure _u _r Right{} = mempty
  calcFailure u r (Left e) = case classifyFailure u r e of
    StandardFailure -> mempty { failureCountStandard = 1 }
    IgnoredForStatsFailure -> mempty { failureCountIgnored = 1 }
  
  addFallbackFetchStats
    :: forall u w req a . DataSource u req
    => Env u w
    -> CallId
    -> req a
    -> ResultVal a w
    -> IO ()
  addFallbackFetchStats Env{..} fid req res = do
    bid <- atomicModifyIORef' statsBatchIdRef $ \x -> (x+1,x+1)
    start <- getTimestamp
    let
      dsName = dataSourceName (Proxy :: Proxy req)
      FailureCount{..} = case res of
        Ok{} -> mempty
        (ThrowHaxl e _) -> calcFailure userEnv req (Left e)
        (ThrowIO e) -> calcFailure userEnv req (Left e)
      this = FetchStats { fetchDataSource = dsName
                        , fetchBatchSize = 1
                        , fetchStart = start
                        , fetchDuration = 0
                        , fetchSpace = 0
                        , fetchFailures = failureCountStandard
                        , fetchIgnoredFailures = failureCountIgnored
                        , fetchBatchId = bid
                        , fetchIds = [fid] }
    atomicModifyIORef' statsRef $ \(Stats fs) -> (Stats (this : fs), ())
  
  addFallbackResult
    :: Env u w
    -> ResultVal a w
    -> IVar u w a
    -> IO ()
  addFallbackResult Env{..} res ivar = do
    atomicallyOnBlocking
      (LogicBug (ReadingCompletionsFailedFetch "addFallbackResult")) $ do
      cs <- readTVar completions
      writeTVar completions (CompleteReq res ivar 0 : cs)
  
  -- | Performs actual fetching of data for a 'Request' from a 'DataSource'.
  dataFetch :: (DataSource u r, Request r a) => r a -> GenHaxl u w a
  dataFetch = dataFetchWithInsert show DataCache.insert
  
  -- | Performs actual fetching of data for a 'Request' from a 'DataSource', using
  -- the given show functions for requests and their results.
  dataFetchWithShow
    :: (DataSource u r, Eq (r a), Hashable (r a), Typeable (r a))
    => ShowReq r a
    -> r a -> GenHaxl u w a
  dataFetchWithShow (showReq, showRes) = dataFetchWithInsert showReq
    (DataCache.insertWithShow showReq showRes)
  
  -- | Performs actual fetching of data for a 'Request' from a 'DataSource', using
  -- the given function to insert requests in the cache.
  dataFetchWithInsert
    :: 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 -> 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
    case res of
      -- This request has not been seen before
      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
          blockedFetchI = BlockedFetchInternal fid
          submitFetch = do
            let hint :: SchedulerHint r
                hint = schedulerHintState (stateGet states) userEnv
            case hint of
              SubmitImmediately ->
                performFetches env [BlockedFetches [blockedFetch] [blockedFetchI]]
              TryToBatch ->
                -- add the request to the RequestStore and continue
                modifyIORef' reqStoreRef $ \bs ->
                  addRequest blockedFetch blockedFetchI bs
            return $ Blocked ivar (Return ivar)
  
        -- if there is a fallback configured try that,
        -- else dispatch the fetch
        case dataCacheFetchFallback of
          Nothing -> submitFetch
          Just (DataCacheLookup dcl) -> do
            mbFallbackRes <- dcl req
            case mbFallbackRes of
              Nothing -> submitFetch
              Just fallbackRes -> do
                addFallbackResult env fallbackRes ivar
                ifReport flags ReportFetchStats $ addFallbackFetchStats
                  env
                  fid
                  req
                  fallbackRes
                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 fid -> do
        ifProfiling flags $ addProfileFetch env req fid True
        return $ Blocked ivar (Return ivar)
  
      -- Cached: either a result, or an exception
      Cached r fid -> do
        ifProfiling flags $ addProfileFetch env req fid True
        done env r
  
  -- | A data request that is not cached.  This is not what you want for
  -- normal read requests, because then multiple identical requests may
  -- return different results, and this invalidates some of the
  -- properties that we expect Haxl computations to respect: that data
  -- fetches can be arbitrarily reordered, and identical requests can be
  -- commoned up, for example.
  --
  -- 'uncachedRequest' is useful for performing writes, provided those
  -- are done in a safe way - that is, not mixed with reads that might
  -- conflict in the same Haxl computation.
  --
  -- if we are recording or running a test, we fallback to using dataFetch
  -- This allows us to store the request in the cache when recording, which
  -- allows a transparent run afterwards. Without this, the test would try to
  -- call the datasource during testing and that would be an exception.
  uncachedRequest
   :: forall a u w (r :: Type -> Type). (DataSource u r, Request r a)
   => r a -> GenHaxl u w a
  uncachedRequest req = do
    flg <- env flags
    if recording flg /= 0
      then dataFetch req
      else GenHaxl $ \e@Env{..} -> do
        ivar <- newIVar
        k <- nextCallId e
        let !rvar = stdResultVar e ivar (Proxy :: Proxy r)
        modifyIORef' reqStoreRef $ \bs ->
          addRequest (BlockedFetch req rvar) (BlockedFetchInternal k) bs
        return $ Blocked ivar (Return ivar)
  
  
  -- | Transparently provides caching. Useful for datasources that can
  -- return immediately, but also caches values.  Exceptions thrown by
  -- the IO operation (except for asynchronous exceptions) are
  -- propagated into the Haxl monad and can be caught by 'catch' and
  -- 'try'.
  cacheResult :: Request r a => r a -> IO a -> GenHaxl u w a
  cacheResult = cacheResultWithInsert show DataCache.insert
  
  -- | Transparently provides caching in the same way as 'cacheResult', but uses
  -- the given functions to show requests and their results.
  cacheResultWithShow
    :: (Eq (r a), Hashable (r a), Typeable (r a))
    => ShowReq r a -> r a -> IO a -> GenHaxl u w a
  cacheResultWithShow (showReq, showRes) = cacheResultWithInsert showReq
    (DataCache.insertWithShow showReq showRes)
  
  -- Transparently provides caching, using the given function to insert requests
  -- into the cache.
  cacheResultWithInsert
    :: Typeable (r a)
    => (r a -> String)    -- See Note [showFn]
    -> (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 $ \env@Env{..} -> do
    mbRes <- DataCache.lookup req dataCache
    case mbRes of
      Nothing -> do
        let
          getResult = do
            eitherResult <- Exception.try val
            case eitherResult of
              Left e -> rethrowAsyncExceptions e
              _ -> return ()
            return $ eitherToResultThrowIO eitherResult
        -- if there is a fallback configured try that
        result <- case dataCacheFetchFallback of
          Nothing -> getResult
          Just (DataCacheLookup dcl) -> do
            mbFallbackRes <- dcl req
            maybe getResult return mbFallbackRes
        ivar <- newFullIVar result
        k <- nextCallId env
        insertFn req (DataCacheItem ivar k) dataCache
        done env result
      Just (DataCacheItem IVar{ivarRef = cr} _) -> do
        e <- readIORef cr
        case e of
          IVarEmpty _ -> raise env corruptCache
          IVarFull r -> done env r
    where
      corruptCache = DataSourceError $ Text.concat
        [ Text.pack (showFn req)
        , " has a corrupted cache value: these requests are meant to"
        , " return immediately without an intermediate value. Either"
        , " the cache was updated incorrectly, or you're calling"
        , " cacheResult on a query that involves a blocking fetch."
        ]
  
  -- | Inserts a request/result pair into the cache. Throws an exception
  -- if the request has already been issued, either via 'dataFetch' or
  -- 'cacheRequest'.
  --
  -- This can be used to pre-populate the cache when running tests, to
  -- avoid going to the actual data source and ensure that results are
  -- deterministic.
  --
  cacheRequest
    :: Request req a => req a -> Either SomeException a -> GenHaxl u w ()
  cacheRequest request result = GenHaxl $ \e@Env{..} -> do
    mbRes <- DataCache.lookup request dataCache
    case mbRes of
      Nothing -> do
        cr <- newFullIVar (eitherToResult result)
        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.
      _other -> raise e $
        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. 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. Env u w -> [BlockedFetches u] -> IO ()
  performFetches env@Env{flags=f, statsRef=sref, statsBatchIdRef=sbref} jobs = do
    t0 <- getTimestamp
  
    ifTrace f 3 $
      forM_ jobs $ \(BlockedFetches reqs _) ->
        forM_ reqs $ \(BlockedFetch r _) -> putStrLn (showp r)
  
    let
      applyFetch i bfs@(BlockedFetches (reqs :: [BlockedFetch r]) _) =
        case stateGet (states env) of
          Nothing ->
            return (FetchToDo reqs (SyncFetch (mapM_ (setError e))))
           where
             e :: ShowP req => req a -> DataSourceError
             e req = DataSourceError $ "data source not initialized: " <> dsName
                    <> ": "
                    <> Text.pack (showp req)
          Just state ->
            return $ FetchToDo reqs
              $ (if testReportFlag ReportFetchStats $ report f
                  then wrapFetchInStats
                          (userEnv env)
                          sref
                          sbref
                          dsName
                          (length reqs)
                          bfs
                  else id)
              $ wrapFetchInTrace i (length reqs) dsName
              $ wrapFetchInCatch reqs
              $ fetch state f (userEnv env)
        where
          dsName = dataSourceName (Proxy :: Proxy r)
  
    fetches <- zipWithM applyFetch [0..] jobs
  
    scheduleFetches fetches (submittedReqsRef env) (flags env)
  
    t1 <- getTimestamp
    let roundtime = fromIntegral (t1 - t0) / 1000000 :: Double
  
    ifTrace f 1 $
      printf "Batch data fetch done (%.4fs)\n" (realToFrac roundtime :: Double)
  
  data FetchToDo where
    FetchToDo
      :: forall (req :: Type -> Type). (DataSourceName req, Typeable req)
      => [BlockedFetch req] -> PerformFetch req -> FetchToDo
  
  -- Catch exceptions arising from the data source and stuff them into
  -- the appropriate requests.  We don't want any exceptions propagating
  -- directly from the data sources, because we want the exception to be
  -- thrown by dataFetch instead.
  --
  wrapFetchInCatch :: [BlockedFetch req] -> PerformFetch req -> PerformFetch req
  wrapFetchInCatch reqs fetch =
    case fetch of
      SyncFetch f ->
        SyncFetch $ \reqs -> f reqs `Exception.catch` handler
      AsyncFetch f ->
        AsyncFetch $ \reqs io -> f reqs io `Exception.catch` handler
        -- this might be wrong: if the outer 'fio' throws an exception,
        -- then we don't know whether we have executed the inner 'io' or
        -- not.  If not, then we'll likely get some errors about "did
        -- not set result var" later, because we haven't executed some
        -- data fetches.  But we can't execute 'io' in the handler,
        -- because we might have already done it.  It isn't possible to
        -- do it completely right here, so we have to rely on data
        -- 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.
      BackgroundFetch f ->
        BackgroundFetch $ \reqs -> f reqs `Exception.catch` handler
    where
      handler :: SomeException -> IO ()
      handler e = do
        rethrowAsyncExceptions e
        mapM_ (forceError e) reqs
  
      -- Set the exception even if the request already had a result.
      -- Otherwise we could be discarding an exception.
      forceError e (BlockedFetch _ rvar) =
        putResult rvar (except e)
  
  
  data FailureCount = FailureCount
    { failureCountStandard :: {-# UNPACK #-} !Int
    , failureCountIgnored :: {-# UNPACK #-} !Int
    }
  
  #if __GLASGOW_HASKELL__ >= 804
  instance Semigroup FailureCount where
    FailureCount s1 i1 <> FailureCount s2 i2 = FailureCount (s1+s2) (i1+i2)
  #endif
  
  instance Monoid FailureCount where
    mempty = FailureCount 0 0
  #if __GLASGOW_HASKELL__ < 804
    mappend (FailureCount s1 i1) (FailureCount s2 i2)
      = FailureCount (s1+s2) (i1+i2)
  #endif
  
  wrapFetchInStats
    :: DataSource u req
    => u
    -> IORef Stats
    -> IORef Int
    -> Text
    -> Int
    -> BlockedFetches u
    -> PerformFetch req
    -> PerformFetch req
  wrapFetchInStats
    u
    !statsRef
    !batchIdRef
    dataSource
    batchSize
    (BlockedFetches _reqs reqsI)
    perform = do
    case perform of
      SyncFetch f ->
        SyncFetch $ \reqs -> do
          bid <- newBatchId
          fail_ref <- newIORef mempty
          (t0,t,alloc,_) <- statsForIO (f (map (addFailureCount u fail_ref)
            (reqsWithFetchDsStats bid reqs)))
          failures <- readIORef fail_ref
          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 mempty
          let inner' = do
                (_,t,alloc,_) <- statsForIO inner
                writeIORef inner_r (t,alloc)
              reqs' = map (addFailureCount u fail_ref) reqs
              reqs'' = reqsWithFetchDsStats bid reqs'
          (t0, totalTime, totalAlloc, _) <- statsForIO (f reqs'' inner')
          (innerTime, innerAlloc) <- readIORef inner_r
          failures <- readIORef fail_ref
          updateFetchStats bid allFids t0 (totalTime - innerTime)
            (totalAlloc - innerAlloc) batchSize failures
      BackgroundFetch io -> do
        BackgroundFetch $ \reqs -> do
          bid <- newBatchId
          startTime <- getTimestamp
          io (reqsWithFetchDsStats bid
            (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)
      reqsWithFetchDsStats = \bid reqs
        -> zipWith (addFetchDatasourceStats bid) reqs reqsI
      addTimer
        u
        bid
        t0
        (BlockedFetch req (ResultVar fn))
        (BlockedFetchInternal fid) =
          BlockedFetch req $ ResultVar $ \result isChildThread stats -> 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 stats
  
      addFetchDatasourceStats
        :: Int
        -> BlockedFetch r
        -> BlockedFetchInternal
        -> BlockedFetch r
      addFetchDatasourceStats bid
        (BlockedFetch req (ResultVar fn))
        (BlockedFetchInternal fid) = BlockedFetch req $ ResultVar
          $ \result isChildThread stats -> do
            let mkStats dss = FetchDataSourceStats
                  { fetchDsStatsCallId = fid
                  , fetchDsStatsDataSource = dataSource
                  , fetchDsStatsStats = dss
                  , fetchBatchId = bid
                  }
            case stats of
              Just dss -> atomicModifyIORef' statsRef
                $ \(Stats fs) -> (Stats (mkStats dss : fs), ())
              Nothing -> return ()
            fn result isChildThread stats
  
  
      updateFetchStats
        :: 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 = failureCountStandard
                              , fetchIgnoredFailures = failureCountIgnored
                              , fetchBatchId = bid
                              , fetchIds = fids }
        atomicModifyIORef' statsRef $ \(Stats fs) -> (Stats (this : fs), ())
  
      addFailureCount :: DataSource u r
        => u -> IORef FailureCount -> BlockedFetch r -> BlockedFetch r
      addFailureCount u ref (BlockedFetch req (ResultVar fn)) =
        BlockedFetch req $ ResultVar $ \result isChildThread stats -> do
          let addFailures r = (r <> calcFailure u req result, ())
          when (isLeft result) $ atomicModifyIORef' ref addFailures
          fn result isChildThread stats
  
  wrapFetchInTrace
    :: Int
    -> Int
    -> Text
    -> PerformFetch req
    -> PerformFetch req
  #ifdef EVENTLOG
  wrapFetchInTrace i n dsName f =
    case f of
      SyncFetch io -> SyncFetch (wrapF "Sync" io)
      AsyncFetch fio -> AsyncFetch (wrapF "Async" . fio . unwrapF "Async")
    where
      d = Text.unpack dsName
      wrapF :: String -> IO a -> IO a
      wrapF ty = bracket_ (traceEventIO $ printf "START %d %s (%d %s)" i d n ty)
                          (traceEventIO $ printf "STOP %d %s (%d %s)" i d n ty)
      unwrapF :: String -> IO a -> IO a
      unwrapF ty = bracket_ (traceEventIO $ printf "STOP %d %s (%d %s)" i d n ty)
                            (traceEventIO $ printf "START %d %s (%d %s)" i d n ty)
  #else
  wrapFetchInTrace _ _ _ f = f
  #endif
  
  time :: IO a -> IO (Timestamp,Microseconds,a)
  time io = do
    t0 <- getTimestamp
    a <- io
    t1 <- getTimestamp
    return (t0, t1 - t0, a)
  
  -- | 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 ()
  scheduleFetches fetches ref flags = do
    -- update ReqCountmap for these fetches
    ifReport flags ReportOutgoneFetches $ sequence_
      [ atomicModifyIORef' ref $
          \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
    async_fetches sync_fetches
   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]
  
    async_fetches :: IO () -> IO ()
    async_fetches = compose
      [f reqs | FetchToDo reqs (AsyncFetch f) <- fetches]
  
    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
  