Assign clientId on connection start

src/main/java/com/yahoo/pasc/paxos/client/ClientState.java

@@ -19,18 +19,21 @@
 import java.util.BitSet;
 
 import com.yahoo.pasc.ProcessState;
+import com.yahoo.pasc.paxos.messages.Hello;
 import com.yahoo.pasc.paxos.messages.Request;
 
 public class ClientState implements ProcessState {
     int clientId;
     int servers;
     int quorum;
     int connected;
+    int disconnected;
     int inlineThreshold;
     TimestampMessage asyncMessages [];
 
     long timestamp = -1;
     Request pendingRequest;
+    Hello pendingHello;
     BitSet acks = new BitSet();
 //    byte[] value;
     private int from;
@@ -77,6 +80,14 @@ public void setConnected(int connected) {
         this.connected = connected;
     }
 
+    public int getDisconnected() {
+        return disconnected;
+    }
+
+    public void setDisconnected(int disconnected) {
+        this.disconnected = disconnected;
+    }
+
     public long getTimestamp() {
         return timestamp;
     }
@@ -93,6 +104,14 @@ public void setPendingRequest(Request pendingRequest) {
         this.pendingRequest = pendingRequest;
     }
 
+    public Hello getPendingHello() {
+        return pendingHello;
+    }
+
+    public void setPendingHello(Hello pendingHello) {
+        this.pendingHello = pendingHello;
+    }
+
     public BitSet getAcks() {
         return acks;
     }

src/main/java/com/yahoo/pasc/paxos/client/PaxosClient.java

@@ -31,15 +31,19 @@
 
 import com.yahoo.pasc.PascRuntime;
 import com.yahoo.pasc.paxos.client.handlers.AsyncMessageHandler;
+import com.yahoo.pasc.paxos.client.handlers.ByeHandler;
 import com.yahoo.pasc.paxos.client.handlers.HelloHandler;
 import com.yahoo.pasc.paxos.client.handlers.ReplyHandler;
+import com.yahoo.pasc.paxos.client.handlers.ServerHelloHandler;
 import com.yahoo.pasc.paxos.client.handlers.SubmitHandler;
 import com.yahoo.pasc.paxos.client.handlers.TimeoutHandler;
 import com.yahoo.pasc.paxos.client.messages.Submit;
 import com.yahoo.pasc.paxos.client.messages.Timeout;
 import com.yahoo.pasc.paxos.messages.AsyncMessage;
+import com.yahoo.pasc.paxos.messages.Bye;
 import com.yahoo.pasc.paxos.messages.Hello;
 import com.yahoo.pasc.paxos.messages.Reply;
+import com.yahoo.pasc.paxos.messages.ServerHello;
 
 public class PaxosClient {
     public static void main(String[] args) throws Exception {
@@ -96,11 +100,13 @@ public static void main(String[] args) throws Exception {
 
             String [] serverHosts = host.split(",");
 
-            HelloHandler hello = new HelloHandler();
+            ServerHelloHandler shello = new ServerHelloHandler();
             ReplyHandler reply = new ReplyHandler();
             SubmitHandler submit = new SubmitHandler();
             TimeoutHandler tout = new TimeoutHandler();
             AsyncMessageHandler asyncm = new AsyncMessageHandler();
+            ByeHandler bye = new ByeHandler();
+            HelloHandler hello = new HelloHandler();
 
             Random rnd = new Random();
 
@@ -109,11 +115,13 @@ public static void main(String[] args) throws Exception {
                 ClientState clientState = new ClientState(clientId, servers, quorum, inlineThreshold, asynSize);
                 final PascRuntime<ClientState> runtime = new PascRuntime<ClientState>(protection);
                 runtime.setState(clientState);
-                runtime.addHandler(Hello.class, hello);
+                runtime.addHandler(ServerHello.class, shello);
                 runtime.addHandler(Reply.class, reply);
                 runtime.addHandler(Submit.class, submit);
                 runtime.addHandler(Timeout.class, tout);
                 runtime.addHandler(AsyncMessage.class, asyncm);
+                runtime.addHandler(Bye.class, bye);
+                runtime.addHandler(Hello.class, hello);
 
                 final PaxosClientHandler handler = new PaxosClientHandler(runtime, new SimpleClient(requestSize), 
                         serverHosts, clientId, clients, timeout, zkConnection, executor);

src/main/java/com/yahoo/pasc/paxos/client/PaxosClientHandler.java

@@ -23,6 +23,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Random;
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.concurrent.BlockingQueue;
@@ -32,7 +33,6 @@
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.Watcher.Event.EventType;
 import org.apache.zookeeper.ZooKeeper;
 import org.jboss.netty.bootstrap.ClientBootstrap;
 import org.jboss.netty.channel.Channel;
@@ -62,12 +62,14 @@
 import com.yahoo.pasc.paxos.messages.Hello;
 import com.yahoo.pasc.paxos.messages.InlineRequest;
 import com.yahoo.pasc.paxos.messages.Request;
+import com.yahoo.pasc.paxos.messages.ServerHello;
 import com.yahoo.pasc.paxos.messages.serialization.ManualDecoder;
 import com.yahoo.pasc.paxos.messages.serialization.ManualEncoder;
 
 public class PaxosClientHandler extends SimpleChannelUpstreamHandler implements PaxosInterface, Watcher {
 
     private static final Logger LOG = LoggerFactory.getLogger(PaxosClientHandler.class);
+    private static final int MAX_CLIENTS = 4096;
 
     private int clientId;
     private int clients;
@@ -107,6 +109,7 @@ public ChannelPipeline getPipeline() throws Exception {
         this.servers = servers;
         this.serverChannels = new Channel[servers.length];
         Arrays.fill(payload, (byte) 5);
+        generateClientId();
         for (int i = 0; i < servers.length; ++i) {
             tryConnect(i);
         }
@@ -171,8 +174,17 @@ public void start() throws KeeperException, InterruptedException {
 
     @Override
     public void process(WatchedEvent event) {
-        if (event.getType() != EventType.NodeChildrenChanged)
+        switch (event.getType()) {
+        case NodeChildrenChanged:
+        case None:
+            break;
+        default:
             return;
+        }
+
+        if (event.getState() != Watcher.Event.KeeperState.SyncConnected) {
+            return;
+        }
 
         try {
             leader = getLeadership(zk.getChildren(ELECTION_PATH, this));
@@ -230,12 +242,23 @@ synchronized void resubmitRequest(long timestamp) {
             timer.schedule(resubmit, timeout);
         }
     }
+
+    private Random random = new Random();
 
-    @Override
-    public synchronized void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
+    private void generateClientId() {
+        clientId = random.nextInt(MAX_CLIENTS);
+    }
+
+    private void sendHello(Channel c) {
         Hello hello = new Hello(clientId);
         hello.storeReplica(hello);
-        e.getChannel().write(hello);
+        runtime.handleMessage(hello);
+        c.write(hello);
+    }
+
+    @Override
+    public synchronized void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
+        sendHello(e.getChannel());
     }
 
     private long messagesReceived = 0;
@@ -245,9 +268,10 @@ public synchronized void channelConnected(ChannelHandlerContext ctx, ChannelStat
     @Override
     public synchronized void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
         Object message = e.getMessage();
-        if (message instanceof Hello) {
-            Hello hello = (Hello) e.getMessage();
-            serverChannels[hello.getClientId()] = e.getChannel();
+        if (message instanceof ServerHello) {
+            ServerHello hello = (ServerHello) e.getMessage();
+            serverChannels[hello.getServerId()] = e.getChannel();
+            LOG.warn("Received hello: " + message);
         }
         LOG.trace("Received {}", message);
         List<Message> messages = runtime.handleMessage((Message) message);
@@ -257,9 +281,15 @@ public synchronized void messageReceived(ChannelHandlerContext ctx, MessageEvent
             if (m instanceof Connected) {
                 lastTime = System.nanoTime();
 
-//                new Thread(new ThroughputMonitor()).start();
+                // new Thread(new ThroughputMonitor()).start();
 
                 clientInterface.connected();
+            } else if (m instanceof Reconnect) {
+                generateClientId();
+                for (Channel c : serverChannels) {
+                    if (c != null)
+                        sendHello(c);
+                }
             } else if (m instanceof Received) {
                 if (resubmit != null)
                     resubmit.cancel();
@@ -434,22 +464,23 @@ public void run() {
                 }
                 System.out.println(String.format("Throughput: %8.8f m/s", (messagesReceived - startMessagges)
                         / ((double) (System.currentTimeMillis() - startTime) / 1000)));
-                System.out.println(String.format("Latency: %4.4f ms", (totalTime / (double) latencyReceived) / 1000000));
+                System.out
+                        .println(String.format("Latency: %4.4f ms", (totalTime / (double) latencyReceived) / 1000000));
 
                 barrier.close();
-    
+
                 for (Channel c : serverChannels) {
                     if (c != null) {
                         c.close();
                     }
                 }
-    
+
                 try {
                     Thread.sleep(2000);
                 } catch (InterruptedException e) {
                     // ignore
                 }
-    
+
                 System.exit(0);
             } catch (KeeperException e) {
                 LOG.error("Couldn't measure throughput", e);

src/main/java/com/yahoo/pasc/paxos/client/Reconnect.java

@@ -0,0 +1,33 @@
+/**
+ * Copyright (c) 2011 Yahoo! Inc. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License. See accompanying LICENSE file.
+ */
+
+package com.yahoo.pasc.paxos.client;
+
+import com.yahoo.pasc.Message;
+
+public class Reconnect extends Message {
+
+    @Override
+    protected boolean verify() {
+        return true;
+    }
+
+    @Override
+    public void storeReplica(Message m) {
+
+    }
+
+}

src/main/java/com/yahoo/pasc/paxos/client/handlers/ByeHandler.java

@@ -0,0 +1,67 @@
+/**
+ * Copyright (c) 2011 Yahoo! Inc. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License. See accompanying LICENSE file.
+ */
+
+package com.yahoo.pasc.paxos.client.handlers;
+
+import java.util.Arrays;
+import java.util.List;
+
+import com.yahoo.pasc.Message;
+import com.yahoo.pasc.MessageHandler;
+import com.yahoo.pasc.paxos.client.ClientState;
+import com.yahoo.pasc.paxos.client.Reconnect;
+import com.yahoo.pasc.paxos.messages.Bye;
+import com.yahoo.pasc.paxos.messages.Hello;
+
+public class ByeHandler implements MessageHandler<Bye, ClientState, Reconnect> {
+
+    @Override
+    public boolean guardPredicate(Bye receivedMessage) {
+        return true;
+    }
+
+    @Override
+    public List<Reconnect> processMessage(Bye bye, ClientState state) {
+        List<Reconnect> descriptors = null;
+        if (!matches(bye, state)) {
+            return null;
+        }
+        int disconnected = state.getDisconnected();
+        disconnected++;
+        state.setDisconnected(disconnected);
+        int maxFaults = state.getServers() / 2 - 1;
+        if (disconnected == maxFaults) {
+            // Send the first message if connected to all servers
+            descriptors = Arrays.asList(new Reconnect());
+        }
+        return descriptors;
+    }
+
+    private boolean matches(Bye bye, ClientState state) {
+        Hello hello = state.getPendingHello();
+        if (hello == null)
+            return false;
+        return hello.getClientId() == bye.getClientId();
+    }
+
+    @Override
+    public List<Message> getOutputMessages(ClientState state, List<Reconnect> descriptors) {
+        if (descriptors != null && descriptors.size() > 0) {
+            return Arrays.<Message> asList(new Reconnect());
+        }
+        return null;
+    }
+}

src/main/java/com/yahoo/pasc/paxos/client/handlers/HelloHandler.java

@@ -22,34 +22,25 @@
 import com.yahoo.pasc.Message;
 import com.yahoo.pasc.MessageHandler;
 import com.yahoo.pasc.paxos.client.ClientState;
-import com.yahoo.pasc.paxos.client.Connected;
 import com.yahoo.pasc.paxos.messages.Hello;
 
-public class HelloHandler implements MessageHandler<Hello, ClientState, Connected> {
+public class HelloHandler implements MessageHandler<Hello, ClientState, Hello> {
 
     @Override
     public boolean guardPredicate(Hello receivedMessage) {
         return true;
     }
 
     @Override
-    public List<Connected> processMessage(Hello hello, ClientState state) {
-        List<Connected> descriptors = null;
-        int connected = state.getConnected();
-        connected++;
-        state.setConnected(connected);
-        if (connected == state.getServers()) {
-            // Send the first message if connected to all servers
-            descriptors = Arrays.asList(new Connected());
-        }
-        return descriptors;
+    public List<Hello> processMessage(Hello hello, ClientState state) {
+        state.setPendingHello(hello);
+        state.setConnected(0);
+        state.setDisconnected(0);
+        return Arrays.asList(hello);
     }
 
     @Override
-    public List<Message> getOutputMessages(ClientState state, List<Connected> descriptors) {
-        if (descriptors != null && descriptors.size() > 0) {
-            return Arrays.<Message> asList(new Connected());
-        }
-        return null;
+    public List<Message> getOutputMessages(ClientState state, List<Hello> messages) {
+        return Arrays.<Message>asList(messages.get(0));
     }
 }