Here's a quick status update.
Logical addresses are almost done and work with the UDP and TCP transports. I have yet to support TUNNEL (GossipRouter), but because Vladimir made a fair amount of changes to GR/TUNNEL on CVS head, I'll have to merge the logical addresses branch back to head first, before I can apply logical addresses to CVS head.
Because I have some other important changes in 2.8 (e.g. anycasting support), I decided to curtail the scope of 2.8 a bit and move some stuff to the newly created 2.9. For example, currently there can only be 1 physical address associated with 1 logical address (UUID). Multiple physical addresses will be supported in 2.9.
Also, canonicalization of UUIDs into shorts was pushed into 2.9. This is largely an optimization.
Speaking of optimizations, performance of the branch looks promising ! Although we now ship both dest and src addresses with a Message, which makes the serialized message a bit bigger (for IPv4, *not* IPv6 !), UUIDs take up less space in memory and thus I got a throughput increase from 105MBytes/sec to 113MBytes/sec ! These are only preliminary results, and I have yet to run a full perf test.
I'll probably merge the branch back to head and then make the necessary changes to TUNNEL/GR this week. Then we might release an alpha, for folks to try out logical addresses.
The thing is that I'll be traveling a bit for the next couple of weeks: next week I'm near Amsterdam and April 6-9 I'll be in Munich, teaching the JBoss Clustering course (JB439). If you want to meet over a beer, or even join the course, drop me an email !
Cheers,
Monday, March 16, 2009
Monday, February 16, 2009
What's cool about logical addresses ?
Finally, logical addresses (https://jira.jboss.org/jira/browse/JGRP-129) will get implemented (in 2.8) !
For those of you who've used JGroups, you'll know that the identity of a node was always its IP address and the port on which the receiver thread was listening, e.g. 192.168.1.5:7800.
While this gives you a relatively compact and readable address (you can deduct from the address on which host it resides), there's also a problem: this type of address is not unique over space and time.
Let's look at an example.
Say we have a cluster of {A,B,C}. C's address is 192.168.1.5:7800. Let's assume A has sent 25 messages to C and C has multicast 104 messages. We're using sequence numbers (seqnos) to order messages, attached to a message via a header.
So the next message that C will multicast is #105 and the next message it expects from A is #26.
This is state that is maintained by the respective protocols in a JGroups stack.
Now let's assume C is killed and restarted. Or C is shunned, therefore leaves the channel and then automatically (if configured) reconnects. Let's also assume that the failure detection protocol has not yet kicked in and therefore A and B will not have received a view {A,B} which excludes C.
Now C rejoins the cluster. Because this is a reincarnation of C, it creates a new protocol stack, and all the state mentioned above is gone. The reincarnated C now sends #1 as next seqno and expects #1 from A as well.
There are 2 things that happen now:
So how are logical address going to change this ?
A logical address consists of
This means that the channel's address will always get printed as "node-4". Under the cover, however, we use a UUID (for equals() and hashCode()), which is unique over space and time. The UUID is recreated on channel connect, so the above reincarnation issue will not happen.
The logical name is syntactic sugar, because if we have views consisting of UUIDs (16 bytes), that's not a pretty sight, so views like {"node-1", "node-2", "node-3", "node-4"} look much better.
Note that the user will be able to pick whether to see UUIDs or logical names.
Also, if null is passed as logical name, JGroups will create a logical name (e.g. using the host name and a counter).
A UUID will get mapped to one or more physical addresses. The mapping is maintained by the transport and there will be an ARP-like protocol (handled by Discovery) to fetch the initial mappings, and to fetch a mapping if not available.
The detailed design is described in http://javagroups.cvs.sourceforge.net/viewvc/javagroups/JGroups/doc/design/LogicalAddresses.txt?revision=1.12&view=markup.
So the most important aspect of logical addresses is that they decouple the identity of a JGroups node from its physical address.
This opens up interesting possibilities.
We might for example associate multiple physical address with a UUID, and load balance over the physical addresses. We could open multiple sockets, and associate each (receiver) socket's physical address with the UUID. We could even change this at runtime: e.g. if a NIC is down and we get exceptions on the socket, simply create another socket, remove the old association across the cluster (there's a call for this) and associate the new physical address with the UUID.
Another possibilty is to implement NATting, firewalling or STUNning this way !
I'll probably make the picking of a physical address given a UUID pluggable, so developers can even provide their own address translation in the transport !
This change is overdue and I'm happy that work has finally started on this. If you want to follow this, the branch is Branch_JGroups_2_8_LogicalAddresses.
For those of you who've used JGroups, you'll know that the identity of a node was always its IP address and the port on which the receiver thread was listening, e.g. 192.168.1.5:7800.
While this gives you a relatively compact and readable address (you can deduct from the address on which host it resides), there's also a problem: this type of address is not unique over space and time.
Let's look at an example.
Say we have a cluster of {A,B,C}. C's address is 192.168.1.5:7800. Let's assume A has sent 25 messages to C and C has multicast 104 messages. We're using sequence numbers (seqnos) to order messages, attached to a message via a header.
So the next message that C will multicast is #105 and the next message it expects from A is #26.
This is state that is maintained by the respective protocols in a JGroups stack.
Now let's assume C is killed and restarted. Or C is shunned, therefore leaves the channel and then automatically (if configured) reconnects. Let's also assume that the failure detection protocol has not yet kicked in and therefore A and B will not have received a view {A,B} which excludes C.
Now C rejoins the cluster. Because this is a reincarnation of C, it creates a new protocol stack, and all the state mentioned above is gone. The reincarnated C now sends #1 as next seqno and expects #1 from A as well.
There are 2 things that happen now:
- When C multicasts its next message with seqno #1, both A and B will drop it. A drops it because it expects C's next message to be #105, not #1. As a matter of fact A will drop the first 104 messages from C !
- A multicasts a message with seqno #26. However, C expects #1 from A and therefore buffers message #26. As a matter of fact, C will buffer all messages from A until it receives #1 which will not happen ! Consequence: C will run out of memory at some point. Even worse: C will prevent stability messages from purging messages seen by all cluster nodes, so in the worst case, all cluster nodes will run out of memory !
So how are logical address going to change this ?
A logical address consists of
- an org.jgroups.util.UUID (copied from java.util.UUID and relieved of some useless fields) and
- a logical name
JChannel channel=new JChannel("node-4", "/home/bela/udp.xml");
This means that the channel's address will always get printed as "node-4". Under the cover, however, we use a UUID (for equals() and hashCode()), which is unique over space and time. The UUID is recreated on channel connect, so the above reincarnation issue will not happen.
The logical name is syntactic sugar, because if we have views consisting of UUIDs (16 bytes), that's not a pretty sight, so views like {"node-1", "node-2", "node-3", "node-4"} look much better.
Note that the user will be able to pick whether to see UUIDs or logical names.
Also, if null is passed as logical name, JGroups will create a logical name (e.g. using the host name and a counter).
A UUID will get mapped to one or more physical addresses. The mapping is maintained by the transport and there will be an ARP-like protocol (handled by Discovery) to fetch the initial mappings, and to fetch a mapping if not available.
The detailed design is described in http://javagroups.cvs.sourceforge.net/viewvc/javagroups/JGroups/doc/design/LogicalAddresses.txt?revision=1.12&view=markup.
So the most important aspect of logical addresses is that they decouple the identity of a JGroups node from its physical address.
This opens up interesting possibilities.
We might for example associate multiple physical address with a UUID, and load balance over the physical addresses. We could open multiple sockets, and associate each (receiver) socket's physical address with the UUID. We could even change this at runtime: e.g. if a NIC is down and we get exceptions on the socket, simply create another socket, remove the old association across the cluster (there's a call for this) and associate the new physical address with the UUID.
Another possibilty is to implement NATting, firewalling or STUNning this way !
I'll probably make the picking of a physical address given a UUID pluggable, so developers can even provide their own address translation in the transport !
This change is overdue and I'm happy that work has finally started on this. If you want to follow this, the branch is Branch_JGroups_2_8_LogicalAddresses.
Wednesday, January 21, 2009
ReplCache: storing your data in the cloud with variable replication
Some time ago, I wrote a prototype of a cache which distributes its elements (key-value pairs) across all cluster nodes. This is done by computing the consistent hash of a key K and picking a cluster node based on the hash mod N where N is the cluster size. So any given element will only ever be stored once in the cluster.
This is great because it maximizes use of the aggregated memory of the 'cloud' (a.k.a. all cluster nodes). For example, if we have 10 nodes, and each node has 1 GB of memory, then the aggregated (cloud) memory is 10 GB. This is similar to a logical volume manager (e.g. LVM in Linux), where we 'see' a virtual volume, the size of which can grow or shrink, and which hides the mapping to physical disks.
So, if we pick a good consistent hash algorithm, then for 1'000 elements, we can assume that in a cluster of 10 nodes, each node stores on average 100 elements. Also, with consistent hashing, if you pick a good hash algorithm, rehashing on view changes is minimal.
Now, the question is what we do when a node crashes. All elements stored by that node are gone, and have to be re-read from somewhere, usually a database.
To provide highly available data and minimize access to the database, a common technique is to replicate data. For example, if we replicate K to all 10 nodes, then we can tolerate 9 nodes going down and will still have K available.
However, this comes at a cost: if everyone replicates all of its elements to all cluster nodes, then we can effectively only use 1/N of the 'cloud memory' (10 GB), which is 1 GB... So we trade access to the large cloud memory for availability.
This is like RAID: if we have 2 disks of 500 GB each, then we can use them as RAID 0 or JBOD (Just a Bunch of Disks) and have 1 TB available for our data. If one of the disks crashes, we lose data that resides on that disk. If we happen to have a file F with 10 blocks, and 5 were stored on the crashed disk, then F is gone.
If we use RAID 1, then the contents of disk-1 are mirrored onto disk-2 and vice versa. This is great, because we can now lose 1 disk and still have all of our data available. However, we now have only 500 MB of disk space available for our data !
Enter ReplCache. This is a prototype I've been working on for the last 2 weeks.
ReplCache allows for variable replication, which means we can tell it on a put(key, value, K) how many copies (replication count) of that element should be stored in the cloud. A replication count K can be:
ReplCache is a superset of PartitionedHashMap, which means it can be used as a PartitionedHashMap: just use K == 1 for all elements to be inserted !
The more important feature, however, is that ReplCache can use more of the available cloud memory and that it allows a user to define availability as a quality of service per data element ! Data that can be re-read from the DB can be stored with K == 1. Data that should be highly available should use K == -1, and data which should be more or less highly available, but can still be read from the DB (but maybe that's costly), should be stored with K > 1.
Compare this to RAID: once we've configured RAID 1, then all data written to disk-1 will always be mirrored to disk-2, even data that could be trashed on a crash, for example data in /tmp.
With ReplCache, the user (who knows his/her data best) takes control and defines QoS for each element !
Below is a screenshot of 2 ReplCache instances (started with java org.jgroups.demos.ReplCacheDemo -props /home/bela/udp.xml) which shows that we've added some data:
It shows that both instance have key "everywhere" because it is replicated to all cluster nodes due to K == -1. The same goes for key "two": because K == 2, it is stored on both instances as we only have 2 cluster nodes.
There are 2 keys with K == 1: "id" and "name". Both are stored on instance 2, but that's coincidence. For K keys and N cluster nodes, every node should store approximately K/N keys.
ReplCache is experimental, and serves as a prototype to play with data partitioning/striping for JBossCache.
ReplCache is in the JGroups CVS (head) and the code can be downloaded here. To run the demo, execute:
java -jar replcachedemo.jar
For the technical details, the design is here.
There is a nice 5 minute demo at http://www.jgroups.org/demos.html.
Feedback is appreciated, use the JGroups mailing lists !
Enjoy !
This is great because it maximizes use of the aggregated memory of the 'cloud' (a.k.a. all cluster nodes). For example, if we have 10 nodes, and each node has 1 GB of memory, then the aggregated (cloud) memory is 10 GB. This is similar to a logical volume manager (e.g. LVM in Linux), where we 'see' a virtual volume, the size of which can grow or shrink, and which hides the mapping to physical disks.
So, if we pick a good consistent hash algorithm, then for 1'000 elements, we can assume that in a cluster of 10 nodes, each node stores on average 100 elements. Also, with consistent hashing, if you pick a good hash algorithm, rehashing on view changes is minimal.
Now, the question is what we do when a node crashes. All elements stored by that node are gone, and have to be re-read from somewhere, usually a database.
To provide highly available data and minimize access to the database, a common technique is to replicate data. For example, if we replicate K to all 10 nodes, then we can tolerate 9 nodes going down and will still have K available.
However, this comes at a cost: if everyone replicates all of its elements to all cluster nodes, then we can effectively only use 1/N of the 'cloud memory' (10 GB), which is 1 GB... So we trade access to the large cloud memory for availability.
This is like RAID: if we have 2 disks of 500 GB each, then we can use them as RAID 0 or JBOD (Just a Bunch of Disks) and have 1 TB available for our data. If one of the disks crashes, we lose data that resides on that disk. If we happen to have a file F with 10 blocks, and 5 were stored on the crashed disk, then F is gone.
If we use RAID 1, then the contents of disk-1 are mirrored onto disk-2 and vice versa. This is great, because we can now lose 1 disk and still have all of our data available. However, we now have only 500 MB of disk space available for our data !
Enter ReplCache. This is a prototype I've been working on for the last 2 weeks.
ReplCache allows for variable replication, which means we can tell it on a put(key, value, K) how many copies (replication count) of that element should be stored in the cloud. A replication count K can be:
- K == 1: the element is stored only once. This is the same as what PartitionedHashMap does
- K == -1: the element is stored on all nodes in the cluster
- K == > 1: the element is stored on K nodes only. ReplCache makes sure to always have K instances of an element available, and if K drops because a node leaves or crashes, ReplCache might copy or move the element to bring K back up to the original value
ReplCache is a superset of PartitionedHashMap, which means it can be used as a PartitionedHashMap: just use K == 1 for all elements to be inserted !
The more important feature, however, is that ReplCache can use more of the available cloud memory and that it allows a user to define availability as a quality of service per data element ! Data that can be re-read from the DB can be stored with K == 1. Data that should be highly available should use K == -1, and data which should be more or less highly available, but can still be read from the DB (but maybe that's costly), should be stored with K > 1.
Compare this to RAID: once we've configured RAID 1, then all data written to disk-1 will always be mirrored to disk-2, even data that could be trashed on a crash, for example data in /tmp.
With ReplCache, the user (who knows his/her data best) takes control and defines QoS for each element !
Below is a screenshot of 2 ReplCache instances (started with java org.jgroups.demos.ReplCacheDemo -props /home/bela/udp.xml) which shows that we've added some data:
It shows that both instance have key "everywhere" because it is replicated to all cluster nodes due to K == -1. The same goes for key "two": because K == 2, it is stored on both instances as we only have 2 cluster nodes.
There are 2 keys with K == 1: "id" and "name". Both are stored on instance 2, but that's coincidence. For K keys and N cluster nodes, every node should store approximately K/N keys.
ReplCache is experimental, and serves as a prototype to play with data partitioning/striping for JBossCache.
ReplCache is in the JGroups CVS (head) and the code can be downloaded here. To run the demo, execute:
java -jar replcachedemo.jar
For the technical details, the design is here.
There is a nice 5 minute demo at http://www.jgroups.org/demos.html.
Feedback is appreciated, use the JGroups mailing lists !
Enjoy !
Monday, January 05, 2009
JGroups 2.7 released
Finally, after almost a year of development, I released 2.7.0.GA this morning. It can be downloaded from http://sourceforge.net/project/showfiles.php?group_id=6081&package_id=94868&release_id=651542.
Although 2.7 has 211 JIRA issues (bugfixes, tasks or features), most of the bugs have been back ported to the 2.6 branch. Why ? Because 2.6.7 is the version that ships with JBoss 5, and we made sure JGroups works optimally with it.
So what's new ?
There are almost no new features ! (Can you tell I'm not a marketing person ? :-))
Most work (besides bug fixes) went into refactoring, e.g. we converted our test suite from JUnit to TestNG, allowing for parallel test execution and thus reduced our testing time from 2.5 hours to 15 minutes !
Another change was that all properties are now set using JSR 175 annotations, so we could remove a lot of boilerplate code from protocol implementations. In my opinion, the more code I can remove (without impacting functionality), the better !
Using annotations for properties also allows us to automatically generate documentation for the properties of all protocols.
I also marked unsupported or experimental classes/methods with @Unsupported or @Experimental annotations.
We were able to increase performance a bit, compared to 2.6, but 2.6 is already quite fast, so unless you need those additional 5-10%, go for 2.6.7.
In a nutshell, 2.7 serves as the groundwork for 2.8, which will have many new features.
Although 2.7 has 211 JIRA issues (bugfixes, tasks or features), most of the bugs have been back ported to the 2.6 branch. Why ? Because 2.6.7 is the version that ships with JBoss 5, and we made sure JGroups works optimally with it.
So what's new ?
There are almost no new features ! (Can you tell I'm not a marketing person ? :-))
Most work (besides bug fixes) went into refactoring, e.g. we converted our test suite from JUnit to TestNG, allowing for parallel test execution and thus reduced our testing time from 2.5 hours to 15 minutes !
Another change was that all properties are now set using JSR 175 annotations, so we could remove a lot of boilerplate code from protocol implementations. In my opinion, the more code I can remove (without impacting functionality), the better !
Using annotations for properties also allows us to automatically generate documentation for the properties of all protocols.
I also marked unsupported or experimental classes/methods with @Unsupported or @Experimental annotations.
We were able to increase performance a bit, compared to 2.6, but 2.6 is already quite fast, so unless you need those additional 5-10%, go for 2.6.7.
In a nutshell, 2.7 serves as the groundwork for 2.8, which will have many new features.
Tuesday, December 09, 2008
Better late than never
This is my new blog.
I know, I said that in 2005 already. This time, though, I'm serious.
I'll write mostly about JGroups, but when I have thoughts on different topics I'll write them down, too.
Take my postings with a grain of salt and remember, these are my presonal thoughts and not necessarily those of my employer, JBoss / RedHat.
Bela Ban
I know, I said that in 2005 already. This time, though, I'm serious.
I'll write mostly about JGroups, but when I have thoughts on different topics I'll write them down, too.
Take my postings with a grain of salt and remember, these are my presonal thoughts and not necessarily those of my employer, JBoss / RedHat.
Bela Ban
Subscribe to:
Posts (Atom)