diff options
Diffstat (limited to 'same')
6 files changed, 71 insertions, 41 deletions
diff --git a/same/src/main/java/com/orbekk/same/ConnectionManager.java b/same/src/main/java/com/orbekk/same/ConnectionManager.java index 4637617..8a6d3a0 100644 --- a/same/src/main/java/com/orbekk/same/ConnectionManager.java +++ b/same/src/main/java/com/orbekk/same/ConnectionManager.java @@ -1,7 +1,6 @@ package com.orbekk.same; import com.orbekk.paxos.PaxosService; -import com.orbekk.same.discovery.DirectoryService; /** * An interface that returns a connection for a participant. @@ -12,5 +11,5 @@ public interface ConnectionManager { ClientService getClient(String url); MasterService getMaster(String url); PaxosService getPaxos(String url); - DirectoryService getDirectory(String url); + Services.Directory getDirectory(String location); } diff --git a/same/src/main/java/com/orbekk/same/ConnectionManagerImpl.java b/same/src/main/java/com/orbekk/same/ConnectionManagerImpl.java index ade469e..a57bbdf 100644 --- a/same/src/main/java/com/orbekk/same/ConnectionManagerImpl.java +++ b/same/src/main/java/com/orbekk/same/ConnectionManagerImpl.java @@ -1,24 +1,35 @@ package com.orbekk.same; +import java.io.IOException; import java.net.MalformedURLException; import java.net.URL; +import java.net.UnknownHostException; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.google.protobuf.RpcChannel; import com.googlecode.jsonrpc4j.ProxyUtil; import com.orbekk.net.MyJsonRpcHttpClient; import com.orbekk.paxos.PaxosService; -import com.orbekk.same.discovery.DirectoryService; +import com.orbekk.protobuf.NewRpcChannel; public class ConnectionManagerImpl implements ConnectionManager { private int connectionTimeout; private int readTimeout; private Map<String, MyJsonRpcHttpClient> connectionCache = new HashMap<String, MyJsonRpcHttpClient>(); - + private ConcurrentMap<String, Future<NewRpcChannel>> channels = + new ConcurrentHashMap<String, Future<NewRpcChannel>>(); + private Logger logger = LoggerFactory.getLogger(getClass()); /** @@ -52,6 +63,44 @@ public class ConnectionManagerImpl implements ConnectionManager { } return service; } + + private NewRpcChannel getChannel(String location) { + Future<NewRpcChannel> channel = channels.get(location); + if (channel == null) { + final String hostname = location.split(":")[0]; + final int port = Integer.valueOf(location.split(":")[1]); + Callable<NewRpcChannel> channelFactory = + new Callable<NewRpcChannel>() { + @Override public NewRpcChannel call() { + try { + return NewRpcChannel.create(hostname, port); + } catch (UnknownHostException e) { + logger.error("Could not connect: ", e); + return null; + } catch (IOException e) { + logger.error("Could not connect: ", e); + return null; + } + } + }; + FutureTask<NewRpcChannel> task = + new FutureTask<NewRpcChannel>(channelFactory); + channel = channels.putIfAbsent(location, task); + if (channel == null) { + task.run(); + channel = task; + } + } + try { + return channel.get(); + } catch (ExecutionException e) { + logger.error("Could not connect: ", e); + return null; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } + } @Override public ClientService getClient(String url) { @@ -69,7 +118,12 @@ public class ConnectionManagerImpl implements ConnectionManager { } @Override - public DirectoryService getDirectory(String url) { - return getClassProxy(url, DirectoryService.class); + public Services.Directory getDirectory(String location) { + RpcChannel channel = getChannel(location); + if (channel != null) { + return Services.Directory.newStub(channel); + } else { + return null; + } } } diff --git a/same/src/main/java/com/orbekk/same/SameController.java b/same/src/main/java/com/orbekk/same/SameController.java index 8c17807..855daa7 100644 --- a/same/src/main/java/com/orbekk/same/SameController.java +++ b/same/src/main/java/com/orbekk/same/SameController.java @@ -6,7 +6,6 @@ import org.slf4j.LoggerFactory; import com.orbekk.paxos.PaxosService; import com.orbekk.paxos.PaxosServiceImpl; import com.orbekk.same.config.Configuration; -import com.orbekk.same.discovery.DirectoryService; import com.orbekk.same.http.ServerContainer; import com.orbekk.same.http.StateServlet; import com.orbekk.same.http.JettyServerBuilder; @@ -143,13 +142,13 @@ public class SameController { return master; } - public DirectoryService getDirectory() { - String directoryUrl = configuration.get("directoryUrl"); - DirectoryService directory = null; - if (directoryUrl != null) { - directory = connections.getDirectory(directoryUrl); + public Services.Directory getDirectory() { + String directoryLocation = configuration.get("directoryLocation"); + if (directoryLocation != null) { + return connections.getDirectory(directoryLocation); + } else { + return null; } - return directory; } public VariableFactory createVariableFactory() { diff --git a/same/src/main/java/com/orbekk/same/TestConnectionManager.java b/same/src/main/java/com/orbekk/same/TestConnectionManager.java index 6f5e5d8..93b2e6a 100644 --- a/same/src/main/java/com/orbekk/same/TestConnectionManager.java +++ b/same/src/main/java/com/orbekk/same/TestConnectionManager.java @@ -4,7 +4,7 @@ import java.util.Map; import java.util.HashMap; import com.orbekk.paxos.PaxosService; -import com.orbekk.same.discovery.DirectoryService; +import com.orbekk.same.Services.Directory; /** * This class is used in test. @@ -16,8 +16,8 @@ public class TestConnectionManager implements ConnectionManager { new HashMap<String, MasterService>(); public Map<String, PaxosService> paxosMap = new HashMap<String, PaxosService>(); - public Map<String, DirectoryService> directoryMap = - new HashMap<String, DirectoryService>(); + public Map<String, Services.Directory> directoryMap = + new HashMap<String, Services.Directory>(); public TestConnectionManager() { } @@ -38,7 +38,7 @@ public class TestConnectionManager implements ConnectionManager { } @Override - public DirectoryService getDirectory(String url) { - return directoryMap.get(url); + public Directory getDirectory(String location) { + return directoryMap.get(location); } } 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 1fad871..6108630 100644 --- a/same/src/main/java/com/orbekk/same/benchmark/ClientBenchmark.java +++ b/same/src/main/java/com/orbekk/same/benchmark/ClientBenchmark.java @@ -81,7 +81,7 @@ public class ClientBenchmark { String host = args[0]; int port = Integer.valueOf(args[1]); try { - benchmark(host, port, 1000, 5000); + benchmark(host, port, 2000, 10000); } catch (InterruptedException e) { System.out.println("Benchmark failed."); } diff --git a/same/src/main/java/com/orbekk/same/discovery/DirectoryService.java b/same/src/main/java/com/orbekk/same/discovery/DirectoryService.java deleted file mode 100644 index fae2bf6..0000000 --- a/same/src/main/java/com/orbekk/same/discovery/DirectoryService.java +++ /dev/null @@ -1,22 +0,0 @@ -package com.orbekk.same.discovery; - -import java.util.List; - -/** - * Maintains a registry of available networks. - * - * The discovery service is only meant to be used for debugging. - */ -public interface DirectoryService { - /** - * Returns a list of network names and master urls interleaved, i.e., - * - * [NetworkName1, MasterUrl1, ...] - */ - List<String> getNetworks() throws Exception; - - /** - * Register a network. - */ - void registerNetwork(String networkName, String masterUrl) throws Exception; -} |