Erlang clustering for servers

@flodihn: Are you using the available tcp_server for client connections, or something else, or rolling your own?
Are you using Erlang process messaging or OTP messaging for clustering clients? If so, in what logical configuration?

Quote:

Actual response by flodihn Quote: Original post by hplus0603 @flodihn: Are you using the available tcp_server for client connections, or something else, or rolling your own? Are you using Erlang process messaging or OTP messaging for clustering clients? If so, in what logical configuration? I use the OTP tcp_server for my connections servers (the frontline servers in my cluster). I do not understand your question about Erlang processes messaging or OTP messaging. I am using the OTP framework to structure my code into OTP applications, if that is what you asked. Also I am developing an MMO engine, not a multiplayer engine so I cluster servers, not clients.

I meant clustering servers, not clients -- that was a typo.

My question for OTP or Erlang messaging is, when you receive a message from a client, and it needs to be forwarded, do you use an OTP application name on a destination node (with gen_server:cast or similar), or do you use a pid, or something else?

I'm working on a clustered OTP server for a messaging fabric at work, and we end up using raw pids for intra-cluster messaging, even though we use OTP for the supervision tree. I want to compare to other applications, to see whether there is something we're missing.

Quote:

Actual post by flodihn I build my servers on a group of gen_servers (in an OTP supervision tree). This means most of my messaging is done with a call to a interface function in a gen_server. However is some places still the "ordinary" ! character for sending messages. For example one object sends an event to nearby object, the object calls my nextgen.libevent.srv:event function which is a gen server. The libevent server iterates a mnesia table, extracts the pids and sends the message by calling obj:async_call(Pid, Msg) for example. In turn obj:async_call just wraps the messaging. When a message arrives from the client, I use for example rpc:call(node(Pid), obj, async_call, [Args]) to forward it to the correct area server. To summarize, most of messaging is done through calling interface functions in OTP servers. Those cases where I do normal Erlang messasing, I wrap it in a function call. I feel we risk of hijacking the OP post. If you want to get into detail I suggest we create a post of our own.

It's the second case I'm interested in. Our fabric uses three kinds of nodes:
- client-facing nodes
- messaging crossbar nodes
- supervisor node

The client-facing nodes is what clients connect to. These are interchangeable. They accept incoming TCP connections using a Google Protocol Buffers based protocol, calls out for authentication, and then forwards requests to the appropriate crossbar nodes.

Crossbar nodes are mapped based on a hash on the messaging realm in question. For example, a chat channel may be named "/chat/public/some-chat-name" and would hash to one of the N crossbar nodes. Each channel is a process, and each client is a process.

When the client node decodes a message that needs to go to a chat channel (in this case), the job of getting to the right process doesn't really have a good solution in OTP that I can find. We ended up building a two-stage dispatch, where the client node finds the right host to forward to using the hash (and a dynamic hash->node map), and a target dispatcher process on the messaging node then dispatches to the right message queue. Yes, this means there's a central bottleneck on each messaging node, so it's not perfect (although because there is no state, we could add N dispatchers per node if needed, and round-robin on the sending side).

Hmmm, I agree that there is something here that is not right.

If "/chat/public/some-chat-name" is a chat which maps to one chat process, why don't store you store that process id instead of the node. When the client process on the client node retreives the channel process id it can send the message directly or through the rpc:call.

Quote:

Original post by hplus0603 When the client node decodes a message that needs to go to a chat channel (in this case), the job of getting to the right process doesn't really have a good solution in OTP that I can find.

What about global:register_name and global:whereis? Too many names?

Quote:

why don't store you store that process id instead of the node

Because I design for 10,000,000 names. Sticking them all in a map is not reasonable. The memory I/O cost alone of looking up the queue for each message for each client would be significant. And imagine the pain when each node needs to receive constant updates to that map -- names are added and removed all the time! (at a minimum, when each user logs in or out)

Instead, I hash all the names, and map (hash mod N) to a set of buckets, and map each bucket to a node. This is similar to "the circle hash" (or memcached fame) but IMO simpler and more straightforward, while performing at least as well. The node, in turn, then can keep a map of its local names (which is a more reasonable number -- this splays over a large-ish number of nodes in the end).

In the extension, this means that I can re-balance load by re-assigning buckets in the map, assuming I can convince a node to transfer all the state for that particular bucket to another node. I can also split buckets, by simply doubling the N in the modulo. This is largely a real-time low-latency message crossbar/infrastructure, though, not a simulation server like for RPG rules. The "everyone shares the state of the zone" problem of game simulation makes that kind of load sharing harder, because when you want 500 people in the same city, with collision checks, that suddenly takes a full core all by itself...

