bricks

v0.1.0

bricks v0.1.0 Bricks.Socket behaviour View Source

A Socket represents a connected Socket - that is it is manufactured after a connection has been established.

As in Erlang/OTP, a Socket is owned by a single process, the owner. No other process may interact with the Socket, but the owner can hand off ownership to another process. If the owner dies, the socket will be closed.

A Socket is at any given time (as per OTP) in one of these modes:

  • passive mode - data can only be received when you call recv.

  • active mode - the socket transforms packets into messages as they arrive and transmits them to the owner.

  • bounded active mode - like active mode, but caps the number of packets that will be buffered in the owner mailbox before requiring a reset by the owner.

You should not expect a Socket to be in a particular mode, rather you should explicitly set it when you receive the Socket with set_active/2. The default activity depends upon the type of Socket and probably the OTP release you’re running under.

Picking a mode

  • passive mode: a safe default when you don’t have aggressive performance requirements. The kernel will buffer some amount of data and the rest will be rejected via (TCP) backpressure.

  • active mode: the process mailbox becomes an effectively unlimited sized buffer. If you fail to keep up, you might buffer until the BEAM has eaten all of the memory and gets killed. But while it doesn’t, it’s low latency and high throughput. WARNING: I hope you know what you’re doing, use bounded active

  • bounded active mode: like active mode, but caps the number of packets that will be buffered in the owner mailbox. Achieves better performance than passive mode but without the risks associated with active mode.

Bounded Active Mode

In addition to taking a boolean value, set_active can take an integer, entering what I call bounded active mode (BAM). BAM provides most of the performance benefits of active mode with the security of not having your BEAM get OOMkilled from overbuffering. If passive mode doesn’t meet your needs, use BAM - avoid OOM!

In BAM, the socket behaves as in active mode, sending messages that are received to the owner. However, it also maintains an internal active window counter of the number of packets that may be received in active mode, after which the socket is placed into passive mode and a notification message is sent to the owner who can put the socket back into BAM with set_active/2.

The bam_window field in the Socket struct holds the intended active window. It is used by extend_active/1 to easily reset BAM.

BAM and setting active

set_active/2 can take the following values:

  • true - enable active mode
  • false - enable passive mode
  • :once - enable BAM for one packet
  • integer() - adjust the internal active window counter by this many, send a message to notify the owner when the socket is made passive

Note that while the first three all set the value, providing an integer can behave in two ways, depending on the current mode:

  • You are not in BAM, the active counter is set to this value. If it is lower than zero, the socket is made passive.

  • You are in BAM, the active counter is adjusted by this value (by addition). If you pass a negative number and cause the counter to go 0 or lower, the socket is made passive.

We recommend only using set_active/2 with a positive integer and when the socket is in passive mode (such as when you have been informed the socket has just been made passive). This simplifies the problem and you can pretend that it does set the internal counter.

If you ignore the last paragraph, you’ll have to keep track of the current active window to know what the new one will be after calling set_active/2. The (undocumented) decr_active/1 function may be useful to call when you receive a packet.

This behaviour is inherited from OTP. We would not choose to implement it this way ourselves.

Link to this section Summary

Types

active()

The activity mode of the socket. See module documentation

data()

The type of data received from the socket. Depends upon socket binary mode

fetch_active_error()

The errors fetch_active/1 may return

fetch_active_return()

The return type of fetch_active/1

handoff_error()

The errors handoff/2 may return

handoff_return()

The return type of handoff/2

host()

A host to connect to (or a filepath to a unix socket)

ip()

An IPv4 or IPv6 address

ipv4()

An IPv4 address

ipv6()

An IPv6 address

new_error()

The errors that new/1 may return

new_opts()

Options provided to new/1

port_num()

A port number

recv_error()

The types of error recv/1, recv/2 and recv/3 may return

recv_result()

The return of recv/1, recv/2 and recv/3

send_error()

The errors send_data/2 may return

send_result()

The return type of send_data/2

set_active_return()

The return type of set_active/2

t()

A connected socket

window()

The activity mode used for extending Bounded Active Mode

Functions

close(socket)

Closes the socket. Always returns :ok

extend_active(socket)

Extends the bounded active mode (BAM) of a socket by its bam_window

extend_active(socket, window)

Extends the active of a socket by the given window

fetch_active(socket)

Fetches the current activity of the socket

