Pastry: A distributed hash table in Go
secondbit.org
secondbit.org
Really? You think this doesn't exist? In fact, this is nothing but a deployment concern for any mature message broker.
http://www.rabbitmq.com/ha.html
http://activemq.apache.org/masterslave.html
http://hornetq.sourceforge.net/docs/hornetq-2.0.0.GA/user-ma...
There are lots more options than these, and you can also use heartbeat/lvs to take something like redis and make it HA.
I'm glad you had fun inventing your own distributed hash infrastructure, but please do not attempt to convince other people that there are no other options out there for reliable and highly available messaging.
Rather than arguing this point again, I'll just point at the other comment thread that addressed these concerns.
Fun fact: I didn't mention highly available messaging in the post at all. I mentioned single points of failure and bottlenecks. Those aren't quite the same.
All of your solutions, based on the conversation below, separate the messaging into its own service. Which isn't really how I want to work with my software. I guess the best way I can describe it is as a different aesthetic. There are certainly ways to accomplish the same thing, technologically, but they all feel wrong to me. Introducing a message queue, with a broker, and needing to implement HA on that simply to remove that single point of failure... That feels like a workaround to me. I prefer working with a decentralised pub/sub framework. Personally. Personally, that difference is worth the development time and lack of maturity. To me.
Which is why I slapped an alpha label on the project. Why I made no attempt to convince people they should switch over, or even that my way was objectively better than any other way.
I had an itch. I scratched it. I released the backscratcher I used as open source so other people with the same itch could scratch it. I'm not sure why some people think this is a contemptible act.
Like I said, I'm glad you had fun with it. But you stated in your article, that there are not existing solutions without a single point of failure, and that simply isn't true. That's the only part I object to.
I stand by what I wrote, as it is the best way I can explain the, frankly, hard to describe issue I take with separating the messaging component into its own servers. If I had to rewrite it, I don't know how I could explain it any differently and get closer to accurately expressing myself. I'm sincerely sorry if you feel I've misled you or am misleading people; it is not my intention in the slightest. I can't imagine what I could conceivably stand to gain from such a deception.
I would, however, suggest that if people are taking their tech advice from me without doing any research of their own (which is what you seem to be suggesting, as you've commented several times that the information that you believe contradicts me is easily available), perhaps my inability to properly express my reservations with the previously available solutions is not the biggest issue at play here?
That's not what the author said, "we couldn't find" is a personal judgement which leaves things wide open.
>I'm glad you had fun inventing your own distributed hash infrastructure, but please do not attempt to convince other people that there are no other options out there for reliable and highly available messaging.
That didn't happen.
“The Spread toolkit provides a high performance messaging service that is resilient to faults across local and wide area networks. Spread functions as a unified message bus for distributed applications, and provides highly tuned application-level multicast, group communication, and point to point support. Spread services range from reliable messaging to fully ordered messages with delivery guarantees.”
We use this for LAN event communication among standalone servers acting together, and WAN event communication among POPs acting together, for billions of events per month. Any server in a group dies, they elect a new master, so no SPoF.
What was it about your use case that made this feel like a single point of failure or bottleneck?
This was Spread 3.x somewhere around 2008, so it may have improved since then, but at that time it just seemed very flaky and slow. This was in a cluster of about 8 nodes, with 100-200 messages/second published by any given node at peak, so it should have been more than adequate.
The worst part was that it was silently losing messages, and there seemed to be no trivial way to recover, get statistics/queue information, or generally determine what was going wrong.
From what I can see in the docs, RabbitMQ is a client/server relationship. Meaning there is a server. Meaning a single point of failure and a bottleneck. I hate those more than I hate writing new software. Not because I think there is less danger of bugs in this than there is danger that the AMQP server will fall over... it just feels wrong to me.
I'm not saying, by any stretch of the imagination, somebody should use this over RabbitMQ. But if a distributed hash table is more suited to their needs, or if, like me, they're allergic to SPoFs and don't mind if the software is a little less mature if it means avoiding them... Well, now they have a package in Go they can use. If they want to.
Just because software runs on multiple machines it does not mean that there is 'no SPoF', for example in a distributed messaging system you need to replicate state, unless you don't mind losing it when you have a crash.
If you don't mind losing state, then it is very easy to run RabbitMQ on more than one server.
If you do mind losing state, then RabbitMQ provides several ways to use multiple servers in a decentralised and redundant manner. See eg http://www.rabbitmq.com/distributed.html
It is very easy to write a system with multiple components. The difficulty is preserving all the additional capability (eg uptime) while maintaining a coherent system. This role can be played by a "broker" which does NOT imply a single machine or "SPoF" - it can be a distributed service. See eg http://www.rabbitmq.com/blog/2010/09/22/broker-vs-brokerless...
Are you familiar with the basics of a distributed hash table? To say that "each machine is potentially a single point of failure" is... inaccurate.
According to the algorithm's authors, you can lose the vast majority of your cluster simultaneously without affecting the stability of the cluster, in certain cases. As a worst-case scenario, you'll need 16 servers to drop offline simultaneously before there's a problem.
Am I saying my solution is better than RabbitMQ? No. Am I saying people should use my solution instead of RabbitMQ? No. Am I saying my solution is closer to what I want than RabbitMQ is? Yes. Yes I am.
My assertion was that "Just because software runs on multiple machines it does not mean that there is 'no SPoF'". You seem to have read this as saying "in all cases where there are multiple machines, there is always a SPoF". In any case, I was trying to make a more general point, which is that (A) lots of cases which seem to have no SPoF, in fact do have SPoFs, (B) some cases get accused of having a SPoF, but doing so is a mistake, and Rabbit is in this category because you can set up a number of multi-machine scenarios with the required redundancy, and (C) in many cases having a SPoF is not a problem anyway, and may even be a good thing.
You said "From what I can see in the docs, RabbitMQ is a client/server relationship. Meaning there is a server. Meaning a single point of failure and a bottleneck. I hate those..". With all due respect this represents a misunderstanding of Rabbit. Moreover you assert that 'there is a server' means that there is a 'single point of failure'. Because the conclusion does not follow from the premise, I assumed you must have meant something else.
Perhaps you could explain why you think your system has fewer 'bottlenecks' than RabbitMQ or other messaging systems. I don't think that it does but would love to be enlightened.
From your more recent comment, you seem to be saying that using a DHT is a good thing. Yes, I agree, sometimes this is the case. Are you familiar with how RabbitMQ does any of the following: clustering, HA, federation? You say that "Am I saying my solution is closer to what I want than RabbitMQ is? Yes. Yes I am.". I would love to understand this.
I get the impression that you feel I asked this in a demeaning manner. I had no such intent. But some of your statements seem to contradict the fundamentals of a DHT, so I was unsure what level of explanation I needed to do. I meant only to gauge your level of prior knowledge, in order to engage you on that level. I apologise if I come off as brusque; everyone seems to be taking the fact that I did not use their technology of choice as a personal insult, so I'm growing weary of explaining why each individual technology is not what I wanted.
> Just to be clear, I don't think DHTs are the same as messaging systems. But, they touch on a set of related issues.
Agreed. To be clear: I released a DHT. I did not release a messaging system of any sort. I do intend to build a messaging system on top of it, but its uses are not limited to that.
> My assertion was that "Just because software runs on multiple machines it does not mean that there is 'no SPoF'". You seem to have read this as saying "in all cases where there are multiple machines, there is always a SPoF".
Not really. My assertion is just that any software that is built on a DHT and the principles behind it has no single point of failure. And if you were not referring to software built on DHTs, I'm not entirely sure how the comment is relevant to the discussion?
> In any case, I was trying to make a more general point, which is that (A) lots of cases which seem to have no SPoF, in fact do have SPoFs
I would argue that many of the alternatives people have proposed to me fall under this description.
> (B) some cases get accused of having a SPoF, but doing so is a mistake, and Rabbit is in this category because you can set up a number of multi-machine scenarios with the required redundancy
Redundancy does not make the SPoF disappear, it just manages the risk of that SPoF. Perhaps I am using the term beyond its meaning here (sorry!), but I think of this more in the architecture of the cluster. If traffic is routed through a small, centralised subset of machines whose specific purpose is to handle or route that traffic, I consider that to be a SPoF, no matter how unlikely that failure may be. I consider it so because rather than architecting your cluster to avoid issues, you are offsetting the issues to an ops/deployment problem. Yes, you can achieve HA, but it is not an inherent part of your cluster; it's bolted on afterwards by some clever duct-taping as you ping servers and swap them out if they seem to be down.
> (C) in many cases having a SPoF is not a problem anyway, and may even be a good thing.
A DHT is not appropriate for all cases, though I would challenge anyone to quote me on ever saying it is. I won't even say it's the best tool for the job I'm using it for; it simply is the tool that fit best with my desired approach to the problem.
> With all due respect this represents a misunderstanding of Rabbit.
This is entirely possible. I have a decidedly rudimentary understanding of Rabbit, something I tried to convey by qualifying all of my statements about it. "From what I can see in the docs", etc.
> Moreover you assert that 'there is a server' means that there is a 'single point of failure'. Because the conclusion does not follow from the premise, I assumed you must have meant something else.
In my understanding, the conclusion does follow from the premise: a server designates a machine that is specifically intended to handle requests. Clearly, I'm misusing the term SPoF. My apologies for the confusion caused by that. In the six months I worked on this, I never once had to explain why I thought a DHT fit my needs; I was speaking to people who worked with distributed systems, so no explanation was needed. This left me a little ill-prepared to explain myself.
> Perhaps you could explain why you think your system has fewer 'bottlenecks' than RabbitMQ or other messaging systems. I don't think that it does but would love to be enlightened.
It's entirely possible my system does not have fewer bottlenecks than RabbitMQ. Again, I'm no RabbitMQ expert. Here's why I think my system has few bottlenecks: * No change in code or deploy practices is needed between one server and one billion servers. * Unless a catastrophic event hits the cluster (an event that would leave your application non-functioning, even if Pastry continued functioning), the cluster will remain healthy as servers come and go. This is not a remedy put in place by ops, it is not a bolted on feature, it is a core premise of the algorithm. I prefer to solve my availability concerns at the software level, rather on the deploy level. This might be a virtue of the fact that my software is open source, so I try to make it simple for others to deploy. It might be a virtue of the fact that I am more familiar with writing software than I am with deploying software. * The messaging component is not an element in the architecture; rather, it is an embedded piece of every single element in the architecture that wishes to take advantage of it. There are no messaging servers, there are no brokers, no queues. There is simply your architecture, except now it can communicate efficiently.
Based on these three points, Pastry fit in with the approach I wanted to take in my architecture. It seemed like every other messaging protocol I could think of preferred to have a messaging server, instead. Even if you have a pool of these servers, allowing for HA through redundancy, that is not really what I was looking for.
> you seem to be saying that using a DHT is a good thing. Yes, I agree, sometimes this is the case.
We are in agreement, then. Examine your problem, then choose a tool for the job. A lot of people seem to be taking issue with the fact that I did not contort the problem until it fit pre-existing solutions, instead of creating a solution to the problem I saw.
> Are you familiar with how RabbitMQ does any of the following: clustering, HA, federation?
I am familiar with this page: http://www.rabbitmq.com/ha.html In addition, I have seen this page: http://www.rabbitmq.com/distributed.html
Both of them feel like a bolted on solution to the problem of distributed message passing, rather than an inherent design characteristic. Allow me to quote:
> some important caveats apply: whilst exchanges and bindings survive the loss of individual nodes, queues and their messages do not. This is because a queue and its contents reside on exactly one node, thus the loss of a node will render its queues unavailable.
This is understandable; queuing is pretty much impossible (as far as I know) to achieve in a distributed system. But I don't need queuing, so why should I let that hamstring me unnecessarily?
> should one node of a cluster fail, the queue can automatically switch to one of the mirrors and continue to operate, with no unavailability of service.
This does not feel like an inherent aspect of the design. This feels a lot like a deploy detail.
> In normal operation, for each mirrored-queue, there is one master and several slaves,
When I hear "master" and "slave", they translate in my head to "bottleneck/SPoF" and "backups".
> Clustering connects multiple machines together to form a single logical broker.
That sounds an awful lot like a single element in the system that is responsible for message traffic. There may be a lot of machines in that single element, but it is a single element nonetheless.
I am not saying that RabbitMQ is a bad solution for message passing, nor am I saying my problem couldn't be solved by RabbitMQ if I changed my architecture to fit Rabbit's needs. All I am saying is that I have a preference for architectures that does not take advantage of the things Rabbit does really, really well, and does take advantage of the things that a DHT does really, really well. So I'm more than a little confused that people are up in arms over the fact that I used the paradigm that matched my preference instead of trying to force something to do what it was not intended to do.
First, re "whilst exchanges and bindings survive the loss of individual nodes, queues and their messages do not" -- this needs to be clarified. When queues are replicated, their messages do survive, in order, in the replica. The queue itself may die but the client can get back into context by finding another node in the group. This is observationally similar to DHTs.
Second, "a single logical broker" and "sounds an awful lot like a single element in the system that is responsible for message traffic".... NOOOOOO the point of 'logical' is the opposite of what you say!
Finally the BIG difference between DHTs and something like Rabbit is that in a DHT each datum is replicated around the ring N times, non-uniformly. Whereas in a system like Rabbit each datum is replicated N times in a uniform manner, such that ordered pairs of messages A and B will exist in every queue which has either A or B.
I think you're missing the point a little.
Looking at the intro, it looks like I have to hardcode the IPs and ports of all of the machines I want to publish or subscribe to. Is that correct? If so, that's a little fragile for my tastes. I like "upgrading servers" by standing up new servers, testing they work, and then changing the floating IP associated with the DNS to point to them instead.
And I can hear you say "Aha! Floating IP! Just use that!" but for most cloud providers (I believe), that is billed as external bandwidth, which is not free like internal requests are. Finally, if I want to stand up more servers to scale horizontally, I'd still have to modify my code and redeploy it to all my servers. I think. Unless I'm missing something about how 0mq works. Which I totally could be.
On the whole, learning what I needed to learn to make Pastry happen made me a better programmer, too. So if for no other reason than that, I'm glad I did it.
0mq is low level, it's not going to do routing like Pastry, but you can get away with having one fixed location for a "lookup server" that keeps track of everything else's location (something like the Doozer project, also in Go https://github.com/ha/doozerd ).
Also, 0mq only needs to know the IP of the publisher, for pub-sub.
I was familiar with doozer, but doozer wasn't(/isn't) being actively maintained and it doesn't compile against the latest version of Go, so I'd have to bring it up to speed before I could use it anyways. Not saying it would be harder, but between that and writing the 0mq library (pretty sure one does not exist in Go yet), I'd estimate the work would be more or less equivalent and yield just about the same stability. Seat of the pants guess, but it makes me feel better, at least.
The mailing list suggests http://github.com/4ad/doozer and http://github.com/4ad/doozerd are being maintained as forks of the original. Or there's Zookeeper. Or brew your own central lookup server using 0mq that does the same job of "set a key, get a key, notify subscribers when it changes". It's a single point of failure, but since the system will run fine without it (only lacking topology updates) it's not a hugely problematic one.
The lookup server is a single point of failure. And while it's not a hugely problematic one, it also is a dedicated machine whose sole purpose is keeping the other machines running. Which feels wrong to me.
It probably wasn't the best business decision to invest time in this, and I won't even argue this is the best technical solution. But it's the only technical solution that didn't make me feel like I was working around limitations; things just worked they way they were supposed to. The API servers received an event, they told the WebSockets servers about it. It felt very conceptually pure to me. I'm a sucker for that.
Doozer is only slightly a SPF, it has high availability by having multi-master replication. It's also not doing a lot of communicating, or a lot of CPU work, and it keeps its data in RAM, so it may be OK for it to live on a non-dedicated machine.
Sorry if I come across as criticizing your admittedly cool creation. It was just that you said there weren't alternatives, and I knew of one.
It doesn't have many-to-many baked in, though. That I can see, at least.
It looks like I may have been able to make this work, but the setup would be a little complex.
... because a distributed hash table is not complex at all. -_-
Suppose your architecture is API -> Web sockets and you want to make it highly scalable, then what I'd do is:
1. Many API -> few relays (use doozerd to keep up to date as to where the relays are and select one randomly, use ZMQ_ROUTER socket type to enqueue API->relay messages).
2. Few Relay -> many web socket (again using doozerd for the web socket nodes to find the relay, and using ZMQ_PUB on the relay and ZMQ_SUB on the web socket nodes).
The problem is you're confusing the messages for data. As far as Pastry is concerned, your messages are not data that it should retain. They are one-off communications. The only data that Pastry is concerned with storing is the state of the cluster; that is to say, Pastry only considers "data" the information about enough nodes to reliably be able to route a message to any node in the cluster. Your messages, by default, are lost the moment they are sent, forwarded, or delivered. The node stops caring about them at that point.
However, on top of this framework you can build applications that do care about the data in the messages. Before a message is forwarded, it fires off a callback to any application that cares to listen for it. When a message is delivered, it fires off a callback to any application that cares to listen to it. You can register callbacks that store this information, then build your own redundancy frameworks on top of that to store the data you'd like to retain.
If you're interested in this, I'd recommend paper on PAST. It's a high availability storage system built on Pastry: http://research.microsoft.com/en-us/um/people/antr/PAST/hoto...
The keys and values, at least in Pastry's case, are the NodeID and the metadata about the Node, respectively. According to my understanding, at least. I had to learn a lot to implement this, so I would not claim to be an expert on distributed hash tables by any stretch of the imagination.
It's about maintaining the property that you can route information from many different places and have that information end in a consistent place, no matter where it's sent from. And, assuming the consistent place doesn't go offline, a large number of servers can fall over without warning before that property is changed. It's about making discovery of nodes in your cluster reliable, and passing information in between them.
The use cases for that are many, and each has their own nuances, which is why this is so confusing and hard to talk about.
We chose to solve the discovery problem a bit differently in NSQ (https://github.com/bitly/nsq) but I could certainly see some interesting opportunities to experiment with a distribute hash table approach as well.
I was very impressed with it, though. You guys did an awesome job, and I'm extremely flattered you stopped by to comment.
A "channel" receives a copy (at the source) of all the messages for a "topic" and has its own set of clients subscribed.
I'd like to experiment a decentralized twitter/reddit-like system using p2p message flooding and machine learning to weed out spam.
Going back to P2P: building a flooding P2P library on top of a DHT seems like a total waste. But maybe the common parts could be extracted.
There's Essayist, a (very alpha) long-form blogging app, hosted at http://essayist.mndj.me, code at https://github.com/mwanji/essayist
Meanwhile, the inventors of Tent have 5 people working 60+ hours a week running Tent.is and developing the protocol and apps for it. They haven't revealed what's next, though it sounds like some non-status apps are in the works.
I'm asking about P2P, even if client/server might be a better answer to that specific problem.
How's your Go? You can build one on Pastry. :)
https://github.com/quartzjer/TeleHash/blob/master/org/v2.md
My conclusion was that TeleHash is a cool concept, but not a solid foundation you can build something on just yet.
Opted to make a choice I was happy with instead.
And I mention it in the blog post, and I mention it in the README, but I feel I should reinforce this:
This. Is. Alpha. Please don't use it in your mission-critical code.
I'd like to run a lot more tests on it. I'd like other people to run tests on it and let me know what they think. This is my first time writing anything this complex, and I want any and all feedback I can get on it.
Thanks!
The basic gist of it is similar to a phone tree. You route a message subscribing to a topic, basing the message ID on the topic name. As the message routes through the network, each Node stores the information about who is subscribed. When an event happens, that information is passed back through to the subscribed nodes.
I may have to tweak it a little, as I did Pastry, but I'm pretty sure the same basic concept will work for my needs.
Spotted a typo on the line "OnNodeExit", "Applcation".
And I guess my point is, I've tested this. I can't seem to find any problems with it. But I do not have unit tests for the larger routing algorithm, only the routing within a Node. In my tests, it always worked out, but I would put my confidence in this at about 70%.
In essence, I was ready to release this software to the Go community, have them kick it around a bit, see what came of it. I was not expecting a HN frontpage. So now I'm slightly panicked, and hoping to God that people don't think I'm an idiot for the rest of forever.
If you're going to have any component that's a single point of failure, redis is a pretty good bet for it.
I just really didn't want to have a single point of failure. Really badly. Pastry has no single point of failure.
[edit] Realised this thread isn't actually all about me. Oops! Sorry for answering a question not directed at me.