message propagation in a network, p2p approach, part 3

Me and a friend sat together and discussed this torrent stuff. It was lovely to get more input. Hopefully i was not pushing him too much about this topic. :-)

We have discussed mainly:

• How to represent an endless stream

Yeah, how?

Lets say, i install a webcam somewhere. Then there is a station, which captures the video stream from the cam and it writes the data into a file, a bytestream always growing in size.

This station creates metadata about this stream, creates a unique id, and publishes it. Other peers pick it up and leech the data.

But torrent files are considered to be static files. They sum up an existing source. It does not work on a stream, which is supposed to grow in size in the future.

Then we came up with the term of "linked torrents".

We assume, that a station knows for sure about the past and it can make statements about the future.

The station creates a bytestream with the length of, lets say, 100MB, which would fit for webcam data of about 10 minutes. It fills it continuously with the data from the webcam. If the stream is full or a timeout arrives (because the cam did not send any data anymore), then the torrent is ready to be completed. It describes the stream and it can be published.

Additionally the station will already add the ID of the next bytestream as well. It links itself to the next torrent. This identifies the torrent for the future:

• torrent[0] = "torrent describing the whole stream project"
• prev(torrent[0]) = "" | "meta torrent of whatever, universe, company, etc. etc. reflecting a status at the present time"
• prev(torrent[n+1] = torrent[n]
• next(torrent[0]) = torrent[1]
• next(torrent[n]) = torrent[n+1]

Each future torrent might quote the content of the previous torrents as well or just their IDs or just a selection of IDS. In any way, the torrent links itself to at least one previous torrent, if its not the starting torrent.

Since the station is the publisher, it can also make statements about the future, when the next torrent might become available. It could say torrent(0) arrives at a time X, torrent(1) arrives at a time Y. Without warranty, of course. This could help indexing the torrents. And for clients its helpful.

So, a client in the future, who watches a live stream of the present time, might be able to rewind 10, 100 years ago. As long as there are peers available, who have stored the data. (Archiver peers, why not, like a national library)

message propagation in a network, p2p approach, part 2

The more i write or think about it, the more i know, what i would like to have:

• There is a bytestream out there. This bytestream is identified by a unique hash.
  
• Peers are ready to seed this bytestream to other peers.
  
• Other peers have a way to find out, what hashes are available, hashes of their interest (which would group peers together)
  
• Other peers have a way to find out, which peers seed the bytestream identified by the hash or which peers have parts of it
  

Lets transform it to the message propagation example:

• A peer has a message queue as a resource. It writes the messages into the queue. The queue itself has an underlying bytestream, which itself could be a simple file it writes into.
  
  
• The peer writes properties about this resource into a file with a certain format and announces it somewhere (the torrent/tracker equivalent).
  
  
• Another peer knows where to look for "torrents" and finds the ones of its own interest. Then it asks someone else, who offers this resource at the moment and starts to leech from it (or lets say: it replicates that resource)
  
  
• On that replicated resource (which can be a file again) sits a bytestream, which is again the source for the message queue, which would extracts the messages from it.
  

Checking, yes, it still sounds understandable to me.

No thoughts about recovery are needed yet!

If a peer crashes, its bytestream can no more be continued, sure. But other peer might have replicated this resource right before its crash, so other peers can still leech it. There might be archiver peers in the network which do nothing else than replicating certain resources. Peers might chose as well to use only the archiver peers to leech from.

And if a crashed peer comes back, it also can recover itself by leeching lost pieces from the other peers and then starts continuing the resource on its own again. Or it creates a new file and abandones the old file.

However, this still sounds a lot like bittorrent.


Okay, summing up shortly:

• PeerA Message Queue writes messages into bytestream BS_A (business logic)
  
• PeerB Message Queue writes messages into bytestream BS_B (business logic)
  
• PeerA replicates BS_B (replication logic)
  
• PeerB replicates BS_A (replication logic)
  
• PeerB Message Queue reads messages from PeerA by reading BS_A (business logic)
  
• PeerA Message Queue reads messages from PeerB by reading BS_B (business logic)
  

I love it already.

There is one missing piece yet. How to represent a bytestream, which has no start and no end. This will be in the next blog entry. I hope.

message propagation in a network, p2p approach

I have several nodes in a network. They find each other by certain mechanisms (multicast, broadcast, ...) And then they deliver messages to each other. I have seen this so by a JXTA implementation and JGroups.

But i have problems to understand the JXTA protocol and i can't see through JGroups. I blame this on me, but its way to complicate for me to understand and to debug both. No bad feelings.

I do not like static factories (in the JXTA reference implementation you can only have one "world" within a jvm, you can not do something like:

jxtaNode1 = new JXTA();
jxtaNode2 = new JXTA();

Good Bye, easy JUnit tests!

And in JGroups i also run into trouble when i want to simulate two nodes in one junit test, fast shutdown/startups within a test will usually fail. Probably it has something to do with the nodeIds, which dont really change and therefore nothing is "from scratch" and recovery is being started, etc etc.

And multicast is always troublesome for me. But thats what the JXTA reference implementation and JGroups rely on, its their defaults, even if it works without multicast.

Nevermind, i am a nobody and i simply enjoy fast learning curves for myself. I am impatient, i prefer more the brain dead stuff, which looks simple, but still results from complex thoughts, when you dig deeper into it. Both JXTA and JGroups are complex, but not "brain dead". Wow, thats a lame bashing. :-)

I like the bittorrent approach, tho:

• There is a resource out there. This resource is identified by a unique hash.
  
  
• A bunch of peers are ready to seed this resource to other peers.
  
  
• Those other peers have found a way to get the hash to a resource.
  
  
• The other peers have found a way to find the seeding peers and can then start to leech/download the resource.
  
  
• The other peers have found a way to find other peers, who have already leeched/downloaded parts of the resource.
  
  
• The more peers have downloaded the resource completely (or peers have pieces only, but all pieces make the puzzle complete again), the more the seeder loses importance.
  

Anyhow, I find this very easy to understand.

But what bittorrent can not offer to me is a resource, which has no start and no end. It all relies on a resource, which does not change.

However, you can fake it, by creating continuously new resources and publishing those hashes.

message propagation in a network, thoughts

Before this article i introduced you to the classical client/server approach.

There are several issues:

1. nodes have to find their hub.

2. the hub itself can crash, no message can be delivered

3. you need to put on hold all nodes, when the hub is being maintained or changes its location

4. since all nodes connect to the hub, the hub might not scale very well.

5. when does a hub know, that a node is not responsive anymore. and what to do? discarding messages? keeping them for a certain time? in any way, you need a protocol for that

and so on. this makes this approach not very attractive.

message propagation in a network, classic client/server

There is an application, which consists of many nodes connected to one hub. When a node sends something, it does so by sending a message to the hub. The hub itself will propagate this message to all other nodes.

A node knows 2 queues, one for incoming and one for outgoing messages. An underlying library knows the same 2 queues. The node expects the library to send messages from the outgoing queue to the hub. And it expects the library to put messages from the hub into the incoming queue.

I like the fact, that the node does not have to interact with the library at all. It just starts the library within the runtime, passing those 2 queues and thats it.

// Bootstrap
Properties props = ...; // data where to connect to
Queue in = new MyQueue();
Queue out = new MyQueue();
ClusterNodeLib lib = new ClusterNodeLib(props,in,out);
App app = new App(lib);

--

The underlying library polls from the outgoing queue and writes the message on the socket. It also polls the socket for incoming messages to add them on the outgoing queue. And in a third way, the library ensures, that there is always a connection to the hub.

//Library Bootstrap
Runnable socketWriter = new Runnable() {
   public void run() {
       Object msg = out.peek();
       try {
          writeOnSocket(msg)
          out.poll();
       } catch (Throwable t) {
          t.printStackTrace();
       }
   }
}

Runnable socketReader = new Runnable() {
   public void run() {
       try {
          Object msg = readFromSocket();
          in.add(msg);
       } catch (Throwable t) {
          t.printStackTrace();
       }
   }
}

Runnable keepSocketAlive = new Runnable() {
   public void run() {
       try {
          if (isSocketBroken()) {
            reconnect();
          }
       } catch (Throwable t) {
          t.printStackTrace();
       }
   }
}

--

The used pattern in this example is a typical one. It is quite easy to create and understand. Java offers queues, which are thread-safe in a concurrent environment and for the socket status, you would need to use some thread-safe variables too. With keeping the runnables alive, for example with a ThreadPool, there is not much more left to do.

iscsi ramsan tests, the software

On all servers (nodes/master):

• Linux: 'Debian Etch', 64bit Installation (Kernel: 2.6.18-5-amd64) on all servers
  
• iscsitarget-0.4.16 for creating the targets on the nodes
  
• qla4xxx-5.01.00.08.tar.gz from Qlogic to drive the iscsi cards
• iozone3_308.tar to create the statistics

iscsi ramsan tests, the hardware

Nodes:

• 32GB RAM
• 2x CPU
• Dual-Port Copper Gigabit Ethernet Adapter (Intel Pro 1000 MT, PCI-X)

Master:

• 8GB RAM
• 2X CPU
• Dual-Port Copper iSCSI Network Adapter (QLogic QL4052, FW Version: 3.0.1.49)
  

Switch:

• 3com Superstack 3870, 24 Gigabit Ports, 9k jumbo frames support