Add tools for concurrent master proposal.

1 files changed, 95 insertions, 10 deletions

diff --git a/same/src/test/java/com/orbekk/paxos/PaxosServiceFunctionalTest.java b/same/src/test/java/com/orbekk/paxos/PaxosServiceFunctionalTest.java
index 32c7dff..b8d146c 100644
--- a/same/src/test/java/com/orbekk/paxos/PaxosServiceFunctionalTest.java
+++ b/same/src/test/java/com/orbekk/paxos/PaxosServiceFunctionalTest.java
@@ -1,22 +1,24 @@
 package com.orbekk.paxos;
 
-import static org.junit.Assert.*;
-
-import com.googlecode.jsonrpc4j.JsonRpcServer;
-import com.orbekk.same.ConnectionManagerImpl;
-import com.orbekk.same.http.RpcServlet;
-import com.orbekk.same.http.JettyServerBuilder;
-import com.orbekk.same.http.JettyServerContainer;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Random;
-import org.eclipse.jetty.server.Handler;
-import org.eclipse.jetty.server.Server;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
+import com.googlecode.jsonrpc4j.JsonRpcServer;
+import com.orbekk.same.ConnectionManagerImpl;
+import com.orbekk.same.http.JettyServerBuilder;
+import com.orbekk.same.http.JettyServerContainer;
+import com.orbekk.same.http.RpcServlet;
+
 public class PaxosServiceFunctionalTest {
     ConnectionManagerImpl connections = new ConnectionManagerImpl(500, 500);
     List<String> paxosUrls = new ArrayList<String>();
@@ -24,6 +26,17 @@ public class PaxosServiceFunctionalTest {
     String myUrl;
     int successfulProposals = 0;
     
+    Runnable sleepForever = new Runnable() {
+        @Override public synchronized void run() {
+            while (!Thread.interrupted()) {
+                try {
+                    wait();
+                } catch (InterruptedException e) {
+                }
+            }
+        }
+    };
+    
     @Before
     public void setUp() throws Exception {
         JettyServerBuilder builder = new JettyServerBuilder(0);
@@ -64,8 +77,80 @@ public class PaxosServiceFunctionalTest {
                 connections);
         assertTrue(m1.propose(1));
     }
+    
+    @Test
+    public void testMasterElectionTask() throws InterruptedException, ExecutionException {
+        MasterProposer m1 = new MasterProposer("http://client1", paxosUrls,
+                connections);
+        Future<Integer> result = m1.startProposalTask(1, null);
+        assertEquals(new Integer(1), result.get());
+    }
+    
+    @Test
+    public void cancelledElection() {
+        MasterProposer m1 = new MasterProposer("http://client1", paxosUrls,
+                connections);
+        assertTrue(m1.propose(1));
+
+        Future<Integer> result = m1.startProposalTask(1, sleepForever);
+        result.cancel(true);
+        assertTrue(result.isCancelled());
+    }
 
     @Test
+    public void testOnlyOneCompletes() throws InterruptedException, ExecutionException {
+        MasterProposer m1 = new MasterProposer("http://OnlyOneCompletes1", paxosUrls,
+                connections);
+        MasterProposer m2 = new MasterProposer("http://OnlyOneCompletes2", paxosUrls,
+                connections);
+        final Future<Integer> result1 = m1.startProposalTask(1, sleepForever);
+        final Future<Integer> result2 = m2.startProposalTask(1, sleepForever);
+        
+        Thread t1 = new Thread(new Runnable() {
+            @Override public void run() {
+                try {
+                    result1.get();
+                    result2.cancel(true);
+                } catch (CancellationException e) {
+                } catch (InterruptedException e) {
+                } catch (ExecutionException e) {
+                }
+            }
+        });
+        
+        Thread t2 = new Thread(new Runnable() {
+            @Override public void run() {
+                try {
+                    result2.get();
+                    result1.cancel(true);
+                } catch (CancellationException e) {
+                } catch (InterruptedException e) {
+                } catch (ExecutionException e) {
+                }
+            }
+        });
+        
+        t1.start();
+        t2.start();
+        try {
+            t1.join();
+        } catch (InterruptedException e) {
+        }
+        try {
+            t2.join();
+        } catch (InterruptedException e) {
+        }
+        
+        assertTrue(result1.isCancelled() || result2.isCancelled());
+        if (!result1.isCancelled()) {
+            assertEquals(new Integer(1), result1.get());
+        }
+        if (!result2.isCancelled()) {
+            assertEquals(new Integer(1), result2.get());
+        }
+    }
+    
+    @Test
     public void testWithCompetition() {
         int proposers = 5;
         List<Thread> masterProposers = new ArrayList<Thread>();