Code spells, smells and sourcery

Phoenix WebSockets Under a Microscope 🔬

This is a code-reading and exploration post about the WebSockets side of Phoenix.
It builds upon some of the tracing techniques showcased in the previous post, to observe
some of the internals of Phoenix.
It also features some tricks I commonly employ to debug WebSocket
related issues. The title and nature of this post are inspired by the
marvellous book Ruby Under a Microscope written by Pat Shaughenessy.

WebSockets

The WebSocket Protocol enables two-way communication between a client
running untrusted code in a controlled environment to a remote host
that has opted-in to communications from that code

The goal of
this technology is to provide a mechanism for browser-based
applications that need two-way communication with servers that does
not rely on opening multiple HTTP connections (e.g., using
XMLHttpRequest or <iframe>s and long polling)

Conceptually, WebSocket is really just a layer on top of TCP that does the following:

Adds a web origin-based security model for browsers

Adds an addressing and protocol naming mechanism to support
multiple services on one port and multiple host names on one IP
address

Layers a framing mechanism on top of TCP to get back to the IP
packet mechanism that TCP is built on, but without length limits

Includes an additional closing handshake in-band that is designed
to work in the presence of proxies and other intermediaries

With the --no-brunch option, we skip generating Brunch
related scaffolding, as none of the examples below require JavaScript and
--no-ecto to skip database related configuration and modules.

Phoenix@v1.2.5 is used for all the examples of this post.

By design, a Phoenix channel is an abstraction on sending and receiving
messages about topics. Phoenix.Transports.WebSocket is the default
transport mechanism, there are others available and you can even
write your own. You may read the documentation about channels
here.

Let’s code our first channel by creating a web/channels/integers_channels.ex.

Two topics are defined, numbers:positive streaming positive numbers
and numbers:negative for negative ones.

You may notice that we’re assigning a :joined_at attribute to each
client. This is just a convention which can come handy to calculate for how
long a client has been connected. For real-time join/leave tracking,
Phoenix.Presence should be preferred.

We can now start the application using:

iex -S mix phoenix.server

Connecting to WebSockets from the terminal

Now that we have the server up and running, we can initiate some
connections using wsta, a cli tool written in Rust ⚙️, which
follows the Unix philosophy letting you pipe streams to and from to
other scripts or files.

You’ll notice that wsta will print Disconnected! and stop streaming numbers after around 1 minute.
This is due to a default Phoenix.Transports.WebSocket:timeout configuration which specifies:

The timeout for keeping websocket connections open after it last received data, defaults to 60_000ms

To make it easier to interact with WebSockets in development you may set
it to :infinity.

Differences between push/3, broadcast_from/3 and broadcast/3

As you may read in the Phoenix.Channel documentation
there are 3 functions push/3, broadcast_from3/ and broadcast/3
which can be used to send data to connected clients, but have some
not so subtle differences.

Phoenix.PubSub

In order to understand how essential it is and how things are tied to
together, we’ll trace the framework’s data flows.

In our example, a client connects to the channel "numbers:42", then from
the server a message {"number": 42} is broadcasted to all connected
clients.

Take a deep breath and let’s probe 🔎 Phoenix to find out what goes on
under the hood, for our message to reach its recipients.

Subscribing

When a client joins a channel,
Phoenix.Socket.Transport.connect/6 is
called. It builds a Phoenix.Socket with pubsub_server set to
the configured name in the :pubsub setting of the Endpoint. For
this sample application it defaults to Numbers.PubSub.

It will call Numbers.UserSocket.connect/2 with the following arguments:

