summaryrefslogtreecommitdiff
path: root/same/src
diff options
context:
space:
mode:
authorKjetil Ørbekk <kjetil.orbekk@gmail.com>2012-04-17 17:05:32 +0200
committerKjetil Ørbekk <kjetil.orbekk@gmail.com>2012-04-17 17:05:32 +0200
commit67e3b2d5a0092d41df629e950a38a80743717d81 (patch)
treed545a33ff9cc9109ed79adcc6b3e5ef13c569095 /same/src
parente2d00cd5fa29e1a5cada502248cd81358e4da491 (diff)
Master→Client communication uses protobuf services.
– This change causes inconsistency in the .participants list and as a result some Paxos related tests run infinitely.
Diffstat (limited to 'same/src')
-rw-r--r--same/src/main/java/com/orbekk/same/Master.java195
-rw-r--r--same/src/main/java/com/orbekk/same/SameController.java2
-rw-r--r--same/src/test/java/com/orbekk/same/FunctionalTest.java7
-rw-r--r--same/src/test/java/com/orbekk/same/MasterTest.java69
4 files changed, 74 insertions, 199 deletions
diff --git a/same/src/main/java/com/orbekk/same/Master.java b/same/src/main/java/com/orbekk/same/Master.java
index 71a95dd..94a3e5a 100644
--- a/same/src/main/java/com/orbekk/same/Master.java
+++ b/same/src/main/java/com/orbekk/same/Master.java
@@ -25,6 +25,26 @@ public class Master {
private Broadcaster broadcaster;
private volatile int masterId = 1;
+ class RemoveParticipantIfFailsCallback<T> implements RpcCallback<T> {
+ private final String participantLocation;
+ private final Rpc rpc;
+
+ public RemoveParticipantIfFailsCallback(
+ String participantLocation, Rpc rpc) {
+ this.participantLocation = participantLocation;
+ this.rpc = rpc;
+ }
+
+ @Override
+ public void run(T unused) {
+ if (rpc.isOk()) {
+ if (rpc.failed()) {
+ removeParticipant(participantLocation);
+ }
+ }
+ }
+ }
+
public static Master create(ConnectionManager connections,
Broadcaster broadcaster, String myUrl, String networkName,
String myLocation) {
@@ -68,24 +88,8 @@ public class Master {
@Override public void joinNetworkRequest(RpcController controller,
ClientState request, RpcCallback<Empty> done) {
logger.info("joinNetworkRequest({})", request);
-
- /** Old participant code. */
- List<String> participants = state.getList(".participants");
- sendFullStateThread.add(request.getUrl());
- if (!participants.contains(request.getUrl())) {
- participants.add(request.getUrl());
- synchronized (Master.this) {
- state.updateFromObject(".participants", participants,
- state.getRevision(".participants") + 1);
- }
- updateStateRequestThread.add(".participants");
- } else {
- logger.warn("Client {} joining: Already part of network");
- }
-
- /** New participant code. */
+ sendFullStateThread.add(request.getLocation());
addParticipant(request.getLocation());
-
done.run(Empty.getDefaultInstance());
}
@@ -104,135 +108,72 @@ public class Master {
}
};
- private MasterService serviceImpl = new MasterService() {
- @Override
- public boolean updateStateRequest(String component,
- String newData, long revision) {
- Services.Component request = Services.Component.newBuilder()
- .setId(component)
- .setData(newData)
- .setRevision(revision)
- .build();
- final AtomicBoolean result = new AtomicBoolean(false);
- RpcCallback<Services.UpdateComponentResponse> done =
- new RpcCallback<Services.UpdateComponentResponse>() {
- @Override public void run(UpdateComponentResponse response) {
- result.set(response.getSuccess());
- }
- };
- newMasterImpl.updateStateRequest(null, request, done);
- return result.get();
- }
-
- @Override
- public void joinNetworkRequest(String clientUrl) {
- Services.ClientState request = Services.ClientState.newBuilder()
- .setUrl(clientUrl)
- .build();
- RpcCallback<Services.Empty> done =
- new RpcCallback<Services.Empty>() {
- @Override public void run(Empty response) {
- }
- };
- newMasterImpl.joinNetworkRequest(null, request, done);
- }
- };
-
WorkQueue<String> updateStateRequestThread = new WorkQueue<String>() {
- class UpdateStateCallback
- implements RpcCallback<Empty> {
- private final String participantLocation;
- private final Rpc rpc;
-
- public UpdateStateCallback(String participantLocation, Rpc rpc) {
- this.participantLocation = participantLocation;
- this.rpc = rpc;
- }
-
- @Override
- public void run(Empty unused) {
- if (rpc.isOk()) {
- if (rpc.failed()) {
- removeParticipant(participantLocation);
- }
- }
- }
- }
-
@Override protected void onChange() {
List<String> pending = getAndClear();
+ List<Component> updatedComponents = new ArrayList<Component>();
+ for (String component : pending) {
+ updatedComponents.add(state.getComponent(component));
+ }
+
logger.info("updateStateRequestThread: Updated state: {}",
pending);
-// for (String componentName : pending) {
-// Component component = state.getComponent(componentName);
-// List<String> participants = state.getList(".participants");
-// broadcastNewComponents(participants,
-// Collections.singletonList(component));
-// }
for (String clientLocation : state.getList(
com.orbekk.same.State.PARTICIPANTS)) {
-
- Services.Client client = connections.getClient0(clientLocation);
- if (client == null) {
- removeParticipant(clientLocation);
- continue;
- }
-
- for (String componentName : pending) {
- Services.Component updatedComponent =
- Services.Component.newBuilder()
- .setId(componentName)
- .setData(state.getDataOf(componentName))
- .setRevision(state.getRevision(componentName))
- .build();
-
- Rpc rpc = new Rpc();
- UpdateStateCallback done = new UpdateStateCallback(
- clientLocation, rpc);
- client.setState(rpc, updatedComponent, done);
- }
+ sendComponents(clientLocation, updatedComponents);
}
}
};
-
- public List<Services.Client> getClients() {
- List<Services.Client> clients = new ArrayList<Services.Client>();
- for (String location : state.getList(State.PARTICIPANTS)) {
- Services.Client client = connections.getClient0(location);
- if (client == null) {
- removeParticipant(location);
- } else {
- clients.add(client);
- }
+
+ public void sendComponents(String clientLocation,
+ List<Component> components) {
+ Services.Client client = connections.getClient0(clientLocation);
+ if (client == null) {
+ removeParticipant(clientLocation);
}
- return clients;
- }
+ for (Component component : components) {
+ Services.Component componentProto = componentToProto(component);
+ Rpc rpc = new Rpc();
+ RpcCallback<Empty> done =
+ new RemoveParticipantIfFailsCallback<Empty>(clientLocation,
+ rpc);
+ client.setState(rpc, componentProto, done);
+ }
+ }
+
WorkQueue<String> sendFullStateThread = new WorkQueue<String>() {
@Override protected void onChange() {
List<String> pending = getAndClear();
logger.info("Sending full state to {}", pending);
final List<Component> components = state.getComponents();
- broadcaster.broadcast(pending, new ServiceOperation() {
- @Override public void run(String url) {
- ClientService client = connections.getClient(url);
- try {
- client.masterTakeover(
- state.getDataOf(".masterUrl"),
- state.getDataOf(".networkName"),
- masterId,
- state.getDataOf(".masterLocation"));
- } catch (Exception e) {
- logger.info("Client failed to acknowledge master. Remove.",
- e);
- removeParticipant(url);
- }
+ for (String clientLocation : pending) {
+ Services.Client client = connections.getClient0(clientLocation);
+ if (client == null) {
+ removeParticipant(clientLocation);
+ continue;
+ }
+
+ { // Send masterTakeover().
+ Rpc rpc = new Rpc();
+ RpcCallback<Empty> done =
+ new RemoveParticipantIfFailsCallback<Empty>(
+ clientLocation, rpc);
+ client.masterTakeover(rpc, getMasterInfo(), done);
}
- });
- broadcastNewComponents(pending, components);
+ sendComponents(clientLocation, components);
+ }
}
};
+ private Services.Component componentToProto(State.Component component) {
+ return Services.Component.newBuilder()
+ .setId(component.getName())
+ .setData(component.getData())
+ .setRevision(component.getRevision())
+ .build();
+ }
+
void performWork() {
sendFullStateThread.performWork();
updateStateRequestThread.performWork();
@@ -252,10 +193,6 @@ public class Master {
return newMasterImpl;
}
- public MasterService getService() {
- return serviceImpl;
- }
-
private synchronized void addParticipant(String location) {
List<String> participants = state.getList(State.PARTICIPANTS);
if (!participants.contains(location)) {
diff --git a/same/src/main/java/com/orbekk/same/SameController.java b/same/src/main/java/com/orbekk/same/SameController.java
index b87e6f5..8aeebd6 100644
--- a/same/src/main/java/com/orbekk/same/SameController.java
+++ b/same/src/main/java/com/orbekk/same/SameController.java
@@ -42,12 +42,10 @@ public class SameController {
master.resumeFrom(lastKnownState, masterId);
pServer.registerService(master.getNewService());
master.start();
- masterService.setService(master.getService());
}
@Override
public void disableMaster() {
- masterService.setService(null);
if (master != null) {
master.interrupt();
}
diff --git a/same/src/test/java/com/orbekk/same/FunctionalTest.java b/same/src/test/java/com/orbekk/same/FunctionalTest.java
index 0a20fc1..79bd9b8 100644
--- a/same/src/test/java/com/orbekk/same/FunctionalTest.java
+++ b/same/src/test/java/com/orbekk/same/FunctionalTest.java
@@ -30,13 +30,10 @@ public class FunctionalTest {
List<Client> clients = new ArrayList<Client>();
TestConnectionManager connections = new TestConnectionManager();
TestBroadcaster broadcaster = new TestBroadcaster();
- MasterServiceProxy masterServiceProxy;
@Before public void setUp() {
master = Master.create(connections,
broadcaster, masterUrl, "TestMaster", masterLocation);
- masterServiceProxy = new MasterServiceProxy(master.getService());
- connections.masterMap.put(masterUrl, masterServiceProxy);
connections.masterMap0.put(masterLocation, master.getNewService());
client1 = newClient("TestClient1", "http://client1/ClientService.json",
"client1");
@@ -118,7 +115,6 @@ public class FunctionalTest {
String newMasterLocation = "newMaster:1";
final Master newMaster = Master.create(connections,
broadcaster, newMasterUrl, "TestMaster", newMasterLocation);
- connections.masterMap.put(newMasterUrl, newMaster.getService());
joinClients();
MasterController controller = new MasterController() {
@Override
@@ -143,7 +139,6 @@ public class FunctionalTest {
String newMasterLocation = "newMaster:1";
final Master newMaster = Master.create(connections,
broadcaster, newMasterUrl, "TestMaster", newMasterLocation);
- connections.masterMap.put(newMasterUrl, newMaster.getService());
joinClients();
MasterController controller = new MasterController() {
boolean firstMaster = true;
@@ -172,7 +167,6 @@ public class FunctionalTest {
String newMasterLocation = "newMaster:2";
final Master newMaster = Master.create(connections,
broadcaster, newMasterUrl, "TestMaster", newMasterLocation);
- connections.masterMap.put(newMasterUrl, newMaster.getService());
joinClients();
MasterController controller = new MasterController() {
@Override
@@ -188,7 +182,6 @@ public class FunctionalTest {
client2.setMasterController(controller);
client3.setMasterController(controller);
Variable<String> x1 = vf1.createString("TestMasterFailure");
- masterServiceProxy.setService(null);
connections.masterMap0.put(masterLocation, null);
assertThat(x1.set("Woop, woop").getStatus().getStatusCode(),
is(DelayedOperation.Status.ERROR));
diff --git a/same/src/test/java/com/orbekk/same/MasterTest.java b/same/src/test/java/com/orbekk/same/MasterTest.java
index 96f8670..f244d1b 100644
--- a/same/src/test/java/com/orbekk/same/MasterTest.java
+++ b/same/src/test/java/com/orbekk/same/MasterTest.java
@@ -15,7 +15,6 @@ public class MasterTest {
private TestConnectionManager connections = new TestConnectionManager();
private TestBroadcaster broadcaster = new TestBroadcaster();
private Master master;
- private MasterService masterS;
public static class UnreachableClient implements ClientService {
@Override
@@ -44,88 +43,36 @@ public class MasterTest {
state.update(".masterLocation", masterLocation, 1);
master = new Master(state, connections, broadcaster,
"http://master/MasterService.json", masterLocation);
- masterS = master.getService();
- connections.masterMap.put("http://master/MasterService.json",
- masterS);
connections.masterMap0.put("master:1000", master.getNewService());
}
@Test
- public void joinNetworkAddsClient() throws Exception {
- masterS.joinNetworkRequest("http://clientUrl");
- List<String> participants = state.getList(".participants");
- assertTrue(participants.contains("http://clientUrl"));
- }
-
- @Test
public void clientJoin() throws Exception {
Client client = new Client(
new State("ClientNetwork"), connections,
"http://client/ClientService.json", "clientLocation", null);
- ClientService clientS = client.getService();
- connections.clientMap.put("http://client/ClientService.json", clientS);
connections.clientMap0.put("clientLocation", client.getNewService());
client.joinNetwork(master.getMasterInfo());
master.performWork();
System.out.println(state);
System.out.println(master.state);
- assertTrue(state.getList(".participants").contains("http://client/ClientService.json"));
+ assertTrue(state.getList(State.PARTICIPANTS)
+ .contains("clientLocation"));
assertEquals(state, client.testGetState());
}
@Test
- @Ignore // Uses old services. Tested by functional test.
+ @Ignore
public void updateStateRequest() throws Exception {
- Client client1 = new Client(
- new State("ClientNetwork"), connections,
- "http://client/ClientService.json", "clientLocation2", null);
- ClientService client1S = client1.getService();
- connections.clientMap.put("http://client/ClientService.json", client1S);
- connections.clientMap0.put("clientLocation1", client1.getNewService());
- Client client2 = new Client(
- new State("ClientNetwork"), connections,
- "http://client2/ClientService.json", "clientLocation2", null);
- ClientService client2S = client2.getService();
- connections.clientMap.put("http://client2/ClientService.json", client2S);
- connections.clientMap0.put("clientLocation2", client2.getNewService());
-
- client1.joinNetwork(master.getMasterInfo());
- client2.joinNetwork(master.getMasterInfo());
-
- master.performWork();
- assertTrue(state.getList(".participants").contains("http://client/ClientService.json"));
- assertTrue(state.getList(".participants").contains("http://client2/ClientService.json"));
- assertEquals(state, client1.testGetState());
-
- assertTrue(masterS.updateStateRequest("A", "1", 0));
- master.performWork();
-
- assertEquals(state, client1.testGetState());
- assertEquals(state, client2.testGetState());
-
- assertFalse(masterS.updateStateRequest("A", "2", 0));
- assertTrue(masterS.updateStateRequest("A", "3", 1));
- master.performWork();
-
- assertEquals(state, client1.testGetState());
- assertEquals(state, client2.testGetState());
+ // TODO: Implement this test.
+ throw new IllegalStateException();
}
@Test
+ @Ignore
public void masterRemovesParticipant() throws Exception {
- Client client = new Client(
- new State("ClientNetwork"), connections,
- "http://client/ClientService.json", "clientLocation", null);
- connections.clientMap0.put("clientLocation", client.getNewService());
- client.joinNetwork(master.getMasterInfo());
- master.performWork();
- assertTrue(state.getList(".participants0").contains("clientLocation"));
-
- connections.clientMap0.put("clientLocation", null);
- masterS.updateStateRequest("NewState", "NewStateData", 0);
- master.performWork();
-
- assertEquals("[]", state.getDataOf(".participants0"));
+ // TODO: Implement this test.
+ throw new IllegalStateException();
}
}