Topics are not Message Types

I periodically have an unproductive conversation about how to use Topics/Categories vs how to use Message Types. Hopefully this time will be better.

Both things appear to be used to "subscribe", and both wind up filtering what a message handler has to process and gets to process. If they can be used for exactly the same purposes, it is "just" policy as to what you use each one for. That has to be wrong, otherwise there would not be *two* concepts. Tus, there has to be a useful distinction. So let's define what they are and what their responsibilities are.

First a definition or two:

• Hierarchical: a name is defined hierarchically if the parent context is needed to ensure the child is distinct from children of other parents when the children have the same name. The parents provide the namespace in which the child is defined.
• Orthogonal: names are independent of one another, like dimensions or axes in mathematics.

Categories are names (or numbers) that are used to decompose a stream of messages into groups. In JMS they are called Topics, but I'm going to avoid that term in case the implementation of Topics in JMS implies something I don't mean. A message is sent on, or "to" a single Category. A consumer subscribes to one or more Categories. Sophisticated message publish/subscribe or producer/consumer implementations can support wildcards or bitmasking to optimize subscription to large sets of Categories. (While not very germane to this discussion, I believe JMS can only have wildcards at the end of a Topic, and only at a dot that separates portions of the Topic. My view of wildcards and Category masking does not have that limitation. But that shouldn't affect my arguments.)

It is critical to have a mechanism that efficiently filters network messages so that a consuming process is not "bothered" by messages arriving that are immediately discarded. Running the TCP stack, for example, can wind up consuming large fractions of the CPU, and if the message is discarded, even after a simple inspection by your message framework, that is totally wasted processing. Further, if the messages are traveling over a low bandwidth link to a player, for example, it can badly affect their experience as it steals network resources from more important traffic. So we want the sender, or some intermediary to filter the messages earlier.

Early distributed simulation implementations (DIS) used multicast groups, and relied on the Network Interface hardware to filter out any messages in groups that the consumer had not subscribed to. Ethernet Multicast tends to broadcast all messages, and rely on the NIC of each host to inspect and filter unwanted messages. That is better than having the kernel do it. Switches get into the picture, but are very simplistic when it comes to multicast. When there are more than a few groups, switches and NICs will become promiscuous, and all messages get broadcast anyway, and wind up in each destination's kernel. They are filtered there, but much of the network stack has already executed. To get around that, physical network segmentation with intelligent bridges were built to copy a message from one segment to another. The bridge or rebroadcaster or smart-router would crack open each message and send it into another segment based on configuration, or a control protocol (subscription request messages).

Ancient history. However, it formed the origin of the concept of numeric Categories. A message is sent to a single Category. A consumer subscribes. The Channel/Category/Subscription manager maintains the declared connectivity and routes the messages.

So. Categories are used to optimize routing. They minimize the arrival of a message to a process. So far, this has nothing to do with what code is run when it arrives.

Message types are also names but are used to identify the meaning of a message; what the message is telling or requesting of the destination; what code should run when the message arrives (or what code should not run). Without a message type, there would be only one generic handler. In the old days, that master-handler would be a switch statement, branching on some field(s) of the message (lets call that field the message type, and be done with it).

There is some coded, static binding of a message type to a piece of code; the message handler. Handler X is for handling messages of type Y. A piece of code cannot process fields of a message different than what it was coded for. There is little reason to make that binding dynamic or data-driven. Static binding is "good". It leads to fewer errors, and those error can be caught much earlier in the development cycle. Distributed systems are hard. You don't really want to catch a message-to-code mismatch after you've launched. One way to think about this static binding is as a Remote Procedure Call. You are telling a remote process to run the code bound to message Y. In fact, you can simplify your life by making the handler have the same name as the message type, and not even register the binding.

A message can be sent to any Category regardless of the message's type. There is no checking in code that a choice is "legal". The Category can be computed, and the message is bound to that value dynamically. Instances of the same message type can be sent to one of any number of Categories. Consumers can subscribe to any Category whether they know how to process all the message types it contains or not.

So. Back to the distinction. When code is declared to be able to handle messages of type Y, that does not imply that all message instances of type Y should arrive at the process with that handler. You may want to do something like load balancing where half the messages of type Y go to one process, and the other half go to a tandem process. So message types are independent of Categories. The two concepts are orthogonal.

When a process is subscribed to a Category, there is no guarantee to the subscriber about the message types that a producer sends to that Category. It is easy to imagine a process receiving messages it does not know how to handle. The sender can't force the receiver to write code, but the sender can put any Category on a message it wants. So Categories are independent of message types. The two concepts are orthogonal.

Now. With respect to hierarchy. Message type names can be declared within a hierarchical namespace. That can be pretty useful. At the end of the day, however, they are simply some strings, or bit strings. In a sophisticated system that maps message types to message classes (code), the class hierarchy may mirror the type name hierarchy, and have interesting semantics (like a handler for a base message class being able to handle a derived message class). But mostly, message type name hierarchy is useful to avoid collisions.

In systems like JMS, Categories (Topics) are also hierarchical. This is also done to avoid collisions in the topic namespace, and for organization. But it is also useful for wildcard subscription.

Now "the" question: are Categories within the Message Type Hierarchy, or are Message Types within the Category hierarchy? Or are they orthogonal to one another? I submit that a message of a given type means the same thing no matter which Category it arrived on. Further, the same message type can be sent to any Category and a Category can transport any number of different message types.

Since there is only one message exchange system, Categories cannot be reused for two purposes without merging the message streams. That leads to inefficiency. If you reuse a message type name for two different purposes, you run the risk of breaking handler code with what appears to be a malformed message. That leads to crashes. You could permit that kind of reuse, and institute policy and testing to keep those things from mingling (e.g. reuse message types, but only on different topics), but it is a looming disaster. I would put in some coordination mechanism or name spacing to keep the mingling from happening at all.

So what are the consequences:

• There is no need to include Category when registering a message handler. 
• Category subscription occurs separately from handler-to-message-type mapping, and affects the entire process.
• There is no need to build a message dispatcher that looks at Categories.

Well. That was pretty long winded. For those of you still here, I have an analogy. I haven't thought it through a lot, but it looks like it fits (although it is about a pull system, not a push system). URLs. The hostname and domain name represent a hierarchical Category or Topic. The path portion is the message type and identifies the handler (web service), and is also hierarchical. You can host your web site on any host on any domain, and the functionality would be the same. You can host any web site on your host. You can host any number of web sites on your host, provided the paths don't collide. If they do collide, you are going to get strange behavior as links refer to the wrong services, or pass the wrong parameters. One would need more hierarchy. Or you don't host the colliding web sites together. You put them on different addresses. But the service code doesn't care what address you choose.

Unless you talk about virtual hosts, or virtual processes, multiple independent connections to the message system, thread-local subscriptions. You can do *anything* in software. But should you?

Adaptive/Dynamic Categorization

Any static partitioning of data flows using a static algorithm (like a geometric grid) will break down. Either due to something intrinsic, or because the players are evil, and delight in ignoring your simplifying assumptions.

So for stability and performance/scale reasons a categorization policy that is dynamic will be better. Periodically a centralized or distributed computation can inspect the in-use categories and determine whether there is a better mapping.

Imagine there was a flash-crowd come to see an epic battle. The initial mapping had that area mapped to one category. The area can be split into two area and given two distinct categories. At a specific point in time, the new mapping will be instituted. Call that the next epoch. We need to do that across all users of those categories. The easiest way would be to broadcast the new mapping. Once all participants have acknowledged they have the new mapping and re-subscribe using that mapping, producers can start using the new categories. Once all participants have acknowledged adopting the new mapping, the old categories can be unsubscribed from.

There are many interesting dynamic categorization policies. They are based on live measurement of the system as the game is being played. The new categorization is then applied live. Clearly this can provide better efficiency than offline analysis and updating a static mapping, since in-game patterns can change minute to minute.

• Split and merge geometric grid cells or triangles. This is how many mmo systems work today, so if that is your preference you can keep it while migrating to using categories.
• Compute "nearness" between producers and consumers, and identify clusters of entities. The set of entities identified is given a category. As entities move around, they subscribe to other clusters of entities. Once the interconnections become too "stretched", a new clustering can be computed. There are pretty simple clustering algorithms, such as inspecting R-squared for an Entity against various proposed clusters to find the nearest one inside some threshold. Note that this approach can be applied to individual entities incrementally as they move.
  
• A similar clustering algorithm can be applied to actual communication patterns as opposed to assuming that geometric nearness always indicates high interaction rates. The recategorization system would record and analyze metrics of interactions as they change in real time.
• The hosting software can detect overloads of various resources, and indicate to the recategorization system that splitting off instances of busy areas would be valuable (assuming your game design supports that approach).
• As dungeons or PvP arenas are spawned, a new set of categories especially for the players involved will be computed and used. Because the subscribers indicate which instance they are currently assigned to, the categorization policy will compute the Categories for that instance, and a player will not see anything from a different instance.
• A game may have unique capabilities like being able to add, remove or move land masses. As coordinate systems drift around, the categorization policy would need to take this into account. It may be able to do that using relative coordinates, or it may need to completely recategorize space.
• Changes in team or guild size may trigger recategorization.
  
• As a clarification, in contrast to dynamic categorization, consider this scenario: As density increases in an area, the policy can adjust the maximum sight distance. While this doesn't directly change the decomposition of the data, it will result in fewer categories being consumed. (This is really a subscription policy change, not a categorization policy change)

Really, the key thing is to consider that real time feedback can be used to help improve the data distribution performance of your system.

Obviously an algorithmic policy is better than an enumerated one, since distributing new parameters to factor into a computation requires less data than a complete enumeration of the descriptions for each category.

Like almost any distributed computation, recategorization can benefit from incremental changes. Local measurements and performance triggers can instigate a categorization change to just a few mappings, and may be able to be distributed to only a few machines. These incremental changes can quickly ripple through the category space and come to an approximately optimal mapping. One benefit of the incremental approach over the periodic approach is that the trigger to recompute can be an event as opposed to a polling of metrics and a recomputation that may have limited or no effect.

Static Categorization Policy

A Static Categorization policy is one that partitions the data flows in an online game using a fixed function (usually a function of the values of the data being partitioned). The algorithm used does not change, and the tuning values do not change during gameplay. The performance of the system is determined by the designer during development testing. There are many examples:

• A 2-d grid with infinite altitude. Any entity in a grid cell transmits to the associated category. Any entity that can see into that grid cell subscribes to that category. And the grid size does not change. This can lead to lots of categories (not a real problem if your categorization implementation is designed for that), too many entities in a cell (if the cells are tuned too large), too many subscription changes (if the cells are too small). Etc.
  
• Variations of the grid approach include non-uniform grids (quad-trees), or arbitrary polygon "tiling".
  
• The designer could identify rooms, or spaces between rendering portals. Each would be assigned a unique category, and all data within them would be grouped. Only when near doors would a consumer subscribe to more than one category.
  
• FPS engines have used BSP trees to optimize rendering and visibility (and auditory effects) calculations (e.g. http://developer.valvesoftware.com/wiki/BSP_Map_Optimization) Whether automatically computed ahead of time or defined with hints, these 3-d volumes could have a category each. It is easy to query the data structure to determine the category corresponding to the current position. Doing an approximate spherical or convex volume query of the BSP tree should also be easy for consumers to find all their needed categories.
  
• Data limited to team, quest party, or clan can be separated each into its own category. A consumer (client) always knows which "entity set" they are in, so can determine where to publish and where to consume. Same with chat channels.
  
• Chat channels are interesting in that you may want to use a 3rd party chat infrastructure, but use in-game position or group membership to determine connectivity. Categories can be very useful for that, or if you want, could be used to route chat through the game engine. That way gameplay mechanics could affect chat (jamming, ambient "noise", battery charge, ...)

Again, all these are considered static because the algorithm and tuning do not change. If something gets overloaded there is no recourse to retune at runtime. Best you can do is change the policy in the next patch.

Note that membership is dynamic, however. That is how the system is able to limit what data is received (for security and bandwidth reduction), but still get everything that is needed to the right consumers.

Dynamic categorization is even more fun, but has some interesting challenges. More next time.

Subscribing ahead

Interest management is the process of delivering interesting data to those interested. An entity ensures that its sensor model has all the correct entities available to it by subscribing to the set of categories known to contain those entities. That subscription fills the local cache with replicas of the interesting entities. Then the sensor can be implemented to immediately query the cache knowing that it contains accurate data.

The policy the application uses to fill the cache can be as simple as a grid, or as complex as a dynamically determined set of clusters of nearby entities. In either case, the consumer knows the category being used by any entities that might be within the consumer's area of interest, field of view, line of sight, range of hearing or what not. Note that the producer generally produces into a single category.

So now producers send updates, and interested consumers get them. But what if someone moves? They will change the category they produce into, or the set of categories they consume from. But what about latency? That is where it gets "fun". A consumer can use its maximum velocity and an estimate of subscription latency to compute the extra distance needed to guarantee that the cache contains those moving entities even when the producer and consumer are moving toward each other at maximum velocity. That will ensure that its cache is populated with any entities that will-be interesting in a few moments. The cache will necessarily have more entities than are logically visible, so the sensor algorithm must filter out ones that are too far away but have been delivered "just in case".

But what about producers that are moving? We use latency hiding and bandwidth optimization to reduce the number of updates being sent, and then predict where the producer is at the current time. But that increases the latency of the produced data. OK. We increase the subscription range even farther. But you have to factor in the producer's maximum velocity. And you don't know what type of entity it is. So you have to use the maximum velocity of the fastest entity in the game.

That can suck if there are a few jet fighters, but tons of infantry. The solution? Separate the fast movers from the slow movers and use two distinct "layers" of categories. You would subscribe out quite a bit further for fast movers, but wouldn't get any slow movers in that wide subscription because they are produced into a different layer. And there won't be very many fast movers to consume. If you think about it, there is just no way to consume all the data from fast movers that are far away and might suddenly move toward you if there are a lot of them. The worst case example is an entity that can teleport anywhere instantly. Consumers would have to be subscribed to everything. Or you would have to change the game design to hide the latency from subscribing on-demand (like making a poof of smoke, or invulnerable just after teleport, or something).

The beauty of category based subscription is that these game-specific factors can be reduced to a set of integer values that can be computed by both producer and consumer. The consumer doesn't need to know that there are entities in the interesting category. If there are, the publish/subscribe system will deliver them. All that system does is match up integer values: hey this thing goes to these guys. The system doesn't need to assume that the categories are related to geometry, or that they are a linear function, or anything. You can use them as sets. Or unique addresses for multi-point to point communication, or for radio broadcasts on a single station. Or anything.

But my point: you have to use prediction to reduce bandwidth. But you also have to use prediction when subscribing or might miss something "interesting".

Networking Resource

Here's an article with lots of links to classic networking techniques:
http://gafferongames.com/2009/01/25/game-networking-resources/

Here are some research papers leaning toward large scale issues:
http://maggotranch.com/biblio.html

Entity Migration Mechanism

From earlier posts, we see that to achieve scalability, we need:

• large numbers of simulators
• an ability to load balance based on load (not just geography)
• an authoritative simulator to avoid DB bottlenecks
• a single-write paradigm to avoid overly complex synchronization

And since our world scales with the number of Entities, not the number of functions in the game, then load balancing, and thus, scalability is realized using Entity migration.

Setting aside the policy and impetus for initiating an Entity migration, lets talk about the mechanics. By separating policy and mechanism, we can experiment or customize the policy to use application-specific information, resulting in a closer to optimal solution. And we won't have to reimplement the mechanism each time.

We know we can run an Entity on any host by using interest management as discussed earlier to feed an Entity everything it needs to operate correctly. So all we really need to realize a migration is:

• getting the Entity state onto the new host
• getting the data flowing to that host that is needed by that Entity
• doing this quickly enough that there are no hiccups visible to the players
• avoid all ordering and race conditions so there is no game logic difference compared to not migrating (no side effects)
  
• survive crashes of any component at the worst possible moment (i.e. preserve important transactionality) without significant impact to the players

First, we use a data-driven means to identify which state variables need to be transferred to the new host. There is no reason to transfer truly temporary variables, but there are reasons to transmit variables that are not needed in the persistent database. E.g. current target. There are many mechanisms to serialize an entity and reconstitute it. One challenging aspect is whether to transfer the execution context (e.g. the stack and program counter) if your simulator uses coroutines to support blocking and waiting in a Behavior script. For example, Stackless Python is famous for pickling coroutines and reconstituting them.

There is a handshake needed to get this state across:

• suspend further execution of the entity so things don't change during the migration
• transmit the state
• recreate the entity on the target host
• resume execution of the entity.
  

Seems simple, but consider that update messages to players from the original and destination host might get out of order, the DB request queue might get backed up on the source (after all you are migrating away from a busy simulator) and save requests to the DB might get out of order, replicated state on the source and target might be at different versions (the entity may see a neighbor jump backward or forward in time).

So we need to add some steps:

• Get the target subscriptions set up and acknowledged before the transfer so when the Entity arrives, all data is available there that it had in its original location
• Have the original simulator "flush" its DB queue so the DB never sees out of order persistence requests, and then stop persisting that entity until after the migration.
  
• increment an "epoch" counter to allow us to discard any replication messages or requests from the past
• Given the increase in time and complexity, it may be worth optimizing the process by pre-loading the target host without actually pausing the original entity. Then once everything is set up, resend states that may have changed during the preload. Of course, you might also make use of any state that was previously replicated to the target.
  

There a quite a few trick available and needed to get this to come out right and be efficient. But distributed transactions like this happen reliably in a lot of "critical" systems in other industries so it is quite solvable. You can see how the need for migration and the requirement to do it quickly without player-visible hitches requires us to adopt many of the design principles already accepted: authoritative simulator, interest management, data-driven persistence and replication traits describing entity state variables, ... All of these key features are intertwined, so if your systems goes off track somewhere there, you may be buying a lot trouble elsewhere.

One of the coolest features of interest management is that you can choose to not migrate and the game still runs the same (but may use more datacenter-only networking). So if you can't migrate an entity until it finishes a behavior (because you can't migrate your stack), no problem, just wait. Program that into your policy. If you find that hitches are visible but only when a player is in a heavy combat situation, your policy can delay initiating the migration until the participants have been quiescent for a while.

Publish/Subscribe Message Delivery

In a previous post, I argued that publish/subscribe was the only tricky thing needed for a totally flexible interest management based online game system.

Publish/subscribe (producer/consumer) based message systems give semantics similar to multicast. A producer sends a message to a channel. All current consumers on that channel receive a copy of that message. To avoid becoming broadcast (where every consumer receives every message sent), the messages are decomposed into channels using a Category, one per channel (so you can think of it as a channel id). A Category is an integer so they are trivial to deal with at the lower level (as opposed to strings or something). For simplicity, each message is only sent to one Category.

This system is very loosely coupled giving it a lot of flexibility and extensibility. A producer does not need to know the existence of any of the consumers. The set of consumers and their implementation can change without touching the producer. For example, a logging system could be attached to a channel without affecting the system, and would give good data for debugging.

To implement the publish/subscribe system efficiently, we must manage the producer and consumer subscription requests efficiently. Broadcasting that a consumer is interested in some Category to each producer is too inefficient. So we introduce the notion of a channel manager that keeps track of the interests of all producers and consumers.

The channel manager is responsible for redistributing each data message. A producer sends a message to the channel manager. The channel manager maintains the list of interested consumers, and forwards a copy of the producer's message to each consumer. We have exchanged the non-scalable broadcast of subscription messages for an extra hop of latency for each data message.

The channel manager can easily be made scalable. The simplest approach is to use the integer Category value and a simple a modulus operation to load balance across any number of channel manager processes. Both producers and consumers use the same computation. And all subscriber messages and all data messages on one Category travel through a single channel manager.

This architecture is the obvious one. There are more sophisticated approaches that can reduce the two hop latency by using a direct connections between producers and consumers. The subscription messages still need to route through a channel manager, but the producers need to maintain the list of interested consumers. This adds the requirement that producers subscribe to produce, and adds more subscription messages and more latency on a subscription. There are also subtle data message ordering problems.

If you want to go nuts, you could use real multicast. The challenge there is that there are limited numbers of multicast groups. So you have to solve the problem of multiple channels sharing one multicast group.

So you get to choose. Easy implementation or optimized but tricky implementation. Like most code. In this case I argue that the simple approach has good enough performance for the needs of online games. The producers and channel manager live in a data center on hosts attached to a high speed switch, so network latency is minuscule.

The design philosophy of this system is to minimize unnecessary computation due to unwanted messages arriving on a host that are just thrown away. Hosts cost money. Bandwidth inside the data center is free. So good interest management is key.

So. We have sliced off the publish/subscribe problem. All we have left is how to approach interest management policies which are application specific.

Back to First Principles: Interest Management

When you think of an Entity needing to interact with its environment, you don't tend to think about arbitrary lines running through the geometry. (There is no colored line on the ground between countries). In fact, players delight in trying to find those exceptions and do things while standing on either side, or jumping back and forth as fast as possible.

The way to think of the problem is that geometric decomposition is solely to support load balancing. Stuff on this side runs on this host, stuff on that side runs on that host. Much of the rest of the system just takes advantage of that assumption. (And it is not such a great assumption.)

But what if we ignore load balancing, and just think of the Entities all over the place trying to interact? At the extreme, each Entity would be on its own host. Now we have classical distributed systems problem, and can tap into that knowledge.

Distributed object technologies, like CORBA, hide the fact that some objects are remote by using a local smart-proxy. Interactions by a locally owned/executed Entity with the local proxy are forwarded to the remote-original object. The big problem here is that CORBA can block the requestor, and the request has a round-trip latency.

The better way to solve this is to ensure the local proxy is already up to date before the local Entity starts interacting. This allows the proxy's state to be as accurate as physically possible (the local proxy is at most out of date by a one-way network latency unit of time).

Now we have to solve the interest management problem. The system wouldn't scale if we broadcast Entity updates (in both network consumption and in space for storing the proxies). Here we rely on a few restrictions that we think are not too onerous. The Entity must declare what it is interested in, and it must never write directly to a local proxy.

The simplest interest management approach is to break the world into tiles. If an Entity can see into a tile at all, it is interested in all of that tile. If another Entity is currently located in that tile, it publishes its state updates to that tile. Using a publish-subscribe communication mechanism, all interested Entities consume every Entity's state that they can see. (There are much more interesting interest management approaches we will discuss later).

The result is, we don't have the nasty load balancing problems of other systems. The host on which an Entity is running doesn't matter. Two Entities can interact with each other no matter where they are hosted. And the simulation operates the same way it would in other systems.

The only remaining technical challenge is building a publish/subscribe system that is reliable and efficient.