handoff(socket, pid)

Hands a Socket off to a new process, which becomes the owner

new(opts)

Creates a new Socket from a map of options

passify(socket)

Turns the socket passive, clearing any active data out of the mailbox. and returning it as one binary Note: Assumes binary mode!

recv(socket)

Receives all available data from the socket. Times out if the Socket’s receive_timeout is breached

recv(socket, size)

Receives data from the socket

recv(socket, size, timeout)

Receives data from the socket

send_data(socket, data)

Sends the given iolist data down the socket

set_active(socket, active)

Changes the socket’s activity mode. Valid activities:

  • true - enable active mode
  • false - enable passive mode
  • :once - enable BAM for one packet
  • integer() - adjust the internal active window counter by this many

Callbacks

close(t)

Closes the socket

fetch_active(t)

Fetches the active mode of the socket

handoff(t, pid)

Makes another process the new owner of the Socket

recv(t, non_neg_integer, timeout)

Receive from the Socket

send_data(t, iodata)

Send the data down the socket

set_active(t, active)

Set the activity mode of the socket

Link to this section Types

Link to this type active() View Source 
active() :: boolean() | :once | -32768..32767

The activity mode of the socket. See module documentation.

Link to this type data() View Source 
data() :: binary() | charlist()

The type of data received from the socket. Depends upon socket binary mode

Link to this type fetch_active_error() View Source 
fetch_active_error() :: Bricks.Error.Closed.t() | Bricks.Error.Posix.t()

The errors fetch_active/1 may return

Link to this type fetch_active_return() View Source 
fetch_active_return() :: {:ok, active()} | {:error, fetch_active_error()}

The return type of fetch_active/1

Link to this type handoff_return() View Source 
handoff_return() :: {:ok, t()} | {:error, handoff_error()}

The return type of handoff/2

Link to this type host() View Source 
host() :: binary() | ipv4() | ipv6()

A host to connect to (or a filepath to a unix socket)

Link to this type ip() View Source 
ip() :: ipv4() | ipv6()

An IPv4 or IPv6 address

Link to this type ipv4() View Source 
ipv4() :: {byte(), byte(), byte(), byte()}

An IPv4 address

Link to this type ipv6() View Source 
ipv6() :: {char(), char(), char(), char(), char(), char(), char(), char()}

An IPv6 address

Link to this type new_error() View Source 
new_error() :: Bricks.Error.BadOption.t() | Bricks.Error.UnknownOptions.t()

The errors that new/1 may return

Link to this type new_opts() View Source 
new_opts() :: %{
  :module => atom(),
  :host => host(),
  :port => port_num(),
  :handle => term(),
  :active => active(),
  :data_tag => term(),
  :error_tag => term(),
  :closed_tag => term(),
  :passive_tag => term(),
  optional(:bam_window) => window(),
  optional(:receive_timeout) => timeout() | nil
}

Options provided to new/1

Link to this type port_num() View Source 
port_num() :: pos_integer()

A port number

Link to this type recv_error() View Source 
recv_error() :: Bricks.Error.Closed.t() | Bricks.Error.Posix.t()

The types of error recv/1, recv/2 and recv/3 may return

Link to this type recv_result() View Source 
recv_result() :: {:ok, data(), t()} | {:error, recv_error()}

The return of recv/1, recv/2 and recv/3

Link to this type send_error() View Source 
send_error() :: Bricks.Error.Closed.t() | Bricks.Error.Posix.t()

The errors send_data/2 may return

Link to this type send_result() View Source 
send_result() :: :ok | {:error, send_error()}

The return type of send_data/2

Link to this type set_active_return() View Source 
set_active_return() :: {:ok, t()} | {:error, Bricks.Error.Posix.t()}

The return type of set_active/2

Link to this type t() View Source 
t() :: %Bricks.Socket{
  active: active(),
  bam_window: window() | nil,
  closed_tag: atom(),
  data_tag: atom(),
  error_tag: atom(),
  handle: term(),
  host: host(),
  module: atom(),
  passive_tag: atom(),
  port: port_num() | :local,
  receive_timeout: timeout()
}

A connected socket

Link to this type window() View Source 
window() :: :once | pos_integer()

The activity mode used for extending Bounded Active Mode

Link to this section Functions

Link to this function close(socket) View Source 
close(t()) :: :ok

Closes the socket. Always returns :ok