I am not sure I fully understand the scope of your problem, rolling your own map structure does not sound very OTP'ish in my ears but if it works good who cares?

You got me somewhat confused here:

Quote:

I design for 10,000,000 names. Sticking them all in a map is not reasonable. Quote: Instead, I hash all the names, and map (hash mod N) to a set of buckets, and map each bucket to a node

Doen't this means you actually store all the names in one map?

Quote:

Because I design for 10,000,000 names. Sticking them all in a map is not reasonable

Are you sure about this, I just added 10 000 000 rows in a mnesia table, it took about 1.2 GB of memory, it takes about 20 seconds to iterate and read all the elements, taking a random row to read takes about 0.012 milliseconds (12 microseconds).
To simulate your situation better you probably want a about 1000 processes that writes and deletes from the table all time while you have 1000 other processes reading the values.

If I would design a chat system with support to 10 000 000 channels I would to something like this:
Client nodes: Frontend the clients connects to, each client is a process.

Channel proxy nodes: For each letter in the alphabet, keep a memory resident table of channels and their pids. (If you many writes slows down lookups in popular table, for example 'a', you can see if dirty_reads/dirty_writes helps or you can split the table into 'ab', 'ac'...

Channel nodes: Keeps all the channel processes, each channel has a list or mnesia table of clients.

When a client receives data to send a message to a channel, it checks the channel pid from the channel proxy node (this value could be cached to lower the load on the channel proxy node), it sends the message directly with ! or perhaps a rpc call like rpc:call(node(ChannelPid), Mod, Fun, Arg).

Back to your original question:

Quote:

My question for OTP or Erlang messaging is, when you receive a message from a client, and it needs to be forwarded, do you use an OTP application name on a destination node (with gen_server:cast or similar), or do you use a pid, or something else? I'm working on a clustered OTP server for a messaging fabric at work, and we end up using raw pids for intra-cluster messaging, even though we use OTP for the supervision tree. I want to compare to other applications, to see whether there is something we're missing.

Most applications probably use both. I read Armstrongs book "Programming Erlang: Software for a Concurrent World", it says nothing that the "low-level" way of sending messages with ! should be avoided.

Quote:

Are you sure about this, I just added 10 000 000 rows in a mnesia table, it took about 1.2 GB of memory, it takes about 20 seconds to iterate and read all the elements, taking a random row to read takes about 0.012 milliseconds (12 microseconds).

Consider a system serving 1,000,000 connected users, spread across, say, 10 gateway nodes. The system will have 10,000,000 daily unique users, and thus an average of 116 adds/drops per second, so peak rates might be 250 per second. Each casuses 10 queues on average (this is not causation, but just for estimation), so 2500 queues are added and deleted per second.

Each gateway node needs to be able to route to all of the possible queues.

Meanwhile, the 1,000,000 simultaneous online users will send one message a second, and receive 10 messages a second (this is conservative, could be more). This means 11,000,000 messages to route per second, or 1,100,000 messages per second per gateway node. 1,000,000 of those are outgoing (from queues, to users), and 100,000 are incoming (from user, to queue).

I want the maximum latency for any message, from arrival at gateway, to distribution to all listeners from their gateways, to be < 100 milliseconds.
Consider the available memory throughput per CPU core is 3 GB/sec (24 GB/sec divided by 8 cores), meaning 300 MB/sec of available memory bandwidth per 100 millisecond time period.

Now, if the name of every queue went into every nodes' lookup dict, it seems as if the memory I/O cost of manipulating the queue and looking up queues would add undue load on the nodes. Not to mention if a new node is added, it would have to receive the entire queue map up front.

My solution is this:

node_handling_queue = bucket_map[hash(queue_name) mod bucket_count]

In this case, say I have 200 buckets -- the amount of memory needed is tiny, meaning it won't put memory pressure on the system. The drawback is that, if you want to move queues between nodes, you have to move an entire bucket's worth of queues. This is similar to the "circle hash" used for memcached load balancing, except I think it's a little simpler because each bucket is deterministic. (I never understood the benefit of the randomly distributed circle points, TBH.)

The other benefit is that when a queue gets created or deleted, there is no change to the map!

I would have been happy to use something available from the OTP, but it doesn't have anything like this.


Now, why are there so many queues? In this design, there's a queue per data source. In MMO terms, each user has a queue for whisper messages, each NPC has a queue for state distribution, etc. Area of Interest can then be implemented as adding/dropping subscriptions to specific queues.