connect(%{},%Phoenix.Socket{assigns:%{},channel:nil,channel_pid:nil,endpoint:Numbers.Endpoint,handler:Numbers.UserSocket,id:nil,joined:false,pubsub_server:Numbers.PubSub,ref:nil,serializer:Phoenix.Transports.WebSocketSerializer,topic:nil,transportPhoenix.Transports.WebSocket,transport_name:websocket,transport_pid:<0.421.0># This is a :cowboy_websocket process})

Then a Phoenix.Channel.Server process is started by Phoenix.Channel.Server.join/2.

defmodulePhoenix.PubSubdodefsubscribe(server,topic,opts)whenis_atom(server)andis_binary(topic)andis_list(opts)docall(server,:subscribe,[self(),topic,opts])end# 👇 For the subscription call/3 will be called with:# server here is Numbers.PubSub# kind is :subscribedefpcall(server,kind,args)do[{^kind,module,head}]=:ets.lookup(server,kind)# :ets.lookup(Numbers.PubSub, :subscribe)apply(module,kind,head++args)endend

The Numbers.PubSub named ETS table will have the following entry for :subscribe:

{:subscribe,Phoenix.PubSub.Local,[Numbers.PubSub,1]}

So apply(module, kind, head ++ args) is in this case is a
Phoenix.PubSub.Local.subscribe/5 call.

We ended up with 2 new entries in the Numbers.PubSub.GC0 and
Numbers.PubSub.Local0 tables. Hopefully it will become clear further
down this post how they’re used.

We’re done with the subscription part 😅, moving on to the broadcast..

Publishing

When we broadcast/3 from one of our channels, as in the following example:

defmoduleNumbers.IntegersChanneldouseNumbers.Web,:channeldefjoin("numbers:"<>type,_params,socket)dosendself(),{:update,type}{:ok,socket}endenddefhandle_info({:update,"42"},socket)do# 👉 We broadcast to all the connected clientsbroadcastsocket,"update",%{number:42}{:noreply,socket}endend

defmodulePhoenix.Channel.Serverdo# 👇 Will be called with arguments:# broadcast(Numbers.PubSub, "numbers:42", "update", %{number: 42})defbroadcast(pubsub_server,topic,event,payload)whenis_binary(topic)andis_binary(event)andis_map(payload)doPubSub.broadcastpubsub_server,topic,%Broadcast{topic:topic,event:event,payload:payload}endend

Next the message is serialized and sent to a cowboy_websocket handler process.
File: lib/phoenix/transports/websocket_serializer.ex (link)

defmodulePhoenix.Transports.WebSocketSerializerdo@moduledocfalse@behaviourPhoenix.Transports.SerializeraliasPhoenix.Socket.MessagealiasPhoenix.Socket.Broadcast@doc"""
Translates a `Phoenix.Socket.Broadcast` into a `Phoenix.Socket.Message`.
"""deffastlane!(%Broadcast{}=msg)do{:socket_push,:text,Poison.encode_to_iodata!(%Message{topic:msg.topic,event:msg.event,payload:msg.payload})}endend

The message is handled by Phoenix.Endpoint.CowboyWebSocket.websocket_info/3.
File: lib/phoenix/endpoint/cowboy_web_socket.ex (link)

At this point it may still not be evident how the return value of
Phoenix.Endpoint.CowboyWebSocket.websocket_info/3 manages to send data
down the socket. For this to be demystified, proceed to the next
section.

Phoenix and Cowboy

When the numbers application is started, its Numbers.Endpoint
supervisor is started, supervising the following children:

ranch_tcp is a wrapper around gen_tcp and gen_tcp.listen/2 sets up a socket
to listen on a random port on the local host.

ranch_acceptors_sup then starts a pool of acceptor supervised processes.

Each one of them waits (with :infinity timeout) to accept a connection
on the listening socket.

If a connection is established, the ranch_acceptor will transfer the
control of the socket to the connections supervisor so that it receives
messages from the socket.

When a TCP connection is established ranch_conns_sup will start a cowboy_protocol process using :cowboy_protocol.start_link/4. When the cowboy_protocol process is started successfully, ranch_conns_sup will transfer the control
of the socket to that process so that it receives messages from the socket.

The cowboy_protocol handler is responsible for receiving and parsing messages of the HTTP protocol. It handles requests
by executing all the layers of its middleware stack. By default this
stack contains cowboy_router.

which is generated from (link) and concludes the WebSocket handshake, so that the data transfer starts.
It is now a two-way communication channel where each side can, independently from the other, send data at will.

After a successful handshake, clients and servers transfer data back
and forth in conceptual units referred to in this specification as
“messages”. On the wire, a message is composed of one or more frames

With an established WebSocket connection, a client (phoenix.js, see link) will push a:

{"topic":"numbers:42","event":"phx_join","payload":{},"ref":"1"}

The cowboy_websocket processs, which has the control of the socket,
receives a message in its mailbox, decodes it (WebSocket Protocol) and
calls Phoenix.Endpoint.CowboyWebSocket.websocket_handle/3 as follows:

Then Phoenix.Transports.WebSocket.ws_handle/3 is called which parses
the message as JSON and calls Phoenix.Socket.Transport.dispatch/3
which will pattern-match on the event part of the message so that
phx_join will finally call Phoenix.Channel.Server.join/2.
What happens next has already been described in the subscribing section.

Debugging WebSockets Essentials

Armed with some knowledge about the inner-workings of Phoenix, we can
have some common questions answered. Use the following code snippets
with caution in production environments, as some of the Phoenix
functions used are documented as:

# With the assumptions that all the connected nodes are members of the PubSub cluster:pool_size=Application.get_env(:numbers,Numbers.Endpoint)[:pubsub][:pool_size]||1fun=&Phoenix.PubSub.Local.subscribers(Numbers.PubSub,"numbers:42",&1):rpc.multicall[node()|Node.list],Enum,:flat_map,[0..(pool_size-1),fun]#=> {[[#PID<0.19570.2363>, #PID<0.4457.2620>, #PID<0.18260.2750>, #PID<0.6396.2762>,#=> #PID<0.7481.2801>, #PID<0.2731.2866>, #PID<0.13608.2904>,#=> #PID<0.20333.2910>],#=> [#PID<37723.31494.2599>, #PID<37723.20571.2660>, #PID<37723.26132.2671>,#=> #PID<37723.30777.2690>, #PID<37723.23804.2712>, #PID<37723.27526.2731>,#=> #PID<37723.7742.2776>]], []}# If you don't like assumptions you can use this to get the actual nodes (with Phoenix.PubSub.PG2 adapter):nodes=:pg2.get_members({:phx,Numbers.PubSub})|>Enum.map(&node/1)[:"numbers1@autoverse",:"numbers2@autoverse"]

Conclusion

The Phoenix codebase is a very interesting one and there’s lot to learn
from it. I really hope that sharing my quest to understand and explore
Phoenix will be beneficial for others. If you found a mistake please
submit a pull request or mention it in the comments.