Mesh - Capability-based routing for Processes on the BEAM

tbh y’all need to chill this thread is kinda unhinged

It seems to me like you know what you’re doing and wrote a library that makes sense for your case, but when describing it in the readme you (understandably) tried to put it into very general terms. But because it’s very general everyone is just going to project their own experiences onto it instead.

In a similar situation I found that explicit examples help. The sidecar thing makes sense to me; consider adding it to the readme.

I think it’s normal to be unfamiliar with all possible terms in the world of technology. I didn’t mean background job, and it certainly has nothing to do with LDAP.

A sidecar is a container that runs alongside another container within a POD. This sidecar container can provide any kind of extra functionality that the main container depends on and that, for some reason, it doesn’t execute on its own.

Here’s a random examples from the internet of what a sidecar is:

Here’s something closer to what I described as a use case: https://www.youtube.com/watch?v=b86DIo0_UoU

And here’s the project that will use Mesh after we stabilize the Mesh API:

https://github.com/eigr/spawn

Sorry, I am not familiar with the term “sidecar”. Do you mean a background job? A server-less function? A system optimize to run hour long simulation jobs will need very different architecture from a system to execute serverless functions finishing in a few milliseconds.

What you described feels like a directory service, which may not even need to be distributed in nature. LDAP has been there for decades; when properly configured it is pretty fast too.

In a sidecar-based system, clients in other programming languages ​​register with a hostname. These containers have routines that contain entities, and these entities have names. They share the same cluster, and the sidecar executes the calls between the various containers via erlang dist. For this to work, the sidecar in Elixir cannot choose a random Node in the cluster because there is an affinity between the sidecar and the host application, and therefore it would fail to call a sidecar that does not have an affinity with the destination sidecar. So we need:
1 - To know which hosts have which entities
2 - When a client invokes an entity, the cluster must call exactly the sidecar that has a host that has the destination entity.

That’s exactly what the library was created for, and it meets these requirements. This happens in practice. However, I see that it’s possible to explore the library for other use cases. For example, I could have applications separated by region and could route a call based on the client’s region. Or I could have specific workloads for specific tasks and route based on that. There are many possible use cases.

Of course I can imagine all that, I even gave you 3 examples so you can tell me which one is closer to your application, and you did not answer my question. My point is: the need for affinity could be mandatory, or advisory, or anything in between. I do not think your library can cover all bases, at least not now, right?

When I “shop” for a library, I look for 3 things, in this order:

1. The concrete problem the creator intended to solve
2. The commitment from the creator to solve it, even without any external contribution
3. The vision of the creator to expend the application field, with external contribution

You have given 3, but not 1 and 2.

Imagine you have a cluster of shared applications that aren’t homogeneous—that is, they are different applications, such as games, chat applications, and others—but for whatever reason you want them to be part of an Erlang cluster. You want to logically group these applications within the same cluster and be able to invoke processes using any label (as mentioned before, not a PID). You expect a process to be started and maybe monitored when this request occurs, and that eventually, this request will go to the same node as long as there is a live process on that node. That’s basically it. We don’t want strong consistency, replication of user process state, nothing like that; it’s eventual consistency and it’s just service discovery. You can implement your own selection strategy if you want.

Oh man, that was rather sad.
I appreciate all constructive feedback, and as I said in the post, the idea was to share it very early to receive positive feedback, without any pretense of being perfect or academic.
I think you made a number of assumptions, both about the library itself and about me, whom you don’t know. I have 25 years of experience, many of which have been spent working with distributed systems, and I have read many books and implemented many systems in production. I also contribute to many libraries, many of them about distributed systems, not only from the Elixir ecosystem but from other ecosystems as well, and I always try to be polite to everyone I interact with.
You mentioned Horde; I am one of the contributors/maintainers of the project, and I am sure I know it very well. Anyway, thank you for the feedback; I appreciate it very much and will take it into account whenever possible.

Can you articulate your intended usage scenario with a concrete example? Do you want “node affinity” based on performance concerns, ie. hot cache, or physical attributes of the nodes ie, big memory vs small memory, or security/regulation concerns ie. to make private enclaves within a public cloud?

I’ve just reread my post, and it sounds a bit toxic. I want to make clear that given that this implementation is poor, it is a good start and the whole idea is not strange, but it is a classic problem which was solved several times by other authors from Elixir community and what you’re doing with Mesh is a good starting point to explore the problem space

My personal recommendation would be to