Link to this function extend_active(socket) View Source 
extend_active(t()) :: set_active_return()

Extends the bounded active mode (BAM) of a socket by its bam_window

Link to this function extend_active(socket, window) View Source 
extend_active(t(), window()) :: set_active_return()

Extends the active of a socket by the given window

Link to this function fetch_active(socket) View Source 
fetch_active(t()) :: fetch_active_return()

Fetches the current activity of the socket

Link to this function handoff(socket, pid) View Source 
handoff(t(), pid()) :: handoff_return()

Hands a Socket off to a new process, which becomes the owner

Link to this function new(opts) View Source 
new(new_opts()) :: {:ok, t()} | {:error, new_error()}

Creates a new Socket from a map of options

Required keys:

  • module: callback module for the given socket tye
  • handle: underlying reference to the socket
  • active: current activity mode
  • data_tag: tag used to identify a data message from the socket
  • error_tag: tag used to identify an error message from the socket
  • closed_tag: tag used to identify a closed message from the socket
  • passive_tag: tag used to identify a passive message from the socket

Optional keys:

  • receive_timeout: default timeout used for calls to recv
  • bam_window: default active window to use for resetting bounded active mode
Link to this function passify(socket) View Source 
passify(t()) ::
  {:ok, binary(), Bricks.Socket.t()} | {:closed, binary()} | {:error, term()}

Turns the socket passive, clearing any active data out of the mailbox. and returning it as one binary Note: Assumes binary mode!

Link to this function recv(socket) View Source

Receives all available data from the socket. Times out if the Socket’s receive_timeout is breached.

Link to this function recv(socket, size) View Source 
recv(t(), non_neg_integer()) :: recv_result()

Receives data from the socket

If the socket is not in raw mode (default) or size is zero: Receives all available data from the socket If the socket is in raw mode and size is not zero: Receives exactly size bytes of data from the socket

Times out if the Socket’s receive_timeout is breached.

Link to this function recv(socket, size, timeout) View Source 
recv(t(), non_neg_integer(), timeout()) :: recv_result()

Receives data from the socket

  • If the socket is not in raw mode (default) or size is zero: Receives all available data from the socket
  • If the socket is in raw mode and size is not zero: Receives exactly size bytes of data from the socket

Times out if timeout is breached.

Link to this function send_data(socket, data) View Source 
send_data(t(), iodata()) :: send_result()

Sends the given iolist data down the socket.

There is no timeout for this operation unless one was specified during construction or set after construction.

Link to this function set_active(socket, active) View Source 
set_active(t(), active()) :: set_active_return()

Changes the socket’s activity mode. Valid activities:

  • true - enable active mode
  • false - enable passive mode
  • :once - enable BAM for one packet
  • integer() - adjust the internal active window counter by this many

Note that while the first three all set the value, providing an integer can behave in two ways, depending on the current mode:

  • You are not in BAM, the active counter is set to this value. If it is lower than zero, the socket is made passive.

  • You are in BAM, the active counter is adjusted by this value (by addition). If you pass a negative number and cause the counter to go below 0, the socket is made passive.

We recommend only using set_active/2 with a positive integer and when the socket is in passive mode (such as when you have been informed the socket has just been made passive). This simplifies the problem and you can pretend that it does set the internal counter.

If you ignore the last paragraph, you’ll have to keep track of the current active window to know what the new one will be after calling set_active/2. The (undocumented) decr_active/1 function may be useful to call when you receive a packet. You could also use fetch_active/1, but this will be slower.

This behaviour is inherited from OTP. We would not choose to implement it this way ourselves.

Link to this section Callbacks

Link to this callback close(t) View Source 
close(t()) :: :ok

Closes the socket

Link to this callback fetch_active(t) View Source 
fetch_active(t()) :: fetch_active_return()

Fetches the active mode of the socket

Link to this callback handoff(t, pid) View Source 
handoff(t(), pid()) :: handoff_return()

Makes another process the new owner of the Socket

Link to this callback recv(t, non_neg_integer, timeout) View Source 
recv(t(), non_neg_integer(), timeout()) :: recv_result()

Receive from the Socket

Link to this callback send_data(t, iodata) View Source 
send_data(t(), iodata()) :: send_result()

Send the data down the socket

Link to this callback set_active(t, active) View Source 
set_active(t(), active()) :: set_active_return()

Set the activity mode of the socket