The GenServer from hell (needs some refactoring)

You may try the erlang :gen_statem. it is a fancier :gen_server with some bells and whistles that you might need. (timeout, postpone, self inserted event…)

There’s an Elixir wrapper for the :gen_statem but I haven’t used that myself:

While :gen_statem may make the SM implementation more concise, I don’t think it helps with the general problem, that I have a GenServer that has too many responsibilities, is hard to understand and hard to test.

I suspect gen_statem would help in your case mainly because of things like internal events and event postponement.

In particular, using internal events would allow you to transform code like the send_frames in handle_dispatch into a pure function that returns something like:

{:next_state, :something, data, [{:next_event, :internal, {:send_frame, :up, up_frame}}, {:next_event, :internal, {:send_frame, :dn, dn_frame}]}

The state functions are pure functions that can return additional “actions” in the last position.

Then the handlers look like (assuming the state machine is in :state_functions mode, and the current state is something):

def something(:internal, {:send_frame, dir, frame}, data) do
...
end

Actions can also include :postpone (which holds the event until the next state change) and three different flavors of timeout

I have an opinionated wrapper of :gen_statem that is designed to help you organize your state machine, and also marshalls the confusing gen_statem callbacks into something that looks more like a gen_server.

Of course, in the wrong hands it could merely wind up being more rope to hang oneself with, but if you are a person that likes organizational clarity, it could be useful.

I just had a quick look at :gen_statem again. Turns out I did not understand what it does when I first looked - and I still don’t, but I’ll have a closer look. Thanks for the pointer, I’ll be back.

If I understand correctly, this protocol resembles the OSI model, where multiple layers are stacked on top of each other with the following properties:

1. Each layer receives input packets and produces decoded packets for the consumer layer (the one directly on top of it).
2. Some layers might need to perform additional imperative logic, such as send a response to the peer layer on the other side, or defer an action for later.

Stacked behaviours

One way of implementing this would be to rely on stacked behaviours. The most foundational one, called Layer0 would receive raw bits (e.g. using gen_tcp) and invoke layer-specific callbacks, such as handle_packet which receives a decoded layer 0 packet. On top of this abstraction you can build Layer1, on top of that Layer2, and so on.

This approach allows you to test at lower levels. I’d still test as much as I can at the final level (Application), moving deeper to test the scenarios which are harder to setup at the higher level, such as lower-level timeout.

One downside of is that the implementation will always be coupled to the physical transport. If layer 0 uses tcp to communicate, you’ll always need a fake tcp peer to test the logic, which could become cumbersome to implement. Another issue is that implementing 6 behaviours could lead to a lot of boilerplate. For example, to accommodate deferred actions you’ll probably need to support the handle_info callback in the corresponding layer and all the layers below it.

Functional core, imperative shell

Another option is to go for functional core, imperative shell. You can see a detailed example in my post To spawn, or not to spawn?.

Basically in this approach we distinguish between the driver and the functional logic. The driver takes care of receiving bits/bytes from the remote client and shipping the responses back. It’s also responsible for time-based logic. So that’s the imperative shell. The functional core is a pure-functional state machine which receives inputs from the driver and produces outputs. At the highest level it could look something like:

defmodule KNX do
  @spec new(...) :: t

  @spec handle_impulse(t, impulse) :: [action]
end

where types have the following meaning:

• t - SM state represented as a map or a struct
• impulse - external actions such as {:bytes, binary}, {:message, payload}
• action - an imperative action that must be executed by the driver, such as {:send_bytes, binary}, {:delayed_impulse, time::pos_integer, payload}

With such contract, the driver would be a stateful process that establishes the connection to the remote client, accept bytes, sends impulses to the functional core (which is kept as the state of the process), and interprets the actions. An immediate benefit here is that you can test the core without needing to fake a remote TCP peer, and without needing to depend on time. For example, if the core sends {:delayed_impulse, ...} (a delayed self-send), the test can immediately pass the impulse back.

Inside the core module, I’d start by implementing the zero layer as two private functions:

@spec new_layer_0(...) :: layer_0_state

@spec handle_layer_0_impulse(layer_0_state, layer_0_impulse) :: [layer_0_actions]

This follows the same pattern as the high-level KNX API, and these functions would initially reside in the same module (I’d move them to a separate “sub-module” if the code grows). The impulses to layer 0 are e.g. raw bits/bytes, and the output actions are decoded packets.

Now on top of this we can build layer 1. When layer 1 receives an impulse, it needs to make a decision. If the impulse specific to layer-1 (e.g. a delayed self-send), the layer handles it and produces output actions. Otherwise, the lower-level layer is invoked to handle the impulse.

In the latter case, layer 0 will return a set of actions. Some of these actions will be decoded packets. Layer 1 has to interpret each decoded packet and produce its own additional actions. So for example, let’s say that layer 0 returns:

[
  layer_0_action_1,
  layer_0_action_2,
  layer_0_decoded_packet_1,
  layer_0_action_3,
  ...
]

Layer 1 could produce:

[
  layer_0_action_1,
  layer_0_action_2,
  layer_1_action_1,
  layer_1_decoded_packet_1,
  layer_0_action_3,
  ...
]

So by stacking all these layers you end up with the top-level handle_impulse that will return actions required by different layers in the proper order.

This approach also supports testing at the lower levels. You can expose each individual layer API, e.g. as public functions marked with @doc false, or by moving each layer into its own submodule. Again, I’d test as much as I can at the highest possible level.

FWIW I used this style to implement the server-side of the PostgreSQL protocol (only 1 layer, but still), and it has its pros and cons. As said, the biggest pro is that you decouple the protocol from the imperative logic, which simplifies testing. On the flip side, any combination of imperative + conditional logic will be trickier to implement. E.g. if you need to send something to the peer and than interpret the peer’s response, you need to produce an output action, store some piece of information in your state, and then reconcile that with a response impulse. This is much trickier than plain old imperative interpret_response(send(peer, ...)), where send is a blocking operation that awaits for the peer’s response.

But I think that you’ll face the same challenge in the first proposed approach (behaviour-based). In general, a straightforward imperative style is blocking, which means that you lose the ability to perform asynchronous actions (e.g. ship something else to the peer while you’re awaiting for its response).

I just had a lot of fun building a toy-protocol based on your

Functional core, imperative shell

proposal. Thank you!

I took the state-machine (the functional part) and the actions from the KNX-stack and removed sequence-numbers and repeat-count. (those do not effect the mechanics, just bring some complexity).

The protocol is very simple, a frame looks like this:

[source-address, destination-address, service, data]

for example

0xDEAD_BEEF

The impulse form the driver for the lowest layer (data-link, :dl) looks like this:

{:dl, :ind, <<0xDEAD_BEEF::32>>} # data-link, indication, frame

def handle({:dl, :ind, <<src::8, dest::8, data::bits>>}, %State{}),
  do: {:nl, :ind, %Frame{src: src, dest: dest, data: data}}

:dl matches on the addresses and builds a %Frame{}. network-layer (:nl) is next:

def handle({:nl, :ind, %Frame{dest: address} = frame}, %State{addr: address}),
  do: {:tl, :ind, frame}

If the destination address matches the device address (addr in %State{}), it is transport-layer’s (:tl) turn.
:tl has to extract the service. (0xBE is the data-connected service, others are ack, connect, connectionless, …)

def handle({:tl, :ind, %Frame{data: data} = frame}, %State{}) do
  {service, data} = Service.decode(data)

  case service do
    :invalid -> {:logger, :error, {:invalid_service, frame}}
    _ -> {:tlsm, :ind, %Frame{frame | service: service, data: data}}
  end
end

Next is the transport-layer-state-machine (:tlsm)

def handle({:tlsm, primitive, %Frame{service: service} = frame}, %State{} = state)
    when service in @connected_services do
  {new_state, action} = state.handler.(state, primitive, frame)
  action.(new_state, frame)
end

The SM has to be in open_idle-state and the source-address has to match the connected-address to let the data-connected service pass:

def open_idle(
     %State{c_addr: c_addr} = state, primitive, 
     %Frame{service: service, src: src}) do

  {next_handler, action} =
    case {primitive, service, src} do
      ...
      {:ind, :t_data_con, ^c_addr} -> {&Sm.open_idle/3, &Action.a02/2}
      ...
    end

  {%{state | handler: next_handler}, action}
end

the SM executes action a02

def a02(%State{c_addr: c_addr} = state, %Frame{} = frame) do
  {
    state,
    [
      {:timer, :reset, :connection},
      {:al, :ind, frame},
      {:tl, :req, %Frame{service: :t_ack, dest: c_addr}}
    ]
  }
end

This action does not change the state (the connect-action for example would set the connected-address).
But it adds three new impulses:

• a command for the timer to reset the connection-timeout
• the indication is forwarded to the application-layer (:al)
• an :ack is scheduled to be sent back

:al is boring (the real thing actually handles about 100 services)

def handle({:al, :ind, %Frame{data: data}}, %State{}),
    do: {:user, :ind, data}

The engine looks like this.
There is a reducer that takes a state, a list of impulses and a handler

def reducer({state, impulses}, handler) do
  {state, impulses} =
    Enum.reduce(impulses, {state, []}, fn a, {state, impulses} ->
      case handler.(a, state) do
        {%State{} = state, impulse} -> {state, [impulse | impulses]}
        impulse -> {state, [impulse | impulses]}
      end
    end)

  {state, impulses |> Enum.reverse() |> List.flatten()}
end

also there is an up() and a down() function that calls each layer with the (reduced) output of the adjacent layer:

def up(state, impulses) do
  {state, impulses}
  |> reducer(&Dl.handle/2)
  |> reducer(&Nl.handle/2)
  |> reducer(&Tl.handle/2)
  |> reducer(&Tlsm.handle/2)
  |> reducer(&Al.handle/2)
  |> IO.inspect()
end

This call for eample:

up(
  %State{c_addr: 0xDE, addr: 0xAD, handler: &Sm.open_idle/3},
  [{:dl, :ind, <<0xDEAD_BEEF::32>>}]
)

yields these impulses:

[
   {:timer, :reset, :connection},
   {:user, :ind, <<0xEF>>},
   {:tl, :req, %Knx.Frame{data: "", dest: 222, service: :t_ack, src: nil}}
]

so now the timer has to do some work, the user has to be notified and the :tl, :req goes into down().

I really like it. The core now is pure and so super-easy to test and understand. Even the annoying things like deferred events are super simple (just keep the impulse).

If someone is interested I can push this to github, but this is most likely very boring stuff for normal people.

I also have to look into :gen_statem but I’ve had a hard time understanding it, and the

Stacked behaviours

You need to always remember that when doing any type of server in Elixir, the server itself shouldn’t have any logic, instead should dispatch it to modules, so that you can rip-off this benefits that you now are aware off after following @sasajuric suggestion.

If you have not read it yet I would recommend you to read his book Elixir in Action:

The book is an eye opener for properly use server processes and message passing in Elixir

I wouldn’t go that far If the server is fairly small, I prefer to implement everything in the same (server) module, postponing the split until the code becomes “too large” and some clear abstractions emerge.