defmodule ZenMonitor.Local do
@moduledoc """
ZenMonitor.Local
Most of the actual logic of monitoring and fan-out is handled by `ZenMonitor.Local.Connector`,
see that module for more information.
`ZenMonitor.Local` is responsible for monitoring the subscribing local processes and cleaning up
monitors if they crash.
"""
use GenStage
use Instruments.CustomFunctions, prefix: "zen_monitor.local"
alias ZenMonitor.Local.{Connector, Tables}
@typedoc """
Effective compatibility of a remote node
"""
@type compatibility :: :compatible | :incompatible
@typedoc """
Represents a future down dispatch for a given pid to be delivered by
`ZenMonitor.Local.Dispatcher`
"""
@type down_dispatch :: {pid, {:DOWN, reference, :process, pid, {:zen_monitor, any}}}
@subscribers_table Module.concat(__MODULE__, "Subscribers")
@hibernation_threshold 1_000
defmodule State do
@moduledoc """
Maintains the internal state for ZenMonitor.Local
- `subscribers` is an ETS table that tracks local subscribers to prevent multiple monitors
- `batch` is the queue of messages awaiting delivery to ZenMonitor.Local.Dispatcher
- `length` is the current length of the batch queue (calculating queue length is an O(n)
operation, it is simple to track it as elements are added / removed)
- `queue_emptied` is the number of times the queue has been emptied. Once this number
exceeds the hibernation_threshold (see `hibernation_threshold/0`) the process will
hibernate
"""
@type t :: %__MODULE__{
subscribers: :ets.tid(),
length: integer,
queue_emptied: integer,
batch: :queue.queue()
}
defstruct [
:subscribers,
length: 0,
queue_emptied: 0,
batch: :queue.new()
]
end
## Delegates
defdelegate compatibility_for_node(remote), to: ZenMonitor.Local.Connector, as: :compatibility
## Client
def start_link(_opts \\ []) do
GenStage.start_link(__MODULE__, [], name: __MODULE__)
end
@doc """
Begin monitoring the given process
Has the same semantics as `Process.monitor/1`, DOWN messages will be delivered
at a pace controlled by the :zen_monitor, :demand_interval and
:zen_monitor, :demand_amount environment variables
"""
@spec monitor(target :: ZenMonitor.destination()) :: reference
def monitor(target) do
increment("monitor")
ref = make_ref()
me = self()
# Write the reference out
:ets.insert(Tables.references(), {{me, ref}, target})
# Enqueue the monitor into the Connector for async monitor
Connector.monitor(target, ref, me)
# Perform reciprocal monitoring (if needed)
unless :ets.member(@subscribers_table, me) do
GenStage.cast(__MODULE__, {:monitor_subscriber, me})
end
# Return the reference to the caller
ref
end
@doc """
Stop monitoring a process by monitor reference
Has the same semantics as `Process.demonitor/2` (although you can pass the `:info` option, it
has no effect and is not honored, `:flush` is honored)
To demonitor a process you should pass in the reference returned from
`ZenMonitor.Local.monitor/1` for the given process
"""
@spec demonitor(ref :: reference, options :: [:flush]) :: true
def demonitor(ref, options \\ []) when is_reference(ref) do
increment("demonitor")
me = self()
# First consume the reference
case :ets.take(Tables.references(), {me, ref}) do
[] ->
# Unknown reference, maybe it's been dispatched, consume any :DOWN messages in the inbox
# if :flush is provided. Dispatch atomically consumes the reference, which is why we only
# need to scan the inbox if we don't find a reference.
if :flush in options do
receive do
{:DOWN, ^ref, _, _, _} -> nil
after
0 ->
nil
end
end
:ok
[{{^me, ^ref}, pid}] ->
# Instruct the Connector to demonitor the monitor
Connector.demonitor(pid, ref)
end
true
end
@doc """
Check the compatiblity of the remote node that owns the provided destination
This is a simple convenience function that looksup the node for the destination and then calls
`ZenMonitor.Local.compatiblity_for_node/1`
"""
@spec compatibility(target :: ZenMonitor.destination()) :: compatibility
def compatibility(target) do
target
|> ZenMonitor.find_node()
|> compatibility_for_node()
end
@doc """
Asynchronously enqueue a list of down dispatches for delivery by the Dispatcher
If called with the empty list, cast will be suppressed.
"""
@spec enqueue(messages :: [down_dispatch]) :: :ok
def enqueue([]), do: :ok
def enqueue(messages) do
GenStage.cast(__MODULE__, {:enqueue, messages})
end
@doc """
Synchronously checks the length of the ZenMonitor.Local's internal batch
"""
@spec batch_length() :: integer()
def batch_length do
GenStage.call(__MODULE__, :batch_length)
end
@doc """
Gets the hibernation threshold from the Application Environment
Every time the demand empties the queue a counter is incremented. When this counter exceeds the
hibernation threshold the ZenMonitor.Local process will be sent into hibernation. See
ZenMonitor.Local's @hibernation_threshold for the default value
This can be controlled at boot and runtime with the {:zen_monitor, :hibernation_threshold}
setting, see ZenMonitor.Local.hibernation_threshold/1 for runtime convenience functionality.
"""
@spec hibernation_threshold() :: integer
def hibernation_threshold do
Application.get_env(:zen_monitor, :hibernation_threshold, @hibernation_threshold)
end
@doc """
Puts the hibernation threshold into the Application Environment
This is a simple convenience function for overwriting the
{:zen_monitor, :hibernation_threshold} setting at runtime.
"""
@spec hibernation_threshold(value :: integer) :: :ok
def hibernation_threshold(value) do
Application.put_env(:zen_monitor, :hibernation_threshold, value)
end
## Server
def init(_opts) do
Process.flag(:message_queue_data, :off_heap)
subscribers =
:ets.new(@subscribers_table, [:protected, :named_table, :set, read_concurrency: true])
{:producer, %State{subscribers: subscribers}}
end
@doc """
Handles demand from `ZenMonitor.Local.Dispatcher`
ZenMonitor.Local maintains a queue of pending messages to be sent to local processes, the actual
dispatch of which are throttled by ZenMonitor.Local.Dispatcher. When
ZenMonitor.Local.Dispatcher requests more messages to dispatch, this handler will collect up to
the requested amount from the batch queue to satisfy the demand.
"""
def handle_demand(demand, %State{length: length} = state) do
if length <= demand do
empty_queue(state)
else
chunk_queue(demand, state)
end
end
# Handle a local subscriber going down
# When a process establishes a remote monitor, ZenMonitor.Local establishes a reciprocal monitor,
# see monitor/1 and handle_cast({:monitor_subscriber, ...}) for more information.
# If the subscriber crashes, all of the ETS records maintained by ZenMonitor.Local and the various
# ZenMonitor.Local.Connectors is no longer needed and will be cleaned up by this handler.
def handle_info(
{:DOWN, _ref, :process, subscriber, _reason},
%State{subscribers: subscribers} = state
) do
for [ref, remote_pid] <- :ets.match(Tables.references(), {{subscriber, :"$1"}, :"$2"}) do
# Remove the reference
:ets.delete(Tables.references(), {subscriber, ref})
# Instruct the Connector to demonitor
Connector.demonitor(remote_pid, ref)
end
# Remove the subscriber from the subscribers table
:ets.delete(subscribers, subscriber)
{:noreply, [], state}
end
# Handles recipricol subscriber monitoring
# When a process establishes a remote monitor, ZenMonitor.Local will establish a reciprocal
# monitor on the subscriber. This is done so that appropriate cleanup can happen if the
# subscriber goes down.
# This handler guarantees that a local subscriber will only ever have one active reciprocal
# monitor at a time by tracking the subscribers in an ETS table.
def handle_cast({:monitor_subscriber, subscriber}, %State{subscribers: subscribers} = state) do
if :ets.insert_new(subscribers, {subscriber}) do
Process.monitor(subscriber)
end
{:noreply, [], state}
end
# Handles enqueuing messages for eventual dispatch
# ZenMonitor.Local.Connector is responsible for generating down dispatches and enqueuing them with
# ZenMonitor.Local. ZenMonitor.Local takes these messages and places them into the
# batch queue to be delivered to ZenMonitor.Local.Dispatcher as demanded.
def handle_cast({:enqueue, messages}, %State{batch: batch, length: length} = state) do
{batch, new_length} =
messages
|> Enum.reduce({batch, length}, fn item, {acc, len} ->
{:queue.in(item, acc), len + 1}
end)
increment("enqueue", new_length - length)
{:noreply, [], %State{state | batch: batch, length: new_length}}
end
# Handles batch length checks
# Returns the current length of the batch
def handle_call(:batch_length, _from, %State{length: length} = state) do
{:reply, length, [], state}
end
## Private
@spec empty_queue(state :: State.t()) ::
{:noreply, [down_dispatch], State.t()}
| {:noreply, [down_dispatch], State.t(), :hibernate}
defp empty_queue(%State{queue_emptied: queue_emptied, batch: batch} = state) do
new_queue_emptied = queue_emptied + 1
response = :queue.to_list(batch)
if new_queue_emptied >= hibernation_threshold() do
{:noreply, response, %State{state | batch: :queue.new(), length: 0, queue_emptied: 0},
:hibernate}
else
{:noreply, response,
%State{state | batch: :queue.new(), length: 0, queue_emptied: new_queue_emptied}}
end
end
@spec chunk_queue(size :: integer(), state :: State.t()) ::
{:noreply, [down_dispatch], State.t()}
defp chunk_queue(size, %State{batch: batch, length: length} = state) do
{messages, new_batch} = :queue.split(size, batch)
{:noreply, :queue.to_list(messages), %State{state | batch: new_batch, length: length - size}}
end
end
