DDTrace.Task (dd_trace_ex v0.1.0)

Copy Markdown View Source

Task, with the caller's trace carried into the child.

One alias is the whole migration for a file: every function Task has, this module has, with the same name, the same arguments in the same order, the same defaults, and the same return value.

alias DDTrace.Task

DDTrace.trace "assets.fetch_all" do
  assets
  |> Task.async_stream(&fetch_asset/1, max_concurrency: 10)
  |> Enum.to_list()
end

Every span fetch_asset/1 opens lands under assets.fetch_all, in the same trace, and Task.await/2 and the rest behave as they always did.

What is different

Only the spawning functions — async/1,3, async_stream/3,5, start/1,3 and start_link/1,3. Each takes a DDTrace.current_context/0 snapshot in the caller, at the call, and attaches it in the child around the work, with DDTrace.with_context/2's restore semantics. A caller with no trace running hands over nothing, and the child runs exactly as a bare Task child would.

Taking the snapshot at the call is what makes the attribution right rather than merely plausible: a stream built inside one trace and enumerated somewhere else entirely still belongs to the trace that built it, because the snapshot was taken when the stream was constructed and every element shares it. Looking the trace up when a child happens to run would attribute the work to whatever the spawning process is doing by then.

await/2, yield/2 and everything else on the awaiting and managing side is Task, delegated: this module adds no behaviour there and takes none away.

Raising

This module raises where Task raises, on the same arguments, in the same process — that is what a drop-in means, and it is the one place the library's usual "nothing raises" promise does not reach. Task.async(42) raises FunctionClauseError at the call site; so does this. await/2 raises on a timeout; so does this.

The tracing this module adds never raises: a snapshot that is nil, or something that is not a snapshot at all, costs a line in the log and the child's place in the trace — never the child.

What is deliberately not here

There is no async_with_span or any other span-opening variant. trace is the one verb that opens a span, and it is written inside the function like anywhere else — a second span-opening surface bolted onto a spawn would be two ways to say the same thing.

There is no way to look a parent up from inside the child either. A child process can see who spawned it, but not what that process was doing at the time — and "what was current when this was spawned" is the only thing a parent span can honestly mean. Looking it up when the child happens to run attributes work to whatever the spawning process has moved on to. So the snapshot travels forwards, taken at the call, and nothing reaches backwards.

Notes worth having read

%Task{} is still Elixir's struct. Aliasing this module shadows the name Task for calls but a pattern like %Task{ref: ref} needs the real one, so spell it %Elixir.Task{} — or match on the field you want from the struct the call returned, which is what most code does.

child_spec/1 is a plain delegate, so {DDTrace.Task, fun} in a supervision tree starts a bare Task. Nothing is lost: a supervisor builds its tree at boot, before any trace exists, and there would be no snapshot to take.

An MFA child is spawned as DDTrace.Task.Wrapper, so that is the initial call :proc_lib reports in a crash report and in Process.info/2. The real module, function and arguments are its arguments, one frame in.

Summary

Types

t()

A task, exactly Task.t/0 — this module returns Elixir's own struct.

Functions

Starts a task carrying this process's trace, to be awaited on.

Starts a task carrying this process's trace, to be awaited on.

Runs fun over enumerable concurrently, each element carrying this process's trace.

Runs module.function_name(element, ...args) over enumerable concurrently, each element carrying this process's trace.

Awaits a task's reply. See Task.await/2.

Awaits a task's reply. See Task.await/2.

Awaits replies from several tasks. See Task.await_many/2.

Awaits replies from several tasks. See Task.await_many/2.

A child specification for a supervision tree. See Task.child_spec/1.

A task that has already finished. See Task.completed/1.

Discards a task's reply. See Task.ignore/1.

Stops a task and its linked process. See Task.shutdown/2.

Stops a task and its linked process. See Task.shutdown/2.

Starts an unlinked, unawaited task carrying this process's trace.

Starts an unlinked, unawaited task carrying this process's trace.

Starts a linked task carrying this process's trace.

Starts a linked task carrying this process's trace.

Temporarily awaits a task's reply. See Task.yield/2.

Temporarily awaits a task's reply. See Task.yield/2.

Temporarily awaits several tasks' replies. See Task.yield_many/2.

Temporarily awaits several tasks' replies. See Task.yield_many/2.

Types

t()

@type t() :: Task.t()

A task, exactly Task.t/0 — this module returns Elixir's own struct.

Functions

async(fun)

@spec async((-> any())) :: t()