1. Learn more about distributed systems. I would love to provide some books and resources that I personally find useful, but unfortunately very few of them are in English
2. Take a look at existing implementations, learn about their tradeoffs. For example, Horde library solves the similar problem, but it uses eventually consistent delta-CRDT merkle tree structure to (eventually) maintain the same version of the information about which processes belong to what groups. Other good example is ProcessHub which is, afaik, still in active development phase.
3. Learn more about consistent hashing algorithms and their applicability. These algorithms are very interesting because there is an exotic mix of finite field algebra and some empirical observations of how distributed systems change their memberships
4. Dont be afraid to ask questions. You can tag me in any thread or use the personal messages. I also provide mentorship and consulting services. But you can also just create topics on the forum. Discussions about distributed systems are especially welcome here

I am looking forward to seeing the new version of Mesh which would use some interesting distributed algorithm. Happy coding

I’ve read the code and here’s my review

TLDR

This is a poor library with a lot of bugs, misleading documentation and poor applicability.

What does it actually do?

It essentially does 2 things.

• You can set some “capabilities” for the node (think of it as of a group name you can assign a current node to) and you can query them

  Mesh.register_capabilities([:game, :chat])
  

• You can perform some request on the group, which will pick a node from the group, spawn some (or use existing) GenServer and do the GenServer.call with specified payload

  {:ok, pid, response} = Mesh.call(%Mesh.Request{
    module: MyApp.GameActor,
    id: "player_123",
    payload: %{action: "move"},
    capability: :game
  })
  

  Each node in the group is going to maintain a shard_count of processes per each capability called “actors” in terms of this library (but I call them just shards here) and this request is routed to one of these shards. That means if the node has two capabilities and max shard count is 10, it is going to maintain 20 processes (all of them are lazily initialized).

Problems

Setting capabilities

Capabilities is a distributed data, and as all such data it is subject to CAP. Current solution is not qualifying for any of those letters and, putting CAP aside, provides very poor guarantees. If other node can’t start the Mesh and set the capabilities in time, the current node won’t know about it. It uses replication on net_kernel.monitor_nodes events, which just tries to do asynchorous rpc to get the capabilities of other node and share current capabilities with it.

That means that any change to cluster configuration or groups is eventually consistent (at best) and it is quite possible that your request would get routed to the already dead node. If so, this library provides no fallback and you’d just have a dead request

Picking a node

Alright, let’s forget about previous paragraph and imagine that every node has the exactly the same information about the groups (aka “capabilities”).

In pseudo-code it looks like this

shard_id = hash(request.id, shard_count)
nodes_in_group = Enum.sort(nodes_for_group(request.node))
node_to_execute_in = Enum.at(nodes_in_group, rem(shard_id, length(nodes_in_group)))
:rpc.call(node_to_execute_in, fn ->
  shard_owner = via(shard_id)  # Yes, shard_id, the same as on the first line
  GenServer.call(shard_owner, {:call, request})
end)

# Where the GenServer.call executes this

pid = start_or_get_shard_process(process, shard_id) # Again, shard_id
response = GenServer.call(pid, request.payload)
{:reply, response, ...}

And it has many problems. First thing is that routing to the node is not consistent. Second is that shard within node is selected twice, which is a bug.

Then, and it is the most beautiful bug here, shard_id is used as a key for selecting a node in a group and the shard process. That means, that if I have two nodes and 20 shards on each node, only 10 shards on each would receive the request, which is very fun. In general, if I have N nodes, only shards / N amount of shard will receive the request.
Because if I hit the first node on the list of two nodes, that maens that shard_id is dividable by two, and all shards with shard_id not dividable by two won’t receive a request on this node ever.

And for problems which I couldn’t reflect in the pseudo code:

• Shard processes are created on demand and routed by shard_id. That means if I send a request to a not present shard with Module1 as the implementation, it would start a shard as GenServer.start_link(Module1, ...), and if I then send a request which would end up in the same shard, but with Module2 passed as the implementation, it would still hit the shard created by the first request, with Module1
• If one shard performs a request, which would eventually get routed to the local node, there would be a deadlock. There is no check to ensure that the call ends up in the different shard

Other problems

• It creates an ActorTable process on each node which just does nothing.
• Misuse of PartitionSupervisor
• Mesh.Cluster.Membership schedules monitoring after 100ms, while it can just do it in handle_continue
• This process also syncs shards on every nodeup and nodedown, but it only changes the local shards, so this whole process is unnecessary
• If an “actor” dies, it would get restarted by the supervisor, but the ActorTable ets table is going to have the pid of the dead “actor”, resulting in every request hitting the dead process
• It uses global for locking on only the current node

Conclusion

Unclear use-case, bad implementation, misleading documentation.

Same functionality (but without a lot of bugs) can be achieved by using any other process group solution (including bultin global_group, pg and third-party Horde, Swarm, ProcessHub?, syn).

But given that shard processes are expected to be stateless and rounting to be inconsistent, it all boils down to just two lines

node = Enum.random(Node.list())
:erpc.call(node, fn -> ... end)