diff options
Diffstat (limited to 'same/src')
9 files changed, 57 insertions, 37 deletions
diff --git a/same/src/main/java/com/orbekk/paxos/MasterProposer.java b/same/src/main/java/com/orbekk/paxos/MasterProposer.java index 048995e..e2723a3 100644 --- a/same/src/main/java/com/orbekk/paxos/MasterProposer.java +++ b/same/src/main/java/com/orbekk/paxos/MasterProposer.java @@ -13,6 +13,7 @@ import org.slf4j.LoggerFactory; import com.google.protobuf.RpcCallback; import com.orbekk.protobuf.Rpc; import com.orbekk.same.ConnectionManager; +import com.orbekk.same.RpcFactory; import com.orbekk.same.Services; import com.orbekk.same.Services.ClientState; import com.orbekk.same.Services.PaxosRequest; @@ -23,12 +24,14 @@ public class MasterProposer extends Thread { private final ClientState client; private final List<String> paxosLocations; private final ConnectionManager connections; + private final RpcFactory rpcf; public MasterProposer(ClientState client, List<String> paxosLocations, - ConnectionManager connections) { + ConnectionManager connections, RpcFactory rpcf) { this.client = client; this.paxosLocations = paxosLocations; this.connections = connections; + this.rpcf = rpcf; } private class ResponseHandler implements RpcCallback<PaxosResponse> { @@ -87,7 +90,7 @@ public class MasterProposer extends Thread { ResponseHandler handler = new ResponseHandler(proposalNumber, paxosLocations.size()); for (String location : paxosLocations) { - Rpc rpc = new Rpc(); + Rpc rpc = rpcf.create(); Services.Paxos paxos = connections.getPaxos0(location); if (paxos == null) { handler.run(null); @@ -107,7 +110,7 @@ public class MasterProposer extends Thread { ResponseHandler handler = new ResponseHandler(proposalNumber, paxosLocations.size()); for (String location : paxosLocations) { - Rpc rpc = new Rpc(); + Rpc rpc = rpcf.create(); Services.Paxos paxos = connections.getPaxos0(location); PaxosRequest request = PaxosRequest.newBuilder() .setClient(client) diff --git a/same/src/main/java/com/orbekk/same/Client.java b/same/src/main/java/com/orbekk/same/Client.java index 6582e4a..27ce787 100644 --- a/same/src/main/java/com/orbekk/same/Client.java +++ b/same/src/main/java/com/orbekk/same/Client.java @@ -30,6 +30,7 @@ public class Client { private volatile MasterController masterController = null; private volatile Future<Integer> currentMasterProposal = null; private volatile MasterState masterInfo; + private final RpcFactory rpcf; private List<StateChangedListener> stateListeners = new ArrayList<StateChangedListener>(); @@ -62,7 +63,7 @@ public class Client { startMasterElection(); return op; } - final Rpc rpc = new Rpc(); + final Rpc rpc = rpcf.create(); RpcCallback<Services.UpdateComponentResponse> done = new RpcCallback<Services.UpdateComponentResponse>() { @Override @@ -160,11 +161,12 @@ public class Client { }; public Client(State state, ConnectionManager connections, - String myUrl, String myLocation) { + String myUrl, String myLocation, RpcFactory rpcf) { this.state = state; this.connections = connections; this.myUrl = myUrl; this.myLocation = myLocation; + this.rpcf = rpcf; } public void start() { @@ -212,7 +214,7 @@ public class Client { Services.Master master = connections.getMaster0(masterInfo.getMasterLocation()); - final Rpc rpc = new Rpc(); + final Rpc rpc = rpcf.create(); RpcCallback<Empty> done = new RpcCallback<Empty>() { @Override public void run(Empty unused) { if (!rpc.isOk()) { @@ -244,7 +246,7 @@ public class Client { List<String> paxosUrls = state.getList(State.PARTICIPANTS); paxosUrls.remove(failedMaster.getMasterLocation()); MasterProposer proposer = new MasterProposer(getClientState(), paxosUrls, - connections); + connections, rpcf); if (masterController == null) { logger.warn("Could not become master: No master controller."); return; @@ -297,7 +299,7 @@ public class Client { }; for (String location : participants) { - Rpc rpc = new Rpc(); + Rpc rpc = rpcf.create(); Services.Client client = connections.getClient0(location); if (client != null) { client.masterDown(rpc, failedMaster, done); diff --git a/same/src/main/java/com/orbekk/same/ConnectionManagerImpl.java b/same/src/main/java/com/orbekk/same/ConnectionManagerImpl.java index ac0abe0..0e199d6 100644 --- a/same/src/main/java/com/orbekk/same/ConnectionManagerImpl.java +++ b/same/src/main/java/com/orbekk/same/ConnectionManagerImpl.java @@ -16,7 +16,6 @@ import java.util.concurrent.FutureTask; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.googlecode.jsonrpc4j.ProxyUtil; import com.orbekk.paxos.PaxosService; import com.orbekk.protobuf.RpcChannel; diff --git a/same/src/main/java/com/orbekk/same/Master.java b/same/src/main/java/com/orbekk/same/Master.java index b97b4ab..397aa12 100644 --- a/same/src/main/java/com/orbekk/same/Master.java +++ b/same/src/main/java/com/orbekk/same/Master.java @@ -23,6 +23,7 @@ public class Master { private String myLocation; // Protobuf server location, i.e., myIp:port State state; private volatile int masterId = 1; + private final RpcFactory rpcf; class RemoveParticipantIfFailsCallback<T> implements RpcCallback<T> { private final String participantLocation; @@ -46,19 +47,20 @@ public class Master { public static Master create(ConnectionManager connections, String myUrl, String networkName, - String myLocation) { + String myLocation, RpcFactory rpcf) { State state = new State(networkName); state.update(".masterUrl", myUrl, 1); state.update(".masterLocation", myLocation, 1); - return new Master(state, connections, myUrl, myLocation); + return new Master(state, connections, myUrl, myLocation, rpcf); } Master(State initialState, ConnectionManager connections, - String myUrl, String myLocation) { + String myUrl, String myLocation, RpcFactory rpcf) { this.state = initialState; this.connections = connections; this.myUrl = myUrl; this.myLocation = myLocation; + this.rpcf = rpcf; } public String getNetworkName() { @@ -131,7 +133,7 @@ public class Master { for (Component component : components) { Services.Component componentProto = componentToProto(component); - Rpc rpc = new Rpc(); + Rpc rpc = rpcf.create(); RpcCallback<Empty> done = new RemoveParticipantIfFailsCallback<Empty>(clientLocation, rpc); @@ -152,7 +154,7 @@ public class Master { } { // Send masterTakeover(). - Rpc rpc = new Rpc(); + Rpc rpc = rpcf.create(); RpcCallback<Empty> done = new RemoveParticipantIfFailsCallback<Empty>( clientLocation, rpc); @@ -232,7 +234,7 @@ public class Master { for (final String location : state.getList(State.PARTICIPANTS)) { Services.Client client = connections.getClient0(location); - final Rpc rpc = new Rpc(); + final Rpc rpc = rpcf.create(); RpcCallback<Empty> done = new RpcCallback<Empty>() { @Override public void run(Empty unused) { if (!rpc.isOk()) { diff --git a/same/src/main/java/com/orbekk/same/SameController.java b/same/src/main/java/com/orbekk/same/SameController.java index c793948..9438c7b 100644 --- a/same/src/main/java/com/orbekk/same/SameController.java +++ b/same/src/main/java/com/orbekk/same/SameController.java @@ -17,6 +17,7 @@ public class SameController { private PaxosServiceImpl paxos; private Configuration configuration; private ConnectionManager connections; + private final RpcFactory rpcf; /** * Timeout for remote operations in milliseconds. @@ -31,7 +32,8 @@ public class SameController { String masterUrl = configuration.get("baseUrl") + "MasterService.json"; master = Master.create(connections, - masterUrl, configuration.get("networkName"), myLocation); + masterUrl, configuration.get("networkName"), myLocation, + rpcf); master.resumeFrom(lastKnownState, masterId); pServer.registerService(master.getNewService()); master.start(); @@ -51,13 +53,15 @@ public class SameController { ConnectionManagerImpl connections = new ConnectionManagerImpl( timeout, timeout); + RpcFactory rpcf = new RpcFactory(timeout); + State clientState = new State(".InvalidClientNetwork"); String baseUrl = String.format("http://%s:%s/", configuration.get("localIp"), configuration.getInt("port")); String clientUrl = baseUrl + "ClientService.json"; Client client = new Client(clientState, connections, - clientUrl, myLocation); + clientUrl, myLocation, rpcf); PaxosServiceImpl paxos = new PaxosServiceImpl(""); SimpleProtobufServer pServer = SimpleProtobufServer.create(pport); @@ -66,7 +70,7 @@ public class SameController { SameController controller = new SameController( configuration, connections, client, - paxos, pServer); + paxos, pServer, rpcf); return controller; } @@ -75,12 +79,14 @@ public class SameController { ConnectionManager connections, Client client, PaxosServiceImpl paxos, - SimpleProtobufServer pServer) { + SimpleProtobufServer pServer, + RpcFactory rpcf) { this.configuration = configuration; this.connections = connections; this.client = client; this.paxos = paxos; this.pServer = pServer; + this.rpcf = rpcf; } public void start() throws Exception { @@ -139,7 +145,7 @@ public class SameController { .setNetworkName(master.getNetworkName()) .setMasterLocation(master.getLocation()) .build(); - final Rpc rpc = new Rpc(); + final Rpc rpc = rpcf.create(); RpcCallback<Services.Empty> done = new RpcCallback<Services.Empty>() { @Override public void run(Services.Empty unused) { if (!rpc.isOk()) { diff --git a/same/src/main/java/com/orbekk/same/benchmark/ClientBenchmark.java b/same/src/main/java/com/orbekk/same/benchmark/ClientBenchmark.java index 4e18416..861454c 100644 --- a/same/src/main/java/com/orbekk/same/benchmark/ClientBenchmark.java +++ b/same/src/main/java/com/orbekk/same/benchmark/ClientBenchmark.java @@ -5,23 +5,26 @@ import java.net.UnknownHostException; import java.util.concurrent.CountDownLatch; import com.google.protobuf.RpcCallback; -import com.orbekk.protobuf.RpcChannel; import com.orbekk.protobuf.Rpc; +import com.orbekk.protobuf.RpcChannel; +import com.orbekk.same.RpcFactory; import com.orbekk.same.benchmark.Example.Data; public class ClientBenchmark { private final Example.Service service; private final int warmupIterations; private final int iterations; + private final RpcFactory rpcf; public static void benchmark(String host, int port, int warmupIterations, int iterations) throws InterruptedException { RpcChannel channel = null; try { channel = RpcChannel.create(host, port); + RpcFactory rpcf = new RpcFactory(5000); Example.Service service = Example.Service.newStub(channel); ClientBenchmark benchmark = new ClientBenchmark( - service, warmupIterations, iterations); + service, warmupIterations, iterations, rpcf); benchmark.benchmark(); } catch (UnknownHostException e) { e.printStackTrace(); @@ -35,10 +38,11 @@ public class ClientBenchmark { } public ClientBenchmark(Example.Service service, - int warmupIterations, int iterations) { + int warmupIterations, int iterations, RpcFactory rpcf) { this.service = service; this.warmupIterations = warmupIterations; this.iterations = iterations; + this.rpcf = rpcf; } private void runBenchmark(int iterations) throws InterruptedException { @@ -48,7 +52,7 @@ public class ClientBenchmark { for (int i = 0; i < iterations; i++) { Example.Data request = Example.Data.newBuilder() .setArg1(i).build(); - Rpc rpc = new Rpc(); + Rpc rpc = rpcf.create(); service.methodA(rpc, request, new RpcCallback<Example.Data>() { @Override public void run(Data ignored) { diff --git a/same/src/test/java/com/orbekk/paxos/PaxosServiceFunctionalTest.java b/same/src/test/java/com/orbekk/paxos/PaxosServiceFunctionalTest.java index 98631b0..2918f3e 100644 --- a/same/src/test/java/com/orbekk/paxos/PaxosServiceFunctionalTest.java +++ b/same/src/test/java/com/orbekk/paxos/PaxosServiceFunctionalTest.java @@ -15,12 +15,14 @@ import org.junit.Test; import com.orbekk.protobuf.SimpleProtobufServer; import com.orbekk.same.ConnectionManagerImpl; +import com.orbekk.same.RpcFactory; import com.orbekk.same.Services.ClientState; public class PaxosServiceFunctionalTest { ConnectionManagerImpl connections = new ConnectionManagerImpl(500, 500); List<String> paxosUrls = new ArrayList<String>(); List<SimpleProtobufServer> servers = new ArrayList<SimpleProtobufServer>(); + RpcFactory rpcf = new RpcFactory(5000); String myUrl; int successfulProposals = 0; ClientState client1 = ClientState.newBuilder() @@ -70,14 +72,14 @@ public class PaxosServiceFunctionalTest { @Test public void testMasterElection() throws InterruptedException { MasterProposer m1 = new MasterProposer(client1, paxosUrls, - connections); + connections, rpcf); assertTrue(m1.propose(1)); } @Test public void testMasterElectionTask() throws InterruptedException, ExecutionException { MasterProposer m1 = new MasterProposer(client1, paxosUrls, - connections); + connections, rpcf); Future<Integer> result = m1.startProposalTask(1, null); assertEquals(new Integer(1), result.get()); } @@ -85,7 +87,7 @@ public class PaxosServiceFunctionalTest { @Test public void cancelledElection() throws InterruptedException { MasterProposer m1 = new MasterProposer(client1, paxosUrls, - connections); + connections, rpcf); assertTrue(m1.propose(1)); Future<Integer> result = m1.startProposalTask(1, sleepForever); @@ -96,10 +98,10 @@ public class PaxosServiceFunctionalTest { @Test public void testOnlyOneCompletes() throws InterruptedException, ExecutionException { MasterProposer m1 = new MasterProposer(client1, paxosUrls, - connections); + connections, rpcf); ClientState client2 = ClientState.newBuilder().setLocation("client2").build(); MasterProposer m2 = new MasterProposer(client2, paxosUrls, - connections); + connections, rpcf); final Future<Integer> result1 = m1.startProposalTask(1, sleepForever); final Future<Integer> result2 = m2.startProposalTask(1, sleepForever); @@ -160,7 +162,7 @@ public class PaxosServiceFunctionalTest { .build(); MasterProposer proposer = new MasterProposer(client, paxosUrls, - connections); + connections, rpcf); try { if (proposer.proposeRetry(1)) { incrementSuccessfulProposals(); diff --git a/same/src/test/java/com/orbekk/same/FunctionalTest.java b/same/src/test/java/com/orbekk/same/FunctionalTest.java index c5d9a56..275eaac 100644 --- a/same/src/test/java/com/orbekk/same/FunctionalTest.java +++ b/same/src/test/java/com/orbekk/same/FunctionalTest.java @@ -29,10 +29,11 @@ public class FunctionalTest { VariableFactory vf3; List<Client> clients = new ArrayList<Client>(); TestConnectionManager connections = new TestConnectionManager(); + RpcFactory rpcf = new RpcFactory(5000); @Before public void setUp() { master = Master.create(connections, - masterUrl, "TestMaster", masterLocation); + masterUrl, "TestMaster", masterLocation, rpcf); connections.masterMap0.put(masterLocation, master.getNewService()); client1 = newClient("TestClient1", "http://client1/ClientService.json", "client1"); @@ -47,7 +48,7 @@ public class FunctionalTest { Client newClient(String clientName, String clientUrl, String location) { Client client = new Client(new State(clientName), connections, - clientUrl, location); + clientUrl, location, rpcf); connections.clientMap0.put(location, client.getNewService()); clients.add(client); String paxosUrl = clientUrl.replace("ClientService", "PaxosService"); @@ -112,7 +113,7 @@ public class FunctionalTest { String newMasterUrl = "http://newMaster/MasterService.json"; String newMasterLocation = "newMaster:1"; final Master newMaster = Master.create(connections, - newMasterUrl, "TestMaster", newMasterLocation); + newMasterUrl, "TestMaster", newMasterLocation, rpcf); joinClients(); MasterController controller = new MasterController() { @Override @@ -136,7 +137,7 @@ public class FunctionalTest { String newMasterUrl = "http://newMaster/MasterService.json"; String newMasterLocation = "newMaster:1"; final Master newMaster = Master.create(connections, - newMasterUrl, "TestMaster", newMasterLocation); + newMasterUrl, "TestMaster", newMasterLocation, rpcf); joinClients(); MasterController controller = new MasterController() { boolean firstMaster = true; @@ -164,7 +165,7 @@ public class FunctionalTest { String newMasterUrl = "http://newMaster/MasterService.json"; String newMasterLocation = "newMaster:2"; final Master newMaster = Master.create(connections, - newMasterUrl, "TestMaster", newMasterLocation); + newMasterUrl, "TestMaster", newMasterLocation, rpcf); joinClients(); MasterController controller = new MasterController() { @Override diff --git a/same/src/test/java/com/orbekk/same/MasterTest.java b/same/src/test/java/com/orbekk/same/MasterTest.java index 50b3302..f22e9e1 100644 --- a/same/src/test/java/com/orbekk/same/MasterTest.java +++ b/same/src/test/java/com/orbekk/same/MasterTest.java @@ -14,6 +14,7 @@ public class MasterTest { private State state = new State("TestNetwork"); private TestConnectionManager connections = new TestConnectionManager(); private Master master; + private RpcFactory rpcf = new RpcFactory(5000); @Before public void setUp() { @@ -21,7 +22,7 @@ public class MasterTest { state.update(".masterUrl", "http://master/MasterService.json", 1); state.update(".masterLocation", masterLocation, 1); master = new Master(state, connections, - "http://master/MasterService.json", masterLocation); + "http://master/MasterService.json", masterLocation, rpcf); connections.masterMap0.put("master:1000", master.getNewService()); } @@ -29,7 +30,7 @@ public class MasterTest { public void clientJoin() throws Exception { Client client = new Client( new State("ClientNetwork"), connections, - "http://client/ClientService.json", "clientLocation"); + "http://client/ClientService.json", "clientLocation", rpcf); connections.clientMap0.put("clientLocation", client.getNewService()); client.joinNetwork(master.getMasterInfo()); master.performWork(); |