Starts a task carrying this process's trace, to be awaited on.

See Task.async/1. The snapshot is taken here, in the caller; spans the function opens are children of the span current at this call.

Examples

task = DDTrace.Task.async(fn -> heavy() end)
DDTrace.Task.await(task)

async(module, function_name, args)

@spec async(module(), atom(), [term()]) :: t()

Starts a task carrying this process's trace, to be awaited on.

See Task.async/3. The child is spawned as an MFA, as asked — the snapshot travels beside it rather than closing over it.

Examples

task = DDTrace.Task.async(Reports, :build, [account])

async_stream(enumerable, fun, options \\ [])

@spec async_stream(Enumerable.t(), (term() -> term()), keyword()) :: Enumerable.t()

Runs fun over enumerable concurrently, each element carrying this process's trace.

See Task.async_stream/3. The snapshot is taken once, when the stream is built, and shared by every element — so a stream enumerated later, or in another process, still belongs to the trace that built it.

Examples

assets
|> DDTrace.Task.async_stream(&fetch_asset/1, max_concurrency: 10)
|> Enum.to_list()

async_stream(enumerable, module, function_name, args, options \\ [])

@spec async_stream(Enumerable.t(), module(), atom(), [term()], keyword()) ::
  Enumerable.t()

Runs module.function_name(element, ...args) over enumerable concurrently, each element carrying this process's trace.

See Task.async_stream/5. The snapshot is taken once, when the stream is built.

Examples

DDTrace.Task.async_stream(orders, Orders, :settle, [rates])

await(task)

@spec await(t()) :: term()

Awaits a task's reply. See Task.await/2.

await(task, timeout)

@spec await(t(), timeout()) :: term()

Awaits a task's reply. See Task.await/2.

await_many(tasks)

@spec await_many([t()]) :: [term()]

Awaits replies from several tasks. See Task.await_many/2.

await_many(tasks, timeout)

@spec await_many([t()], timeout()) :: [term()]

Awaits replies from several tasks. See Task.await_many/2.

child_spec(arg)

@spec child_spec(term()) :: Supervisor.child_spec()

A child specification for a supervision tree. See Task.child_spec/1.

Delegated whole, so a supervised child starts a bare Task. A supervisor builds its tree before there is any trace to carry, so there is nothing here to snapshot.

completed(result)

@spec completed(term()) :: t()

A task that has already finished. See Task.completed/1.

ignore(task)

@spec ignore(t()) :: {:ok, term()} | {:exit, term()} | nil

Discards a task's reply. See Task.ignore/1.

shutdown(task)

@spec shutdown(t()) :: {:ok, term()} | {:exit, term()} | nil

Stops a task and its linked process. See Task.shutdown/2.

shutdown(task, shutdown)

@spec shutdown(t(), timeout() | :brutal_kill) :: {:ok, term()} | {:exit, term()} | nil

Stops a task and its linked process. See Task.shutdown/2.

start(fun)

@spec start((-> any())) :: {:ok, pid()}

Starts an unlinked, unawaited task carrying this process's trace.

See Task.start/1.

Examples

DDTrace.Task.start(fn -> log_asynchronously(event) end)

start(module, function_name, args)

@spec start(module(), atom(), [term()]) :: {:ok, pid()}

Starts an unlinked, unawaited task carrying this process's trace.

See Task.start/3.

start_link(fun)

@spec start_link((-> any())) :: {:ok, pid()}

Starts a linked task carrying this process's trace.

See Task.start_link/1.

Examples

DDTrace.Task.start_link(fn -> warm_cache() end)

start_link(module, function, args)

@spec start_link(module(), atom(), [term()]) :: {:ok, pid()}

Starts a linked task carrying this process's trace.

See Task.start_link/3.

yield(task)

@spec yield(t()) :: {:ok, term()} | {:exit, term()} | nil

Temporarily awaits a task's reply. See Task.yield/2.

yield(task, timeout)

@spec yield(t(), timeout()) :: {:ok, term()} | {:exit, term()} | nil

Temporarily awaits a task's reply. See Task.yield/2.

yield_many(tasks)

@spec yield_many([t()]) :: [{t(), {:ok, term()} | {:exit, term()} | nil}]

Temporarily awaits several tasks' replies. See Task.yield_many/2.

yield_many(tasks, opts)

@spec yield_many([t()], timeout() | keyword()) :: [
  {t(), {:ok, term()} | {:exit, term()} | nil}
]

Temporarily awaits several tasks' replies. See Task.yield_many/2.