Haxl.Core.DataSource

The DataSource class and related types and functions. This module is provided for access to Haxl internals only; most users should import Haxl.Core instead.

Data fetching

class (DataSourceName req, StateKey req, ShowP req) => DataSource u req where

The class of data sources, parameterised over the request type for that data source. Every data source must implement this class.

A data source keeps track of its state by creating an instance of StateKey to map the request type to its state. In this case, the type of the state should probably be a reference type of some kind, such as IORef.

For a complete example data source, see Examples.

Minimal complete definition

fetch

Methods

fetch

Arguments

:: State req

Current state.

-> Flags

Tracing flags.

-> u

User environment.

-> PerformFetch req

Fetch the data; see PerformFetch.

Issues a list of fetches to this DataSource. The BlockedFetch objects contain both the request and the ResultVars into which to put the results.

schedulerHint :: u -> SchedulerHint req

Instances

(Typeable * tag, ShowP (ConcurrentIOReq tag), ConcurrentIO tag) => DataSource u (ConcurrentIOReq tag)

Methods

fetch :: State (ConcurrentIOReq tag) -> Flags -> u -> PerformFetch (ConcurrentIOReq tag)

schedulerHint :: u -> SchedulerHint (ConcurrentIOReq tag)

class DataSourceName (req :: * -> *) where

Minimal complete definition

dataSourceName

Methods

dataSourceName :: Proxy req -> Text

The name of this DataSource, used in tracing and stats. Must take a dummy request.

Instances

Typeable * tag => DataSourceName (ConcurrentIOReq tag)

Methods

dataSourceName :: Proxy (* -> *) (ConcurrentIOReq tag) -> Text

type Request req a = (Eq (req a), Hashable (req a), Typeable (req a), Show (req a), Show a)

A convenience only: package up Eq, Hashable, Typeable, and Show for requests into a single constraint.

data BlockedFetch r

A BlockedFetch is a pair of

• The request to fetch (with result type a)
• A ResultVar to store either the result or an error

We often want to collect together multiple requests, but they return different types, and the type system wouldn't let us put them together in a list because all the elements of the list must have the same type. So we wrap up these types inside the BlockedFetch type, so that they all look the same and we can put them in a list.

When we unpack the BlockedFetch and get the request and the ResultVar out, the type system knows that the result type of the request matches the type parameter of the ResultVar, so it will let us take the result of the request and store it in the ResultVar.

Constructors

BlockedFetch (r a) (ResultVar a)

data PerformFetch req

A data source can fetch data in one of four ways.

Constructors

SyncFetch ([BlockedFetch req] -> IO ())

Fully synchronous, returns only when all the data is fetched. See syncFetch for an example.

AsyncFetch ([BlockedFetch req] -> IO () -> IO ())

Asynchronous; performs an arbitrary IO action while the data is being fetched, but only returns when all the data is fetched. See asyncFetch for an example.

BackgroundFetch ([BlockedFetch req] -> IO ())

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.

data SchedulerHint (req :: * -> *)

Hints to the scheduler about this data source

Constructors

TryToBatch

Hold data-source requests while we execute as much as we can, so that we can hopefully collect more requests to batch.

SubmitImmediately

Submit a request via fetch as soon as we have one, don't try to batch multiple requests. This is really only useful if the data source returns BackgroundFetch, otherwise requests to this data source will be performed synchronously, one at a time.

Result variables

newtype ResultVar a

A sink for the result of a data fetch in BlockedFetch

Constructors

ResultVar (Either SomeException a -> Bool -> IO ())

mkResultVar :: (Either SomeException a -> Bool -> IO ()) -> ResultVar a

putFailure :: Exception e => ResultVar a -> e -> IO ()

putResult :: ResultVar a -> Either SomeException a -> IO ()

putResultFromChildThread :: ResultVar a -> Either SomeException a -> IO ()

Like putResult, but used to get correct accounting when work is being done in child threads. This is particularly important for data sources that are using BackgroundFetch, The allocation performed in the child thread up to this point will be propagated back to the thread that called runHaxl.

Note: if you're doing multiple putResult calls in the same thread ensure that only the last one is putResultFromChildThread. If you make multiple putResultFromChildThread calls, the allocation will be counted multiple times.

If you are reusing a thread for multiple fetches, you should call System.Mem.setAllocationCounter 0 after putResultFromChildThread, so that allocation is not counted multiple times.

putSuccess :: ResultVar a -> a -> IO ()

Default fetch implementations

asyncFetch

Arguments

:: ((service -> IO ()) -> IO ())

Wrapper to perform an action in the context of a service.

-> (service -> IO ())

Dispatch all the pending requests and wait for the results

-> (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

asyncFetchWithDispatch

Arguments

:: ((service -> IO ()) -> IO ())

Wrapper to perform an action in the context of a service.

-> (service -> IO ())

Dispatch all the pending requests

-> (service -> IO ())

Wait for the results

-> (forall a. service -> request a -> IO (IO (Either SomeException a)))

Enqueue an individual request to the service.

-> State request

Currently unused.

-> Flags

Currently unused.

-> u

Currently unused.

-> PerformFetch request

Common implementation templates for fetch of DataSource.

Example usage:

fetch = syncFetch MyDS.withService MyDS.retrieve
  $ \service request -> case request of
    This x -> MyDS.fetchThis service x
    That y -> MyDS.fetchThat service y

asyncFetchAcquireRelease

Arguments

:: IO service

Resource acquisition for this datasource

-> (service -> IO ())

Resource release

-> (service -> IO ())

Dispatch all the pending requests and wait for the results

-> (service -> IO ())

Wait for the results

-> (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

A version of asyncFetch (actually asyncFetchWithDispatch) that handles exceptions correctly. You should use this instead of asyncFetch or asyncFetchWithDispatch. The danger with asyncFetch is that if an exception is thrown by withService, the inner action won't be executed, and we'll drop some data-fetches in the same round.

asyncFetchAcquireRelease behaves like the following:

asyncFetchAcquireRelease acquire release dispatch wait enqueue =
  AsyncFetch $ \requests inner ->
    bracket acquire release $ \service -> do
      getResults <- mapM (submitFetch service enqueue) requests
      dispatch service
      inner
      wait service
      sequence_ getResults

except that inner is run even if acquire, enqueue, or dispatch throws, unless an async exception is received.

stubFetch :: Exception e => (forall a. r a -> e) -> State r -> Flags -> u -> PerformFetch r

syncFetch

Arguments

:: ((service -> IO ()) -> IO ())

Wrapper to perform an action in the context of a service.

-> (service -> IO ())

Dispatch all the pending requests and wait for the results

-> (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

Utilities

except :: Exception e => e -> Either SomeException a

setError :: Exception e => (forall a. r a -> e) -> BlockedFetch r -> IO ()

Function for easily setting a fetch to a particular exception