ft chat, doc beginning

chat/chat.rb

@@ -0,0 +1,102 @@
+require 'rubygems'
+require 'bud'
+
+module ChatProtocol
+  state do
+    channel :connect, [:@addr, :client] => [:nick]
+    channel :chatter
+  end
+
+  DEFAULT_ADDR = "127.0.0.1:12345"
+
+  def pretty_print(val)
+    str = "\033[34m"+val[1].to_s + ": " + "\033[31m" + (val[3].to_s || '') + "\033[0m"
+    pad = "(" + val[2].strftime("%I:%M.%S").to_s + ")"
+    return str + " "*[66 - str.length,2].max + pad
+  end
+end
+
+
+
+module ChatClient
+  include ChatProtocol
+
+  def initialize(nick=nil, server=DEFAULT_ADDR, opts={})
+    @nick = nick
+    @server = server
+    super opts
+  end
+
+  bootstrap do
+    connect <~ [[@server, ip_port, @nick]]
+    # record the fact that we too are a node.
+    # at first, we'll suspect that we're the leader.  soon, when the bootstrap @server forwards
+    # us its node list, we'll have a better idea of who the leader is.
+    nodelist <+ [[ip_port, @nick]]
+  end
+
+  bloom do
+    chatter <~ (stdio * leader).pairs do |s, l|
+      [l.addr, [ip_port, @nick, Time.new, s.line]]
+    end
+    stdio <~ chatter { |m| [pretty_print(m.val)] }
+  end
+end
+
+module ChatServer
+  include ChatProtocol
+
+  state { table :nodelist }
+
+  bloom do
+    nodelist <= connect{|c| [c.client, c.nick]}
+
+    # forward a message if a) we're the leader, and b) it hasn't already been relayed to us.
+    chatter <~ (chatter * nodelist * leader).combos do |m, n, l| 
+      if l.addr == ip_port and n.key != ip_port
+        [n.key, m.val]
+      end
+    end
+  end
+end
+
+
+module Members
+  state do
+    periodic :interval, 1
+    channel :heartbeat, [:@to, :from]
+    table :recently_seen, heartbeat.key_cols + [:rcv_time]
+    scratch :live_nodes, [:addr]
+    interface output, :leader, [:addr]
+  end
+
+  bloom do
+    heartbeat <~ (interval * nodelist).rights{|n| [n.key, ip_port]}
+    recently_seen <= heartbeat{|h| h.to_a + [Time.now.to_i]}
+    live_nodes <= recently_seen.group([:from], max(:rcv_time)) do |n|
+      [n.first] unless  (Time.now.to_i - n.last > 3)
+    end
+    leader <= live_nodes.group([], min(:addr))
+  end
+
+  bloom :dissem do
+    connect <~ (interval * nodelist * nodelist).combos do |h, n1, n2|
+      [n1.key, n2.key, n2.val]
+    end
+  end
+end
+
+class SingleChat
+  include Bud
+  include ChatClient
+  include ChatServer
+  include Members
+
+  bloom :eggy do
+    stdio <~ (chatter * leader).pairs do |m, l|
+      if m.val.last == "LEADER"
+        ["LEADER: #{l}"]
+      end
+    end
+  end
+end

chat/p2p-ft-chat.md

@@ -0,0 +1,47 @@
+=Peer to peer, fault-tolerant chat=
+
+Writing a simple chat server in Bloom is extremely <a href="">simple</a>.
+The job of a chat <a href="https://github.com/bloom-lang/bud/blob/master/examples/chat/chat.rb">client</a> is simply to forward messages (typed into the keyboard)to a central server, and to print (to the screen) messages relayed by that server.  The job of a <a href="https://github.com/bloom-lang/bud/blob/master/examples/chat/chat_server.rb>server</a> is to maintain a list of members, and forward all messages to all members.
+
+In this short demo, we'll evolve that toy chat server into a distributed system that is *decentralized* and *fault-tolerant*.  When we're done, we'll
+have a chat program that behaves essentially the same, except that 1) all nodes can play the role of client or server, and 2) when the server node fails,
+one of the clients can assume its role.
+
+==Tweaks==
+
+It turns out that the modifications to the original program are minimal, and fairly obvious.
+
+    mcast <~ stdio do |s|
+      [@server, [ip_port, @nick, Time.new.strftime("%I:%M.%S"), s.line]]
+    end
+
+In the original chat program, we forwarded all messages to a distinguished server running a different chunk of code.  In our p2p chat, any node could
+be the server.  So we replace the reference to an instance variable with a reference to a Bloom collection:
+
+    chatter <~ (stdio * leader).pairs do |s, l|
+      [l.addr, [ip_port, @nick, Time.new, s.line]]
+    end
+
+We'll leave the declaration of leader and the rules that define its contents for later.
+
+The original chat server blindly multicasted all messages it received to all clients:
+
+    mcast <~ (mcast * nodelist).pairs { |m,n| [n.key, m.val] }
+
+We need to make this multicast conditional on the server's belief that it is the current leader:
+
+    chatter <~ (chatter * nodelist * leader).combos do |m, n, l|
+      if l.addr == ip_port and n.key != ip_port
+        [n.key, m.val]
+      end
+    end
+
+A node wearing the server hat will relay a message if it believes it is the leader -- and will relay it to everyone except himself (after all,
+he has already received it).
+
+==The distributed systems part==
+
+Now comes the fun part.  How does a node know if it is the leader, and how does a non-leader know who the leader is?  How does this knowledge
+persist or change under delay and failure?  We need a notion of *group membership*, and some form of *leader election*.  These are tricky things
+to get right in general, but since in a chat application we require only best-effort behavior (if I send a message and no one sees it, I am OK with
+attempting to send it again) we can roll ourselves very simple versions of both.

chat/single.rb

@@ -0,0 +1,11 @@
+require './chat.rb'
+
+# a name, a server to attemp to connect to to bootstrap, and (optionally) a port to bind to.
+# for this 1-computer demo, let's save ourselves some typing and leave off the ip part: just ports.
+
+port = (ARGV.length == 3) ? ARGV[2] : Socket::INADDR_ANY
+
+program = SingleChat.new(ARGV[0], "127.0.0.1:#{ARGV[1]}", :stdin => $stdin, :port => port)
+program.run_fg
+
+