A general observation: if you’re ignoring the GenServer’s state in every handler, you probably don’t need a GenServer. For instance, save_with_call could be simplified to:
def save_with_call(user) do
Phoenix.PubSub.broadcast_from(MyApp.PubSub, self(), @topic, {:perform_new, user})
end
As a bonus, this approach doesn’t have a single process (the GenServer) forcing everything to happen one-at-a-time.
OK, good point, in fact, I really wondered, indeed, but how do I respond to :perform_new if not from a Genserver? As you say, I only use the messaging part.
You can have a GenServer (or any interested process) subscribe to and handle the messages, or a pool of them with a pool manager dispatching the messages, but you don’t need to send the messages from the GenServer if that’s all that save_with_* is doing. You can pubsub those messages from any process and your MyApp.Repo servers will subscribe to that topic. Then you’ll avoid routing new messages through the same process that wants to receive them, which significantly reduces the work load on each one and increases their throughput receiving the messages
Firstly apologize, just a beginner with Elixir. Then yes I believe I started with this, not using GenServer.call. This relieves a bit of pressure but I may still need the GenServer behaviour. Nodes subscribe to this pubusub topic. Nodes also listen to :net_kernel.monitor_nodes events, and broadcast on a :nodeup event. I need to capture the message the nodes broadcasts and do something, so I don’t know how to do this is not using a handle_info matcher offered by the server
No need to apologize! OTP takes time to learn but it’s a rewarding process.
Yes exactly, you can have a process or pool of processes on each node subscribing to the pubsub messages, even though they can be broadcasted by any caller in the cluster.
You will also need a long running process to listen for these :nodeup events and rebroadcast them, and any interested process can subscribe to the topic. You should probably have a NodeListener with the sole responsibility to handle and rebroadcast these events. All of the messages we’ve discussed so far can be handled by the handle_info/2 callback of the GenServer behaviour.
You only need to call a GenServer if it holds some state (data, connection, etc.) or performs some computation and you need the results returned. It’s always synchronous. If the calling process can do the work itself then it probably should unless you need to offload it for async behavior or error isolation, in which case you might consider spawning a Task instead of sending it to a single GenServer.
A basic implementation of what I think you’re looking for
defmodule MyApp.NodeListener do
use GenServer
require Logger
def start_link(arg), do: GenServer.start_link(__MODULE__, arg, name: __MODULE__)
def init(_arg) do
:net_kernel.monitor_nodes(true)
{:ok, []}
end
def handle_info({:nodeup, _} = payload, state) do
Phoenix.PubSub.broadcast!(MyApp.PubSub, "cluster", payload)
{:noreply, state}
end
def handle_info(_, state), do: {:noreply, state}
end
defmodule MyApp.Repo do
use GenServer
require Logger
def start_link(arg), do: GenServer.start_link(__MODULE__, arg, name: __MODULE__)
def init(_arg) do
Phoenix.PubSub.subscribe(MyApp.PubSub, "cluster")
Phoenix.PubSub.subscribe(MyApp.PubSub, "users")
{:ok, []}
end
def handle_info({:new, user}, state) do
Logger.info("Repo (#{inspect(node())}) received user: #{inspect(user)}")
{:noreply, state}
end
def handle_info({:nodeup, _} = msg, state) do
Logger.info("Repo (#{inspect(node())}) received #{inspect(msg)}")
{:noreply, state}
end
end
defmodule MyApp.Users do
def save(user) do
Phoenix.PubSub.broadcast!(MyApp.PubSub, "users", {:new, user})
end
end
defmodule MyApp.Application do
use Application
def start(_type, _args) do
children = [
{Phoenix.PubSub, name: MyApp.PubSub},
MyApp.NodeListener,
MyApp.Repo
]
opts = [strategy: :one_for_one, name: MyApp.Supervisor]
Supervisor.start_link(children, opts)
end
end
$ iex --name a@127.0.0.1 -S mix
# in another terminal
$ iex --name b@127.0.0.1 -S mix
iex(a@127.0.0.1)1> Node.connect(:"b@127.0.0.1")
true
19:43:49.050 [info] Repo (:"a@127.0.0.1") received {:nodeup, :"b@127.0.0.1"}
19:43:49.050 [info] Repo (:"b@127.0.0.1") received {:nodeup, :"a@127.0.0.1"}
iex(b@127.0.0.1)1> MyApp.Users.save(1)
:ok
19:44:10.905 [info] Repo (:"b@127.0.0.1") received user: 1
19:44:10.905 [info] Repo (:"a@127.0.0.1") received user: 1
Thanks, I refactored with your comments and challenged myself. Quite a few takeaways there. First about backpressure. Not sure to understand this term? Is this related to filling up the server’s mailbox? So you mentioned a pool of servers to push it a bit further. I will probably keep this for later as I imagine you may need libraries like Poolboy for this.
Then besides code readability or “testability”, does segregation bring better performance in terms of robustness on load? For example:
Finally, as a side comment, I noticed the catch-all clause you introduced: handle_info(_, state), do: {:noreply, state}.
Backpressure describes the act of not accepting more work before having completed previous/existing work. This allows you to prevent parts of your system from being overloaded, because at best you cannot sent them more work than they can handle. This will push work to queue up in known places where you can then deal with it.
I think you would also be able to implement something simple with the PartitionSupervisor coming in 1.14. You would have to do some hashing to make sure the messages are not duped on multiple server partitions.
To me, node up and new user are separate topics, so by splitting them you can have other processes subscribe to one or the other in the future without coupling them together
I put the NodeListener as its own process precisely because it has one responsibility. It reduces the load on Repo because Repo doesn’t care about nodedown (at least that’s what I took away from what you said). And if other processes may be interested in cluster membership, the NodeListener can be the hub for those events for the whole node, without needing multiple processes to be subscribing to net_kernel.monitor_nodes.
If there is an error that you expect might happen and you have a way to recover from it, then I think you handle it explicitly with pattern matching. If there is a real exception with no recovery then “let it crash”, and the supervisor will restart your process at a known good state.
I added this because you said
So I assumed you were only interested in :nodeup, not :nodedown. If you monitor nodes you will receive both, so I discarded the :nodedown messages. Maybe that was a wrong assumption 