Distributed Computing – How to Simulate a Distributed Environment Effectively

Writing a basic queuing system is fairly simple, but as you noted above with all of the challenges, doing it right is another matter. I've used home grown systems for which I wrote the source code, 3rd party systems, and various JMS providers. JMS (Java Messaging Service) by far is the most complete solution I've encountered thus far. Much of what you ask is available in JMS. My favorite JMS provider is ActiveMQ. Free, performant, easy to install, and more importantly easy to embed in my app with Spring. JMS providers don't provide everything you asked for out of the box, but they provide a set of tools to handle much of what you asked about should your application need it. I haven't found lots of applications need everything you listed. Ordering might not be important (it's best if it isn't), durable topics might not be important, guaranteed delivery, etc. You just have to stick to the problem and use what it demands.

http://activemq.apache.org/what-open-source-integration-solution-works-best-with-activemq-.html

Does it have strong or lose ordering? Yes. It has both depending on your programs needs. Here are the details: http://activemq.apache.org/total-ordering.html.

Does it have idempotent put? No, but this is trivial to implement in your application layer should you need that.

Can we have more queues than what can fit on a single machine? Yes. You can have clustered servers, and if you wanted to setup multiple machines with different queues you could, and pull from either.

Can we have more data in a queue than what can fit on a single machine? Yes most JMS providers have to use some sort of DB/persistent storage to ensure messages aren't dropped or lost if the JMS provider goes down.

How many machines can crash before we potentially lose data? This is a little harder to answer because it's timing related. However, you can crash a JMS provider and provided the disk isn't corrupt it will come back up and start where it received the last commit. This means messages could be delivered twice, but if you code your app to handle this it's not a problem. As long as you have at least one of each type (producers, consumers, or JMS servers) it will complete. You can also have load/balance/failover for redundancy should a disk go out on you.

Can it tollerate net-splits? I think I understand what you mean by "net-split", but I'm not entirely sure. I guess you mean if the JMS servers are clustered, and we loose connection with one of the servers will it jump to another server and pickup where it left off. Yes, but again these types of situations can lead to duplicate messages depending on at what point the client lost connection.

Can it automatically reconcile data when a net-split is fixed? If you are using transacted sessions it will only redeliver any message that has had a commit called on it to existing clients that are up.

Can it guarantee delivery when clients can crash? Yes this is one of the main goals of JMS. Guaranteed delivery means that if a message is queued it's guaranteed to be handled by a client.

Can it guarantee that the same message is not delivered more than once? Yes if the transacted sessions are being used. That means a client has accepted the message and called commit/rollback. Once the commit is called it won't redeliver the message.

Can a node crash at any given point, come back up, and not send out junk? In the case where you have durable clustered queues. Yes it won't spew "junk" if the other node in the cluster has delivered the message. It can still redeliver anything that hasn't been acknowledged.

Can you add nodes to, or remove nodes from, a running cluster without down time? Yes.

Can you upgrade nodes in a running cluster without down time? This is a little trickier for me to answer, but I believe that yes you can do this.

Can it run without problems on heterogeneous servers? What does this mean exactly? I've found most JMS providers are very easy to run in environments using different hardware, OS, etc. Although, if you mean performance, that's a whole another thing. Any distributed processing system can be negatively impacted by a slow node. I had 2 8 Core Intel servers running the queue and the consumers. That's 16 cores together, and I got better performance from using only those two boxes, than when I added a single core machine as a consumer. That single core machine was so much slower it slowed down the entire grid by a factor of 2x. This had nothing to do with JMS per se.

Can you “stick” queues to a group of servers? Short answer yes. I can think of a way where you can run a cluster that's only in the european data center, and configure the queue there. Then in your spring config setup your consumers to consume that queue as well as other queues on other clusters. You might want to consult the docs:

http://activemq.apache.org/clustering.html

Can it make sure to put data replicas in at least two datacenters, if so available? Again I believe so, but it's best to consult the clustering docs.

Again JMS has lots of options you can tweak as your need dictates. Using transacted sessions and durable queues comes with a performance cost. I've seen turning on all the bells and whistles impact performance as much as 10x. When I used JBossMQ if we turned off some of these features we could get around 10,000 messages/s, but turning them on brought us down to 1000 messages/s. Big drop.

There are a number of important missing bits of information:

• Why is OpenMPI relevant?
• Why is heterogeneous relevant?
• What is the work that the server is orchestrating?
• You mention one thread per connection is a problem.

If you're using Boost::Asio and you're having the problem of one thread per client, then you're likely doing something wrong. Make sure that you're doing things asynchronously and not blocking any one thread too long. This is probably the easiest thing for you to do at the moment.

OpenMPI is not what you need. MPI was designed originally for distributed memory architectures. It is possible to use it over the internet, but it is probably not the best choice. The fact you mention heterogeneous and MPI makes me think that you've done a CS course in HPC. That is not relevant in this case.

When deciding on a protocol and architecture, the type work that the clients are doing is important, how you share state between the clients and server and how long each message takes to process.

If you're optimizing for throughput, then a per message overhead is less important, so verbose serialization formats like XML and JSON are ok. You can even tear down connections between message processing jobs - meaning that the server does not have to maintain a thread.

If you're trying to keep things low latency, then per message overhead is important, so maintaining a connection and using terse serialization formats.

A Messages Broker is an architectural pattern that you could consider. It is responsible for distributing your messages between your applications.

You're better off using internet-friendly protocols, like HTTP. You don't want to worry about Proxies or NAT traversal. The internet can be considered a massive distributed system with millions of heterogeneous clients. REST is an architectural style inspired by that so it might be appropriate.

I'm going to assume that:

• the server produces work item for clients to do and aggregates responses.
• your clients receives a message, does some work and sends a response back.

My default choice of technology, would be a web server to implement this. It would have two URLs. One where the clients can GET new work, the other for clients to POST responses.

When you initially start the client it should periodically poll for new work (you could use WS or Long Poll HTTP requests). When the server returns a work item, one client should consume it. Each work item should be stamped with a unique id, so that you can ensure that their are no duplicates and correlate responses with the initial job.

When the client has completed the task it should POST the response and restart the process of polling for new work.

Web Servers don't work well for long requests so you would want to hand it off to something else, perhaps via a DB, IPC or a message queue.

The server then does not have to worry about the capabilities of each client (so it makes heterogeneous irrelevant). The clients will just consume at the rate they can.