You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@giraph.apache.org by ac...@apache.org on 2012/08/27 20:54:25 UTC
svn commit: r1377792 - in /giraph/trunk: ./
src/main/java/org/apache/giraph/bsp/ src/main/java/org/apache/giraph/comm/
src/main/java/org/apache/giraph/comm/messages/
src/main/java/org/apache/giraph/graph/ src/test/java/org/apache/giraph/comm/
Author: aching
Date: Mon Aug 27 18:54:23 2012
New Revision: 1377792
URL: http://svn.apache.org/viewvc?rev=1377792&view=rev
Log:
GIRAPH-313: Open Netty client and server on master. (majakabiljo via
aching)
Added:
giraph/trunk/src/main/java/org/apache/giraph/comm/MasterClient.java
giraph/trunk/src/main/java/org/apache/giraph/comm/MasterClientServer.java
giraph/trunk/src/main/java/org/apache/giraph/comm/MasterRequest.java
giraph/trunk/src/main/java/org/apache/giraph/comm/MasterRequestServerHandler.java
giraph/trunk/src/main/java/org/apache/giraph/comm/MasterServer.java
giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterClient.java
giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterClientServer.java
giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterServer.java
giraph/trunk/src/main/java/org/apache/giraph/comm/RequestType.java
giraph/trunk/src/main/java/org/apache/giraph/comm/WorkerRequest.java
giraph/trunk/src/main/java/org/apache/giraph/comm/WorkerRequestServerHandler.java
Removed:
giraph/trunk/src/main/java/org/apache/giraph/comm/RequestRegistry.java
Modified:
giraph/trunk/CHANGELOG
giraph/trunk/src/main/java/org/apache/giraph/bsp/CentralizedServiceWorker.java
giraph/trunk/src/main/java/org/apache/giraph/comm/NettyClient.java
giraph/trunk/src/main/java/org/apache/giraph/comm/NettyServer.java
giraph/trunk/src/main/java/org/apache/giraph/comm/NettyWorkerClient.java
giraph/trunk/src/main/java/org/apache/giraph/comm/NettyWorkerServer.java
giraph/trunk/src/main/java/org/apache/giraph/comm/RequestDecoder.java
giraph/trunk/src/main/java/org/apache/giraph/comm/RequestEncoder.java
giraph/trunk/src/main/java/org/apache/giraph/comm/RequestInfo.java
giraph/trunk/src/main/java/org/apache/giraph/comm/RequestServerHandler.java
giraph/trunk/src/main/java/org/apache/giraph/comm/SendPartitionMessagesRequest.java
giraph/trunk/src/main/java/org/apache/giraph/comm/SendPartitionMutationsRequest.java
giraph/trunk/src/main/java/org/apache/giraph/comm/SendVertexRequest.java
giraph/trunk/src/main/java/org/apache/giraph/comm/WritableRequest.java
giraph/trunk/src/main/java/org/apache/giraph/comm/messages/SendPartitionCurrentMessagesRequest.java
giraph/trunk/src/main/java/org/apache/giraph/graph/BspService.java
giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceMaster.java
giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceWorker.java
giraph/trunk/src/test/java/org/apache/giraph/comm/ConnectionTest.java
giraph/trunk/src/test/java/org/apache/giraph/comm/RequestFailureTest.java
giraph/trunk/src/test/java/org/apache/giraph/comm/RequestTest.java
Modified: giraph/trunk/CHANGELOG
URL: http://svn.apache.org/viewvc/giraph/trunk/CHANGELOG?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/CHANGELOG (original)
+++ giraph/trunk/CHANGELOG Mon Aug 27 18:54:23 2012
@@ -2,6 +2,9 @@ Giraph Change Log
Release 0.2.0 - unreleased
+ GIRAPH-313: Open Netty client and server on master. (majakabiljo via
+ aching)
+
GIRAPH-249: Move part of the graph out-of-core when memory is low
(apresta via aching).
Modified: giraph/trunk/src/main/java/org/apache/giraph/bsp/CentralizedServiceWorker.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/bsp/CentralizedServiceWorker.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/bsp/CentralizedServiceWorker.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/bsp/CentralizedServiceWorker.java Mon Aug 27 18:54:23 2012
@@ -176,6 +176,13 @@ public interface CentralizedServiceWorke
void assignMessagesToVertex(Vertex<I, V, E, M> vertex, Iterable<M> messages);
/**
+ * Get master info
+ *
+ * @return Master info
+ */
+ WorkerInfo getMasterInfo();
+
+ /**
* Get the GraphMapper that this service is using. Vertices need to know
* this.
*
Added: giraph/trunk/src/main/java/org/apache/giraph/comm/MasterClient.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/MasterClient.java?rev=1377792&view=auto
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/MasterClient.java (added)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/MasterClient.java Mon Aug 27 18:54:23 2012
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.giraph.comm;
+
+import org.apache.giraph.graph.WorkerInfo;
+
+/**
+ * Interface for master to send messages to workers
+ */
+public interface MasterClient {
+ /**
+ * Fix workers for current superstep
+ *
+ * @param workers Information about workers
+ */
+ void fixWorkerAddresses(Iterable<WorkerInfo> workers);
+
+ /**
+ * Flush all outgoing messages. This will synchronously ensure that all
+ * messages have been send and delivered prior to returning.
+ */
+ void flush();
+
+ /**
+ * Closes all connections.
+ */
+ void closeConnections();
+}
+
Added: giraph/trunk/src/main/java/org/apache/giraph/comm/MasterClientServer.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/MasterClientServer.java?rev=1377792&view=auto
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/MasterClientServer.java (added)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/MasterClientServer.java Mon Aug 27 18:54:23 2012
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.giraph.comm;
+
+/**
+ * Interface for both master client and master server
+ */
+public interface MasterClientServer extends MasterClient, MasterServer {
+}
Added: giraph/trunk/src/main/java/org/apache/giraph/comm/MasterRequest.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/MasterRequest.java?rev=1377792&view=auto
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/MasterRequest.java (added)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/MasterRequest.java Mon Aug 27 18:54:23 2012
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.giraph.comm;
+
+/**
+ * Interface for requests sent to master to extend
+ */
+public interface MasterRequest {
+ /**
+ * Execute the request
+ */
+ void doRequest();
+}
Added: giraph/trunk/src/main/java/org/apache/giraph/comm/MasterRequestServerHandler.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/MasterRequestServerHandler.java?rev=1377792&view=auto
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/MasterRequestServerHandler.java (added)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/MasterRequestServerHandler.java Mon Aug 27 18:54:23 2012
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.giraph.comm;
+
+import org.apache.hadoop.conf.Configuration;
+
+/** Handler for requests on master */
+public class MasterRequestServerHandler extends
+ RequestServerHandler<MasterRequest> {
+ /**
+ * Constructor
+ *
+ * @param workerRequestReservedMap Worker request reservation map
+ * @param conf Configuration
+ */
+ public MasterRequestServerHandler(
+ WorkerRequestReservedMap workerRequestReservedMap, Configuration conf) {
+ super(workerRequestReservedMap, conf);
+ }
+
+ @Override
+ public void processRequest(MasterRequest request) {
+ request.doRequest();
+ }
+
+ /**
+ * Factory for {@link MasterRequestServerHandler}
+ */
+ public static class Factory implements RequestServerHandlerFactory {
+ @Override
+ public RequestServerHandler newHandler(
+ WorkerRequestReservedMap workerRequestReservedMap,
+ Configuration conf) {
+ return new MasterRequestServerHandler(workerRequestReservedMap, conf);
+ }
+ }
+}
Added: giraph/trunk/src/main/java/org/apache/giraph/comm/MasterServer.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/MasterServer.java?rev=1377792&view=auto
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/MasterServer.java (added)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/MasterServer.java Mon Aug 27 18:54:23 2012
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.giraph.comm;
+
+import java.net.InetSocketAddress;
+
+/**
+ * Interface for master to receive messages from workers
+ */
+public interface MasterServer {
+ /**
+ * Get server address
+ *
+ * @return Address used by this server
+ */
+ InetSocketAddress getMyAddress();
+
+ /**
+ * Shuts down.
+ */
+ void close();
+}
Modified: giraph/trunk/src/main/java/org/apache/giraph/comm/NettyClient.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/NettyClient.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/NettyClient.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/NettyClient.java Mon Aug 27 18:54:23 2012
@@ -34,8 +34,6 @@ import java.util.concurrent.atomic.Atomi
import org.apache.giraph.graph.GiraphJob;
import org.apache.giraph.utils.TimedLogger;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.log4j.Logger;
import org.jboss.netty.bootstrap.ClientBootstrap;
@@ -51,16 +49,8 @@ import org.jboss.netty.handler.codec.fra
/**
* Netty client for sending requests.
- *
- * @param <I> Vertex id
- * @param <V> Vertex data
- * @param <E> Edge data
- * @param <M> Message data
*/
-@SuppressWarnings("rawtypes")
-public class NettyClient<I extends WritableComparable,
- V extends Writable, E extends Writable,
- M extends Writable> {
+public class NettyClient {
/** Do we have a limit on number of open requests we can have */
public static final String LIMIT_NUMBER_OF_OPEN_REQUESTS =
"giraph.waitForRequestsConfirmation";
@@ -420,7 +410,7 @@ public class NettyClient<I extends Writa
*/
public void sendWritableRequest(Integer destWorkerId,
InetSocketAddress remoteServer,
- WritableRequest<I, V, E, M> request) {
+ WritableRequest request) {
if (clientRequestIdRequestInfoMap.isEmpty()) {
byteCounter.resetAll();
}
@@ -471,8 +461,7 @@ public class NettyClient<I extends Writa
*/
private void waitSomeRequests(int maxOpenRequests) {
List<ClientRequestId> addedRequestIds = Lists.newArrayList();
- List<RequestInfo<I, V, E, M>> addedRequestInfos =
- Lists.newArrayList();
+ List<RequestInfo> addedRequestInfos = Lists.newArrayList();
while (clientRequestIdRequestInfoMap.size() > maxOpenRequests) {
// Wait for requests to complete for some time
@@ -521,7 +510,7 @@ public class NettyClient<I extends Writa
"destination = " + writeFuture.getChannel().getRemoteAddress() +
" " + requestInfo);
addedRequestIds.add(entry.getKey());
- addedRequestInfos.add(new RequestInfo<I, V, E, M>(
+ addedRequestInfos.add(new RequestInfo(
requestInfo.getDestinationAddress(), requestInfo.getRequest()));
}
}
@@ -529,7 +518,7 @@ public class NettyClient<I extends Writa
// Add any new requests to the system, connect if necessary, and re-send
for (int i = 0; i < addedRequestIds.size(); ++i) {
ClientRequestId requestId = addedRequestIds.get(i);
- RequestInfo<I, V, E, M> requestInfo = addedRequestInfos.get(i);
+ RequestInfo requestInfo = addedRequestInfos.get(i);
if (clientRequestIdRequestInfoMap.put(requestId, requestInfo) ==
null) {
Added: giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterClient.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterClient.java?rev=1377792&view=auto
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterClient.java (added)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterClient.java Mon Aug 27 18:54:23 2012
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.giraph.comm;
+
+import org.apache.giraph.graph.WorkerInfo;
+import org.apache.hadoop.mapreduce.Mapper;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import java.net.InetSocketAddress;
+import java.util.Collection;
+import java.util.Set;
+
+/**
+ * Netty implementation of {@link MasterClient}
+ */
+public class NettyMasterClient implements MasterClient {
+ /** Netty client that does the actual I/O */
+ private final NettyClient nettyClient;
+ /** Worker information for current superstep */
+ private Collection<WorkerInfo> workers;
+
+ /**
+ * Constructor
+ *
+ * @param context Context from mapper
+ */
+ public NettyMasterClient(Mapper<?, ?, ?, ?>.Context context) {
+ this.nettyClient = new NettyClient(context);
+ workers = Lists.newArrayList();
+ }
+
+ @Override
+ public void fixWorkerAddresses(Iterable<WorkerInfo> workers) {
+ this.workers.clear();
+ Iterables.addAll(this.workers, workers);
+ Set<InetSocketAddress> addresses =
+ Sets.newHashSetWithExpectedSize(this.workers.size());
+ for (WorkerInfo worker : workers) {
+ addresses.add(worker.getInetSocketAddress());
+ }
+ nettyClient.connectAllAddresses(addresses);
+ }
+
+ @Override
+ public void flush() {
+ nettyClient.waitAllRequests();
+ }
+
+ @Override
+ public void closeConnections() {
+ nettyClient.stop();
+ }
+}
Added: giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterClientServer.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterClientServer.java?rev=1377792&view=auto
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterClientServer.java (added)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterClientServer.java Mon Aug 27 18:54:23 2012
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.giraph.comm;
+
+import org.apache.giraph.graph.WorkerInfo;
+import org.apache.hadoop.mapreduce.Mapper;
+
+import java.net.InetSocketAddress;
+
+/**
+ * Netty based implementation of the {@link MasterClientServer} interface.
+ */
+public class NettyMasterClientServer implements MasterClientServer {
+ /** Client that sends requests */
+ private final MasterClient client;
+ /** Server that processes requests */
+ private final MasterServer server;
+
+ /**
+ * Constructor
+ *
+ * @param context Mapper context
+ */
+ public NettyMasterClientServer(Mapper<?, ?, ?, ?>.Context context) {
+ client = new NettyMasterClient(context);
+ server = new NettyMasterServer(context.getConfiguration());
+ }
+
+ @Override
+ public void fixWorkerAddresses(Iterable<WorkerInfo> workers) {
+ client.fixWorkerAddresses(workers);
+ }
+
+ @Override
+ public void flush() {
+ client.flush();
+ }
+
+ @Override
+ public void closeConnections() {
+ client.closeConnections();
+ }
+
+ @Override
+ public InetSocketAddress getMyAddress() {
+ return server.getMyAddress();
+ }
+
+ @Override
+ public void close() {
+ server.close();
+ }
+}
Added: giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterServer.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterServer.java?rev=1377792&view=auto
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterServer.java (added)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/NettyMasterServer.java Mon Aug 27 18:54:23 2012
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.giraph.comm;
+
+import org.apache.hadoop.conf.Configuration;
+
+import java.net.InetSocketAddress;
+
+/**
+ * Netty implementation of {@link MasterServer}
+ */
+public class NettyMasterServer implements MasterServer {
+ /** Netty client that does the actual I/O */
+ private final NettyServer nettyServer;
+
+ /**
+ * Constructor
+ *
+ * @param conf Hadoop configuration
+ */
+ public NettyMasterServer(Configuration conf) {
+ nettyServer = new NettyServer(conf,
+ new MasterRequestServerHandler.Factory());
+ nettyServer.start();
+ }
+
+ @Override
+ public InetSocketAddress getMyAddress() {
+ return nettyServer.getMyAddress();
+ }
+
+ @Override
+ public void close() {
+ nettyServer.stop();
+ }
+}
Modified: giraph/trunk/src/main/java/org/apache/giraph/comm/NettyServer.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/NettyServer.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/NettyServer.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/NettyServer.java Mon Aug 27 18:54:23 2012
@@ -23,13 +23,10 @@ import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadPoolExecutor;
-import org.apache.giraph.comm.messages.SendPartitionCurrentMessagesRequest;
import org.apache.giraph.graph.GiraphJob;
+import org.apache.giraph.comm.RequestServerHandler.RequestServerHandlerFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
import org.apache.log4j.Logger;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
@@ -47,16 +44,8 @@ import com.google.common.util.concurrent
/**
* This server uses Netty and will implement all Giraph communication
- *
- * @param <I> Vertex id
- * @param <V> Vertex data
- * @param <E> Edge data
- * @param <M> Message data
*/
-@SuppressWarnings("rawtypes")
-public class NettyServer<I extends WritableComparable,
- V extends Writable, E extends Writable,
- M extends Writable> {
+public class NettyServer {
/** Default maximum thread pool size */
public static final int MAXIMUM_THREAD_POOL_SIZE_DEFAULT = 32;
/** Class logger */
@@ -67,8 +56,6 @@ public class NettyServer<I extends Writa
private ChannelFactory channelFactory;
/** Accepted channels */
private final ChannelGroup accepted = new DefaultChannelGroup();
- /** Worker thread pool (if implemented as a ThreadPoolExecutor) */
- private ThreadPoolExecutor workerThreadPool = null;
/** Local hostname */
private final String localHostname;
/** Address of the server */
@@ -77,10 +64,8 @@ public class NettyServer<I extends Writa
private final int maximumPoolSize;
/** TCP backlog */
private final int tcpBacklog;
- /** Request reqistry */
- private final RequestRegistry requestRegistry = new RequestRegistry();
- /** Server data */
- private final ServerData<I, V, E, M> serverData;
+ /** Factory for {@link RequestServerHandler} */
+ private final RequestServerHandlerFactory requestServerHandlerFactory;
/** Server bootstrap */
private ServerBootstrap bootstrap;
/** Byte counter for this client */
@@ -100,20 +85,12 @@ public class NettyServer<I extends Writa
* Constructor for creating the server
*
* @param conf Configuration to use
- * @param serverData Server data to operate on
+ * @param requestServerHandlerFactory Factory for request handlers
*/
- public NettyServer(Configuration conf, ServerData<I, V, E, M> serverData) {
+ public NettyServer(Configuration conf,
+ RequestServerHandlerFactory requestServerHandlerFactory) {
this.conf = conf;
- this.serverData = serverData;
- requestRegistry.registerClass(
- new SendVertexRequest<I, V, E, M>());
- requestRegistry.registerClass(
- new SendPartitionMessagesRequest<I, V, E, M>());
- requestRegistry.registerClass(
- new SendPartitionMutationsRequest<I, V, E, M>());
- requestRegistry.registerClass(
- new SendPartitionCurrentMessagesRequest<I, V, E, M>());
- requestRegistry.shutdown();
+ this.requestServerHandlerFactory = requestServerHandlerFactory;
sendBufferSize = conf.getInt(GiraphJob.SERVER_SEND_BUFFER_SIZE,
GiraphJob.DEFAULT_SERVER_SEND_BUFFER_SIZE);
@@ -163,17 +140,18 @@ public class NettyServer<I extends Writa
return Channels.pipeline(
byteCounter,
new LengthFieldBasedFrameDecoder(1024 * 1024 * 1024, 0, 4, 0, 4),
- new RequestDecoder<I, V, E, M>(conf, requestRegistry, byteCounter),
- new RequestServerHandler<I, V, E, M>(serverData,
- workerRequestReservedMap, conf));
+ new RequestDecoder(conf, byteCounter),
+ requestServerHandlerFactory.newHandler(workerRequestReservedMap,
+ conf));
}
});
int taskId = conf.getInt("mapred.task.partition", -1);
int numTasks = conf.getInt("mapred.map.tasks", 1);
- int numWorkers = conf.getInt(GiraphJob.MAX_WORKERS, numTasks);
+ // number of workers + 1 for master
+ int numServers = conf.getInt(GiraphJob.MAX_WORKERS, numTasks) + 1;
int portIncrementConstant =
- (int) Math.pow(10, Math.ceil(Math.log10(numWorkers)));
+ (int) Math.pow(10, Math.ceil(Math.log10(numServers)));
int bindPort = conf.getInt(GiraphJob.RPC_INITIAL_PORT,
GiraphJob.RPC_INITIAL_PORT_DEFAULT) +
taskId;
@@ -241,6 +219,7 @@ public class NettyServer<I extends Writa
bossExecutorService.shutdownNow();
workerExecutorService.shutdownNow();
bootstrap.releaseExternalResources();
+ channelFactory.releaseExternalResources();
}
public InetSocketAddress getMyAddress() {
Modified: giraph/trunk/src/main/java/org/apache/giraph/comm/NettyWorkerClient.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/NettyWorkerClient.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/NettyWorkerClient.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/NettyWorkerClient.java Mon Aug 27 18:54:23 2012
@@ -64,7 +64,7 @@ public class NettyWorkerClient<I extends
/** Hadoop configuration */
private final Configuration conf;
/** Netty client that does that actual I/O */
- private final NettyClient<I, V, E, M> nettyClient;
+ private final NettyClient nettyClient;
/** Centralized service, needed to get vertex ranges */
private final CentralizedServiceWorker<I, V, E, M> service;
/**
@@ -101,7 +101,7 @@ public class NettyWorkerClient<I extends
public NettyWorkerClient(Mapper<?, ?, ?, ?>.Context context,
CentralizedServiceWorker<I, V, E, M> service,
ServerData<I, V, E, M> serverData) {
- this.nettyClient = new NettyClient<I, V, E, M>(context);
+ this.nettyClient = new NettyClient(context);
this.conf = context.getConfiguration();
this.service = service;
maxMessagesPerPartition = conf.getInt(GiraphJob.MSG_SIZE,
@@ -148,6 +148,11 @@ public class NettyWorkerClient<I extends
partitionOwner.getPartitionId()));
}
}
+ boolean useNetty = conf.getBoolean(GiraphJob.USE_NETTY,
+ GiraphJob.USE_NETTY_DEFAULT);
+ if (useNetty) {
+ addresses.add(service.getMasterInfo().getInetSocketAddress());
+ }
nettyClient.connectAllAddresses(addresses);
}
@@ -197,11 +202,11 @@ public class NettyWorkerClient<I extends
*/
private void doRequest(WorkerInfo workerInfo,
InetSocketAddress remoteServerAddress,
- WritableRequest<I, V, E, M> writableRequest) {
+ WritableRequest writableRequest) {
// If this is local, execute locally
if (service.getWorkerInfo().getPartitionId() ==
workerInfo.getPartitionId()) {
- writableRequest.doRequest(serverData);
+ ((WorkerRequest) writableRequest).doRequest(serverData);
} else {
nettyClient.sendWritableRequest(
workerInfo.getPartitionId(), remoteServerAddress, writableRequest);
@@ -229,7 +234,7 @@ public class NettyWorkerClient<I extends
getInetSocketAddress(partitionOwner.getWorkerInfo(), partitionId);
Map<I, Collection<M>> partitionMessages =
sendMessageCache.removePartitionMessages(partitionId);
- WritableRequest<I, V, E, M> writableRequest =
+ WritableRequest writableRequest =
new SendPartitionMessagesRequest<I, V, E, M>(
partitionId, partitionMessages);
doRequest(partitionOwner.getWorkerInfo(), remoteServerAddress,
@@ -248,7 +253,7 @@ public class NettyWorkerClient<I extends
}
int partitionId = partition.getId();
- WritableRequest<I, V, E, M> vertexRequest =
+ WritableRequest vertexRequest =
new SendVertexRequest<I, V, E, M>(partitionId,
partition.getVertices());
doRequest(workerInfo, remoteServerAddress, vertexRequest);
@@ -269,7 +274,7 @@ public class NettyWorkerClient<I extends
"sendPartitionReq: Got IOException ", e);
}
if (messagesInMap > maxMessagesPerPartition) {
- WritableRequest<I, V, E, M> messagesRequest = new
+ WritableRequest messagesRequest = new
SendPartitionCurrentMessagesRequest<I, V, E, M>(partitionId, map);
doRequest(workerInfo, remoteServerAddress, messagesRequest);
map.clear();
@@ -277,7 +282,7 @@ public class NettyWorkerClient<I extends
}
}
if (!map.isEmpty()) {
- WritableRequest<I, V, E, M> messagesRequest = new
+ WritableRequest messagesRequest = new
SendPartitionCurrentMessagesRequest<I, V, E, M>(partitionId, map);
doRequest(workerInfo, remoteServerAddress, messagesRequest);
}
@@ -300,7 +305,7 @@ public class NettyWorkerClient<I extends
getInetSocketAddress(partitionOwner.getWorkerInfo(), partitionId);
Map<I, VertexMutations<I, V, E, M>> partitionMutations =
sendMutationsCache.removePartitionMutations(partitionId);
- WritableRequest<I, V, E, M> writableRequest =
+ WritableRequest writableRequest =
new SendPartitionMutationsRequest<I, V, E, M>(
partitionId, partitionMutations);
doRequest(partitionOwner.getWorkerInfo(), remoteServerAddress,
@@ -390,7 +395,7 @@ public class NettyWorkerClient<I extends
sendMessageCache.removeAllPartitionMessages();
for (Entry<Integer, Map<I, Collection<M>>> entry :
remainingMessageCache.entrySet()) {
- WritableRequest<I, V, E, M> writableRequest =
+ WritableRequest writableRequest =
new SendPartitionMessagesRequest<I, V, E, M>(
entry.getKey(), entry.getValue());
PartitionOwner partitionOwner =
@@ -408,7 +413,7 @@ public class NettyWorkerClient<I extends
sendMutationsCache.removeAllPartitionMutations();
for (Entry<Integer, Map<I, VertexMutations<I, V, E, M>>> entry :
remainingMutationsCache.entrySet()) {
- WritableRequest<I, V, E, M> writableRequest =
+ WritableRequest writableRequest =
new SendPartitionMutationsRequest<I, V, E, M>(
entry.getKey(), entry.getValue());
PartitionOwner partitionOwner =
Modified: giraph/trunk/src/main/java/org/apache/giraph/comm/NettyWorkerServer.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/NettyWorkerServer.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/NettyWorkerServer.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/NettyWorkerServer.java Mon Aug 27 18:54:23 2012
@@ -63,7 +63,7 @@ public class NettyWorkerServer<I extends
/** Service worker */
private final CentralizedServiceWorker<I, V, E, M> service;
/** Netty server that does that actual I/O */
- private final NettyServer<I, V, E, M> nettyServer;
+ private final NettyServer nettyServer;
/** Server data storage */
private final ServerData<I, V, E, M> serverData;
@@ -98,7 +98,8 @@ public class NettyWorkerServer<I extends
serverData = new ServerData<I, V, E, M>(conf, storeFactory);
}
- nettyServer = new NettyServer<I, V, E, M>(conf, serverData);
+ nettyServer = new NettyServer(conf,
+ new WorkerRequestServerHandler.Factory<I, V, E, M>(serverData));
nettyServer.start();
}
Modified: giraph/trunk/src/main/java/org/apache/giraph/comm/RequestDecoder.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/RequestDecoder.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/RequestDecoder.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/RequestDecoder.java Mon Aug 27 18:54:23 2012
@@ -19,8 +19,6 @@
package org.apache.giraph.comm;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.log4j.Logger;
import org.jboss.netty.buffer.ChannelBuffer;
@@ -31,23 +29,13 @@ import org.jboss.netty.handler.codec.one
/**
* Decodes encoded requests from the client.
- *
- * @param <I> Vertex id
- * @param <V> Vertex data
- * @param <E> Edge data
- * @param <M> Message data
*/
-@SuppressWarnings("rawtypes")
-public class RequestDecoder<I extends WritableComparable,
- V extends Writable, E extends Writable,
- M extends Writable> extends OneToOneDecoder {
+public class RequestDecoder extends OneToOneDecoder {
/** Class logger */
private static final Logger LOG =
Logger.getLogger(RequestDecoder.class);
/** Configuration */
private final Configuration conf;
- /** Registry of requests */
- private final RequestRegistry requestRegistry;
/** Byte counter to output */
private final ByteCounter byteCounter;
@@ -55,14 +43,10 @@ public class RequestDecoder<I extends Wr
* Constructor.
*
* @param conf Configuration
- * @param requestRegistry Request registry
* @param byteCounter Keeps track of the decoded bytes
*/
- public RequestDecoder(
- Configuration conf, RequestRegistry requestRegistry,
- ByteCounter byteCounter) {
+ public RequestDecoder(Configuration conf, ByteCounter byteCounter) {
this.conf = conf;
- this.requestRegistry = requestRegistry;
this.byteCounter = byteCounter;
}
@@ -84,15 +68,13 @@ public class RequestDecoder<I extends Wr
ChannelBuffer buffer = (ChannelBuffer) msg;
ChannelBufferInputStream inputStream = new ChannelBufferInputStream(buffer);
int enumValue = inputStream.readByte();
- RequestRegistry.Type type = RequestRegistry.Type.values()[enumValue];
+ RequestType type = RequestType.values()[enumValue];
if (LOG.isDebugEnabled()) {
LOG.debug("decode: Got a request of type " + type);
}
- @SuppressWarnings("unchecked")
- Class<? extends WritableRequest<I, V, E, M>> writableRequestClass =
- (Class<? extends WritableRequest<I, V, E, M>>)
- requestRegistry.getClass(type);
- WritableRequest<I, V, E, M> writableRequest =
+ Class<? extends WritableRequest> writableRequestClass =
+ type.getRequestClass();
+ WritableRequest writableRequest =
ReflectionUtils.newInstance(writableRequestClass, conf);
writableRequest.readFields(inputStream);
return writableRequest;
Modified: giraph/trunk/src/main/java/org/apache/giraph/comm/RequestEncoder.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/RequestEncoder.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/RequestEncoder.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/RequestEncoder.java Mon Aug 27 18:54:23 2012
@@ -18,8 +18,6 @@
package org.apache.giraph.comm;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
import org.apache.log4j.Logger;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBufferOutputStream;
@@ -30,16 +28,8 @@ import org.jboss.netty.handler.codec.one
/**
* Requests have a request type and an encoded request.
- *
- * @param <I> Vertex id
- * @param <V> Vertex data
- * @param <E> Edge data
- * @param <M> Message data
*/
-@SuppressWarnings("rawtypes")
-public class RequestEncoder<I extends WritableComparable,
- V extends Writable, E extends Writable,
- M extends Writable> extends OneToOneEncoder {
+public class RequestEncoder extends OneToOneEncoder {
/** Class logger */
private static final Logger LOG = Logger.getLogger(RequestEncoder.class);
/** Holds the place of the message length until known */
@@ -48,13 +38,12 @@ public class RequestEncoder<I extends Wr
@Override
protected Object encode(ChannelHandlerContext ctx,
Channel channel, Object msg) throws Exception {
- if (!(msg instanceof WritableRequest<?, ?, ?, ?>)) {
+ if (!(msg instanceof WritableRequest)) {
throw new IllegalArgumentException(
"encode: Got a message of type " + msg.getClass());
}
@SuppressWarnings("unchecked")
- WritableRequest<I, V, E, M> writableRequest =
- (WritableRequest<I, V, E, M>) msg;
+ WritableRequest writableRequest = (WritableRequest) msg;
ChannelBufferOutputStream outputStream =
new ChannelBufferOutputStream(ChannelBuffers.dynamicBuffer(
10, ctx.getChannel().getConfig().getBufferFactory()));
Modified: giraph/trunk/src/main/java/org/apache/giraph/comm/RequestInfo.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/RequestInfo.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/RequestInfo.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/RequestInfo.java Mon Aug 27 18:54:23 2012
@@ -20,26 +20,18 @@ package org.apache.giraph.comm;
import java.net.InetSocketAddress;
import java.util.Date;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
import org.jboss.netty.channel.ChannelFuture;
/**
* Help track requests throughout the system
- *
- * @param <I> Vertex id
- * @param <V> Vertex data
- * @param <E> Edge data
- * @param <M> Message data
*/
-public class RequestInfo<I extends WritableComparable,
- V extends Writable, E extends Writable, M extends Writable> {
+public class RequestInfo {
/** Destination of the request */
private final InetSocketAddress destinationAddress;
/** When the request was started */
private final long startedMsecs;
/** Request */
- private final WritableRequest<I, V, E, M> request;
+ private final WritableRequest request;
/** Future of the write of this request*/
private volatile ChannelFuture writeFuture;
@@ -50,7 +42,7 @@ public class RequestInfo<I extends Writa
* @param request Request that is sent
*/
public RequestInfo(InetSocketAddress destinationAddress,
- WritableRequest<I, V, E, M> request) {
+ WritableRequest request) {
this.destinationAddress = destinationAddress;
this.request = request;
this.startedMsecs = System.currentTimeMillis();
@@ -73,7 +65,7 @@ public class RequestInfo<I extends Writa
return System.currentTimeMillis() - startedMsecs;
}
- public WritableRequest<I, V, E, M> getRequest() {
+ public WritableRequest getRequest() {
return request;
}
Modified: giraph/trunk/src/main/java/org/apache/giraph/comm/RequestServerHandler.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/RequestServerHandler.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/RequestServerHandler.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/RequestServerHandler.java Mon Aug 27 18:54:23 2012
@@ -20,8 +20,6 @@ package org.apache.giraph.comm;
import org.apache.giraph.graph.GiraphJob;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
import org.apache.log4j.Logger;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
@@ -34,15 +32,10 @@ import org.jboss.netty.channel.SimpleCha
/**
* Generic handler of requests.
*
- * @param <I> Vertex id
- * @param <V> Vertex data
- * @param <E> Edge data
- * @param <M> Message data
+ * @param <R> Request type
*/
-@SuppressWarnings("rawtypes")
-public class RequestServerHandler<I extends WritableComparable,
- V extends Writable, E extends Writable,
- M extends Writable> extends SimpleChannelUpstreamHandler {
+public abstract class RequestServerHandler<R> extends
+ SimpleChannelUpstreamHandler {
/** Number of bytes in the encoded response */
public static final int RESPONSE_BYTES = 13;
/** Class logger */
@@ -52,25 +45,20 @@ public class RequestServerHandler<I exte
private static volatile boolean ALREADY_CLOSED_FIRST_REQUEST = false;
/** Close connection on first request (used for simulating failure) */
private final boolean closeFirstRequest;
- /** Data that can be accessed for handling requests */
- private final ServerData<I, V, E, M> serverData;
/** Request reserved map (for exactly one semantics) */
private final WorkerRequestReservedMap workerRequestReservedMap;
/** My worker id */
private final int myWorkerId;
/**
- * Constructor with external server data
+ * Constructor
*
- * @param serverData Data held by the server
* @param workerRequestReservedMap Worker request reservation map
* @param conf Configuration
*/
public RequestServerHandler(
- ServerData<I, V, E, M> serverData,
WorkerRequestReservedMap workerRequestReservedMap,
Configuration conf) {
- this.serverData = serverData;
this.workerRequestReservedMap = workerRequestReservedMap;
closeFirstRequest = conf.getBoolean(
GiraphJob.NETTY_SIMULATE_FIRST_REQUEST_CLOSED,
@@ -85,9 +73,7 @@ public class RequestServerHandler<I exte
LOG.debug("messageReceived: Got " + e.getMessage().getClass());
}
- @SuppressWarnings("unchecked")
- WritableRequest<I, V, E, M> writableRequest =
- (WritableRequest<I, V, E, M>) e.getMessage();
+ WritableRequest writableRequest = (WritableRequest) e.getMessage();
// Simulate a closed connection on the first request (if desired)
if (closeFirstRequest && !ALREADY_CLOSED_FIRST_REQUEST) {
@@ -104,7 +90,7 @@ public class RequestServerHandler<I exte
if (workerRequestReservedMap.reserveRequest(
writableRequest.getClientId(),
writableRequest.getRequestId())) {
- writableRequest.doRequest(serverData);
+ processRequest((R) writableRequest);
alreadyDone = 0;
} else {
LOG.info("messageReceived: Request id " +
@@ -122,6 +108,13 @@ public class RequestServerHandler<I exte
e.getChannel().write(buffer);
}
+ /**
+ * Process request
+ *
+ * @param request Request to process
+ */
+ public abstract void processRequest(R request);
+
@Override
public void channelConnected(ChannelHandlerContext ctx,
ChannelStateEvent e) throws Exception {
@@ -146,4 +139,20 @@ public class RequestServerHandler<I exte
LOG.warn("exceptionCaught: Channel failed with " +
"remote address " + ctx.getChannel().getRemoteAddress(), e.getCause());
}
+
+ /**
+ * Factory for {@link RequestServerHandler}
+ */
+ public interface RequestServerHandlerFactory {
+ /**
+ * Create new {@link RequestServerHandler}
+ *
+ * @param workerRequestReservedMap Worker request reservation map
+ * @param conf Configuration to use
+ * @return New {@link RequestServerHandler}
+ */
+ RequestServerHandler newHandler(
+ WorkerRequestReservedMap workerRequestReservedMap,
+ Configuration conf);
+ }
}
Added: giraph/trunk/src/main/java/org/apache/giraph/comm/RequestType.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/RequestType.java?rev=1377792&view=auto
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/RequestType.java (added)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/RequestType.java Mon Aug 27 18:54:23 2012
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.giraph.comm;
+
+import org.apache.giraph.comm.messages.SendPartitionCurrentMessagesRequest;
+
+/**
+ * Type of the request
+ */
+public enum RequestType {
+ /** Sending vertices request */
+ SEND_VERTEX_REQUEST(SendVertexRequest.class),
+ /** Sending a partition of messages for next superstep */
+ SEND_PARTITION_MESSAGES_REQUEST(SendPartitionMessagesRequest.class),
+ /**
+ * Sending a partition of messages for current superstep
+ * (used during partition exchange)
+ */
+ SEND_PARTITION_CURRENT_MESSAGES_REQUEST
+ (SendPartitionCurrentMessagesRequest.class),
+ /** Send a partition of mutations */
+ SEND_PARTITION_MUTATIONS_REQUEST(SendPartitionMutationsRequest.class);
+
+ /** Class of request which this type corresponds to */
+ private final Class<? extends WritableRequest> requestClass;
+
+ /**
+ * Constructor
+ *
+ * @param requestClass Class of request which this type corresponds to
+ */
+ private RequestType(Class<? extends WritableRequest> requestClass) {
+ this.requestClass = requestClass;
+ }
+
+ /**
+ * Get class of request which this type corresponds to
+ *
+ * @return Class of request which this type corresponds to
+ */
+ public Class<? extends WritableRequest> getRequestClass() {
+ return requestClass;
+ }
+}
Modified: giraph/trunk/src/main/java/org/apache/giraph/comm/SendPartitionMessagesRequest.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/SendPartitionMessagesRequest.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/SendPartitionMessagesRequest.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/SendPartitionMessagesRequest.java Mon Aug 27 18:54:23 2012
@@ -18,7 +18,6 @@
package org.apache.giraph.comm;
-import org.apache.giraph.comm.RequestRegistry.Type;
import org.apache.giraph.graph.BspUtils;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
@@ -45,8 +44,8 @@ import java.util.Map.Entry;
*/
@SuppressWarnings("rawtypes")
public class SendPartitionMessagesRequest<I extends WritableComparable,
- V extends Writable, E extends Writable,
- M extends Writable> extends WritableRequest<I, V, E, M> {
+ V extends Writable, E extends Writable, M extends Writable> extends
+ WritableRequest implements WorkerRequest<I, V, E, M> {
/** Class logger */
private static final Logger LOG =
Logger.getLogger(SendPartitionMessagesRequest.class);
@@ -108,8 +107,8 @@ public class SendPartitionMessagesReques
}
@Override
- public Type getType() {
- return Type.SEND_PARTITION_MESSAGES_REQUEST;
+ public RequestType getType() {
+ return RequestType.SEND_PARTITION_MESSAGES_REQUEST;
}
@Override
Modified: giraph/trunk/src/main/java/org/apache/giraph/comm/SendPartitionMutationsRequest.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/SendPartitionMutationsRequest.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/SendPartitionMutationsRequest.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/SendPartitionMutationsRequest.java Mon Aug 27 18:54:23 2012
@@ -25,7 +25,6 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
-import org.apache.giraph.comm.RequestRegistry.Type;
import org.apache.giraph.graph.BspUtils;
import org.apache.giraph.graph.VertexMutations;
import org.apache.hadoop.io.Writable;
@@ -44,8 +43,8 @@ import com.google.common.collect.Maps;
*/
@SuppressWarnings("rawtypes")
public class SendPartitionMutationsRequest<I extends WritableComparable,
- V extends Writable, E extends Writable,
- M extends Writable> extends WritableRequest<I, V, E, M> {
+ V extends Writable, E extends Writable, M extends Writable> extends
+ WritableRequest implements WorkerRequest<I, V, E, M> {
/** Class logger */
private static final Logger LOG =
Logger.getLogger(SendPartitionMutationsRequest.class);
@@ -103,8 +102,8 @@ public class SendPartitionMutationsReque
}
@Override
- public Type getType() {
- return Type.SEND_PARTITION_MUTATIONS_REQUEST;
+ public RequestType getType() {
+ return RequestType.SEND_PARTITION_MUTATIONS_REQUEST;
}
@Override
Modified: giraph/trunk/src/main/java/org/apache/giraph/comm/SendVertexRequest.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/SendVertexRequest.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/SendVertexRequest.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/SendVertexRequest.java Mon Aug 27 18:54:23 2012
@@ -18,7 +18,6 @@
package org.apache.giraph.comm;
-import org.apache.giraph.comm.RequestRegistry.Type;
import org.apache.giraph.graph.BspUtils;
import org.apache.giraph.graph.Vertex;
import org.apache.hadoop.io.Writable;
@@ -42,8 +41,8 @@ import java.util.Collection;
*/
@SuppressWarnings("rawtypes")
public class SendVertexRequest<I extends WritableComparable,
- V extends Writable, E extends Writable,
- M extends Writable> extends WritableRequest<I, V, E, M> {
+ V extends Writable, E extends Writable, M extends Writable> extends
+ WritableRequest implements WorkerRequest<I, V, E, M> {
/** Class logger */
private static final Logger LOG =
Logger.getLogger(SendVertexRequest.class);
@@ -91,8 +90,8 @@ public class SendVertexRequest<I extends
}
@Override
- public Type getType() {
- return Type.SEND_VERTEX_REQUEST;
+ public RequestType getType() {
+ return RequestType.SEND_VERTEX_REQUEST;
}
@Override
Added: giraph/trunk/src/main/java/org/apache/giraph/comm/WorkerRequest.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/WorkerRequest.java?rev=1377792&view=auto
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/WorkerRequest.java (added)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/WorkerRequest.java Mon Aug 27 18:54:23 2012
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.giraph.comm;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+/**
+ * Interface for requests sent to worker to extend
+ *
+ * @param <I> Vertex id
+ * @param <V> Vertex data
+ * @param <E> Edge data
+ * @param <M> Message data
+ */
+public interface WorkerRequest<I extends WritableComparable,
+ V extends Writable, E extends Writable, M extends Writable> {
+ /**
+ * Execute the request
+ *
+ * @param serverData Accessible data that can be mutated per the request
+ */
+ void doRequest(ServerData<I, V, E, M> serverData);
+}
Added: giraph/trunk/src/main/java/org/apache/giraph/comm/WorkerRequestServerHandler.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/WorkerRequestServerHandler.java?rev=1377792&view=auto
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/WorkerRequestServerHandler.java (added)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/WorkerRequestServerHandler.java Mon Aug 27 18:54:23 2012
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.giraph.comm;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+/**
+ * Handler for requests on worker
+ *
+ * @param <I> Vertex id
+ * @param <V> Vertex data
+ * @param <E> Edge data
+ * @param <M> Message data
+ */
+public class WorkerRequestServerHandler<I extends WritableComparable,
+ V extends Writable, E extends Writable, M extends Writable> extends
+ RequestServerHandler<WorkerRequest<I, V, E, M>> {
+ /** Data that can be accessed for handling requests */
+ private final ServerData<I, V, E, M> serverData;
+
+ /**
+ * Constructor with external server data
+ *
+ * @param serverData Data held by the server
+ * @param workerRequestReservedMap Worker request reservation map
+ * @param conf Configuration
+ */
+ public WorkerRequestServerHandler(ServerData<I, V, E, M> serverData,
+ WorkerRequestReservedMap workerRequestReservedMap,
+ Configuration conf) {
+ super(workerRequestReservedMap, conf);
+ this.serverData = serverData;
+ }
+
+ @Override
+ public void processRequest(WorkerRequest<I, V, E, M> request) {
+ request.doRequest(serverData);
+ }
+
+ /** Factory for {@link WorkerRequestServerHandler} */
+ public static class Factory<I extends WritableComparable,
+ V extends Writable, E extends Writable, M extends Writable> implements
+ RequestServerHandlerFactory {
+ /** Data that can be accessed for handling requests */
+ private final ServerData<I, V, E, M> serverData;
+
+ /**
+ * Constructor
+ *
+ * @param serverData Data held by the server
+ */
+ public Factory(ServerData<I, V, E, M> serverData) {
+ this.serverData = serverData;
+ }
+
+ @Override
+ public RequestServerHandler newHandler(
+ WorkerRequestReservedMap workerRequestReservedMap,
+ Configuration conf) {
+ return new WorkerRequestServerHandler<I, V, E,
+ M>(serverData, workerRequestReservedMap, conf);
+ }
+ }
+}
Modified: giraph/trunk/src/main/java/org/apache/giraph/comm/WritableRequest.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/WritableRequest.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/WritableRequest.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/WritableRequest.java Mon Aug 27 18:54:23 2012
@@ -21,24 +21,14 @@ package org.apache.giraph.comm;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
-import org.apache.giraph.comm.RequestRegistry.Type;
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
/**
* Interface for requests to implement
- *
- * @param <I> Vertex id
- * @param <V> Vertex data
- * @param <E> Edge data
- * @param <M> Message data
*/
-@SuppressWarnings("rawtypes")
-public abstract class WritableRequest<I extends WritableComparable,
- V extends Writable, E extends Writable,
- M extends Writable> implements Writable, Configurable {
+public abstract class WritableRequest implements Writable, Configurable {
/** Configuration */
private Configuration conf;
/** Client id */
@@ -67,7 +57,7 @@ public abstract class WritableRequest<I
*
* @return Request type
*/
- public abstract Type getType();
+ public abstract RequestType getType();
/**
* Serialize the request
@@ -83,13 +73,6 @@ public abstract class WritableRequest<I
*/
abstract void writeRequest(DataOutput output) throws IOException;
- /**
- * Execute the request
- *
- * @param serverData Accessible data that can be mutated per the request
- */
- public abstract void doRequest(ServerData<I, V, E, M> serverData);
-
@Override
public final Configuration getConf() {
return conf;
Modified: giraph/trunk/src/main/java/org/apache/giraph/comm/messages/SendPartitionCurrentMessagesRequest.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/comm/messages/SendPartitionCurrentMessagesRequest.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/comm/messages/SendPartitionCurrentMessagesRequest.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/comm/messages/SendPartitionCurrentMessagesRequest.java Mon Aug 27 18:54:23 2012
@@ -18,7 +18,7 @@
package org.apache.giraph.comm.messages;
-import org.apache.giraph.comm.RequestRegistry.Type;
+import org.apache.giraph.comm.RequestType;
import org.apache.giraph.comm.SendPartitionMessagesRequest;
import org.apache.giraph.comm.ServerData;
import org.apache.hadoop.io.Writable;
@@ -55,8 +55,8 @@ public class SendPartitionCurrentMessage
}
@Override
- public Type getType() {
- return Type.SEND_PARTITION_CURRENT_MESSAGES_REQUEST;
+ public RequestType getType() {
+ return RequestType.SEND_PARTITION_CURRENT_MESSAGES_REQUEST;
}
@Override
Modified: giraph/trunk/src/main/java/org/apache/giraph/graph/BspService.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/graph/BspService.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/graph/BspService.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/graph/BspService.java Mon Aug 27 18:54:23 2012
@@ -94,6 +94,8 @@ public abstract class BspService<I exten
"/_applicationAttemptsDir";
/** Where the master election happens */
public static final String MASTER_ELECTION_DIR = "/_masterElectionDir";
+ /** Denotes current master's info */
+ public static final String CURRENT_MASTER_INFO_NODE = "/_currentMaster";
/** Superstep scope */
public static final String SUPERSTEP_DIR = "/_superstepDir";
/** Where the merged aggregators are located */
@@ -208,6 +210,8 @@ public abstract class BspService<I exten
protected final String checkpointBasePath;
/** Path to the master election path */
protected final String masterElectionPath;
+ /** Path to current master info */
+ protected final String currentMasterPath;
/** Private ZooKeeper instance that implements the service */
private final ZooKeeperExt zk;
/** Has the Connection occurred? */
@@ -339,6 +343,7 @@ public abstract class BspService<I exten
GiraphJob.CHECKPOINT_DIRECTORY,
GiraphJob.CHECKPOINT_DIRECTORY_DEFAULT + "/" + getJobId());
masterElectionPath = basePath + MASTER_ELECTION_DIR;
+ currentMasterPath = basePath + CURRENT_MASTER_INFO_NODE;
if (LOG.isInfoEnabled()) {
LOG.info("BspService: Connecting to ZooKeeper with job " + jobId +
", " + getTaskPartition() + " on " + serverPortList);
Modified: giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceMaster.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceMaster.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceMaster.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceMaster.java Mon Aug 27 18:54:23 2012
@@ -23,6 +23,8 @@ import org.apache.giraph.bsp.Application
import org.apache.giraph.bsp.BspInputFormat;
import org.apache.giraph.bsp.CentralizedServiceMaster;
import org.apache.giraph.bsp.SuperstepState;
+import org.apache.giraph.comm.MasterClientServer;
+import org.apache.giraph.comm.NettyMasterClientServer;
import org.apache.giraph.graph.GraphMapper.MapFunctions;
import org.apache.giraph.graph.partition.MasterGraphPartitioner;
import org.apache.giraph.graph.partition.PartitionOwner;
@@ -139,6 +141,10 @@ public class BspServiceMaster<I extends
private AggregatorWriter aggregatorWriter;
/** Master class */
private MasterCompute masterCompute;
+ /** Communication service */
+ private MasterClientServer commService;
+ /** Master info */
+ private WorkerInfo masterInfo;
/** Limit locality information added to each InputSplit znode */
private final int localityLimit = 5;
@@ -790,7 +796,21 @@ public class BspServiceMaster<I extends
throw new IllegalStateException("becomeMaster: " +
"Couldn't initialize aggregatorWriter", e);
}
- LOG.info("becomeMaster: I am now the master!");
+
+ boolean useNetty = getConfiguration().getBoolean(GiraphJob.USE_NETTY,
+ GiraphJob.USE_NETTY_DEFAULT);
+ if (useNetty) {
+ commService = new NettyMasterClientServer(getContext());
+ masterInfo = new WorkerInfo(getHostname(), getTaskPartition(),
+ commService.getMyAddress().getPort());
+ // write my address to znode so workers could read it
+ WritableUtils.writeToZnode(getZkExt(), currentMasterPath, -1,
+ masterInfo);
+ }
+
+ if (LOG.isInfoEnabled()) {
+ LOG.info("becomeMaster: I am now the master!");
+ }
isMaster = true;
return isMaster;
}
@@ -1448,6 +1468,12 @@ public class BspServiceMaster<I extends
}
}
+ boolean useNetty = getConfiguration().getBoolean(GiraphJob.USE_NETTY,
+ GiraphJob.USE_NETTY_DEFAULT);
+ if (useNetty) {
+ commService.fixWorkerAddresses(chosenWorkerInfoList);
+ }
+
currentWorkersCounter.increment(chosenWorkerInfoList.size() -
currentWorkersCounter.getValue());
assignPartitionOwners(allPartitionStatsList,
@@ -1736,6 +1762,13 @@ public class BspServiceMaster<I extends
}
}
aggregatorWriter.close();
+
+ boolean useNetty = getConfiguration().getBoolean(GiraphJob.USE_NETTY,
+ GiraphJob.USE_NETTY_DEFAULT);
+ if (useNetty) {
+ commService.closeConnections();
+ commService.close();
+ }
}
try {
Modified: giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceWorker.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceWorker.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceWorker.java (original)
+++ giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceWorker.java Mon Aug 27 18:54:23 2012
@@ -101,6 +101,8 @@ public class BspServiceWorker<I extends
inputSplitCache = new HashMap<PartitionOwner, Partition<I, V, E, M>>();
/** Communication service */
private final WorkerClientServer<I, V, E, M> commService;
+ /** Master info */
+ private WorkerInfo masterInfo = new WorkerInfo();
/** Have the partition exchange children (workers) changed? */
private final BspEvent partitionExchangeChildrenChanged;
/** Regulates the size of outgoing Collections of vertices read
@@ -528,6 +530,11 @@ public class BspServiceWorker<I extends
}
@Override
+ public WorkerInfo getMasterInfo() {
+ return masterInfo;
+ }
+
+ @Override
public void setup() {
// Unless doing a restart, prepare for computation:
// 1. Start superstep INPUT_SUPERSTEP (no computation)
@@ -939,6 +946,15 @@ public class BspServiceWorker<I extends
"startSuperstep: InterruptedException getting assignments", e);
}
+
+ boolean useNetty = getConfiguration().getBoolean(GiraphJob.USE_NETTY,
+ GiraphJob.USE_NETTY_DEFAULT);
+ if (useNetty) {
+ // get address of master
+ WritableUtils.readFieldsFromZnode(getZkExt(), currentMasterPath, false,
+ null, masterInfo);
+ }
+
if (LOG.isInfoEnabled()) {
LOG.info("startSuperstep: Ready for computation on superstep " +
getSuperstep() + " since worker " +
Modified: giraph/trunk/src/test/java/org/apache/giraph/comm/ConnectionTest.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/test/java/org/apache/giraph/comm/ConnectionTest.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/test/java/org/apache/giraph/comm/ConnectionTest.java (original)
+++ giraph/trunk/src/test/java/org/apache/giraph/comm/ConnectionTest.java Mon Aug 27 18:54:23 2012
@@ -18,7 +18,10 @@
package org.apache.giraph.comm;
+import com.google.common.collect.Sets;
+import java.util.Set;
import org.apache.giraph.comm.messages.SimpleMessageStore;
+import org.apache.giraph.comm.RequestServerHandler.RequestServerHandlerFactory;
import org.apache.giraph.utils.MockUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
@@ -28,12 +31,9 @@ import org.junit.Test;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
-import com.google.common.collect.Sets;
-
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Collections;
-import java.util.Set;
/**
* Test the netty connections
@@ -55,14 +55,12 @@ public class ConnectionTest {
new ServerData<IntWritable, IntWritable, IntWritable, IntWritable>
(conf, SimpleMessageStore.newFactory(
MockUtils.mockServiceGetVertexPartitionOwner(1), conf));
- NettyServer<IntWritable, IntWritable, IntWritable, IntWritable> server =
- new NettyServer<IntWritable, IntWritable, IntWritable, IntWritable>(
- conf, serverData);
+ NettyServer server =
+ new NettyServer(conf,
+ new WorkerRequestServerHandler.Factory(serverData));
server.start();
- NettyClient<IntWritable, IntWritable, IntWritable, IntWritable> client =
- new NettyClient<IntWritable, IntWritable, IntWritable,
- IntWritable>(context);
+ NettyClient client = new NettyClient(context);
client.connectAllAddresses(Collections.singleton(server.getMyAddress()));
client.stop();
@@ -85,23 +83,17 @@ public class ConnectionTest {
new ServerData<IntWritable, IntWritable, IntWritable, IntWritable>
(conf, SimpleMessageStore.newFactory(
MockUtils.mockServiceGetVertexPartitionOwner(1), conf));
+ RequestServerHandlerFactory requestServerHandlerFactory =
+ new WorkerRequestServerHandler.Factory(serverData);
- NettyServer<IntWritable, IntWritable, IntWritable, IntWritable> server1 =
- new NettyServer<IntWritable, IntWritable, IntWritable, IntWritable>(
- conf, serverData);
+ NettyServer server1 = new NettyServer(conf, requestServerHandlerFactory);
server1.start();
- NettyServer<IntWritable, IntWritable, IntWritable, IntWritable> server2 =
- new NettyServer<IntWritable, IntWritable, IntWritable, IntWritable>(
- conf, serverData);
+ NettyServer server2 = new NettyServer(conf, requestServerHandlerFactory);
server2.start();
- NettyServer<IntWritable, IntWritable, IntWritable, IntWritable> server3 =
- new NettyServer<IntWritable, IntWritable, IntWritable, IntWritable>(
- conf, serverData);
+ NettyServer server3 = new NettyServer(conf, requestServerHandlerFactory);
server3.start();
- NettyClient<IntWritable, IntWritable, IntWritable, IntWritable> client =
- new NettyClient<IntWritable, IntWritable, IntWritable,
- IntWritable>(context);
+ NettyClient client = new NettyClient(context);
Set<InetSocketAddress> serverAddresses = Sets.newHashSet();
serverAddresses.add(server1.getMyAddress());
serverAddresses.add(server2.getMyAddress());
@@ -130,22 +122,15 @@ public class ConnectionTest {
new ServerData<IntWritable, IntWritable, IntWritable, IntWritable>
(conf, SimpleMessageStore.newFactory(
MockUtils.mockServiceGetVertexPartitionOwner(1), conf));
- NettyServer<IntWritable, IntWritable, IntWritable, IntWritable> server =
- new NettyServer<IntWritable, IntWritable, IntWritable, IntWritable>(
- conf, serverData);
+ NettyServer server = new NettyServer(conf,
+ new WorkerRequestServerHandler.Factory(serverData));
server.start();
- NettyClient<IntWritable, IntWritable, IntWritable, IntWritable> client1 =
- new NettyClient<IntWritable, IntWritable, IntWritable,
- IntWritable>(context);
+ NettyClient client1 = new NettyClient(context);
client1.connectAllAddresses(Collections.singleton(server.getMyAddress()));
- NettyClient<IntWritable, IntWritable, IntWritable, IntWritable> client2 =
- new NettyClient<IntWritable, IntWritable, IntWritable,
- IntWritable>(context);
+ NettyClient client2 = new NettyClient(context);
client2.connectAllAddresses(Collections.singleton(server.getMyAddress()));
- NettyClient<IntWritable, IntWritable, IntWritable, IntWritable> client3 =
- new NettyClient<IntWritable, IntWritable, IntWritable,
- IntWritable>(context);
+ NettyClient client3 = new NettyClient(context);
client3.connectAllAddresses(Collections.singleton(server.getMyAddress()));
client1.stop();
Modified: giraph/trunk/src/test/java/org/apache/giraph/comm/RequestFailureTest.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/test/java/org/apache/giraph/comm/RequestFailureTest.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/test/java/org/apache/giraph/comm/RequestFailureTest.java (original)
+++ giraph/trunk/src/test/java/org/apache/giraph/comm/RequestFailureTest.java Mon Aug 27 18:54:23 2012
@@ -53,11 +53,9 @@ public class RequestFailureTest {
private ServerData<IntWritable, IntWritable, IntWritable, IntWritable>
serverData;
/** Server */
- private NettyServer<IntWritable, IntWritable, IntWritable, IntWritable>
- server;
+ private NettyServer server;
/** Client */
- private NettyClient<IntWritable, IntWritable, IntWritable, IntWritable>
- client;
+ private NettyClient client;
/** Mock context */
private Context context;
@@ -89,8 +87,7 @@ public class RequestFailureTest {
when(context.getConfiguration()).thenReturn(conf);
}
- private WritableRequest<IntWritable, IntWritable, IntWritable,
- IntWritable> getRequest() {
+ private WritableRequest getRequest() {
// Data to send
int partitionId = 17;
Map<IntWritable, Collection<IntWritable>> vertexIdMessages =
@@ -139,20 +136,15 @@ public class RequestFailureTest {
new ServerData<IntWritable, IntWritable, IntWritable, IntWritable>
(conf, SimpleMessageStore.newFactory(
MockUtils.mockServiceGetVertexPartitionOwner(1), conf));
- server =
- new NettyServer<IntWritable, IntWritable, IntWritable, IntWritable>(
- conf, serverData);
+ server = new NettyServer(conf,
+ new WorkerRequestServerHandler.Factory(serverData));
server.start();
- client =
- new NettyClient<IntWritable, IntWritable, IntWritable, IntWritable>
- (context);
+ client = new NettyClient(context);
client.connectAllAddresses(Collections.singleton(server.getMyAddress()));
// Send the request 2x
- WritableRequest<IntWritable, IntWritable, IntWritable,
- IntWritable> request1 = getRequest();
- WritableRequest<IntWritable, IntWritable, IntWritable,
- IntWritable> request2 = getRequest();
+ WritableRequest request1 = getRequest();
+ WritableRequest request2 = getRequest();
client.sendWritableRequest(-1, server.getMyAddress(), request1);
client.sendWritableRequest(-1, server.getMyAddress(), request2);
client.waitAllRequests();
@@ -179,20 +171,15 @@ public class RequestFailureTest {
new ServerData<IntWritable, IntWritable, IntWritable, IntWritable>
(conf, SimpleMessageStore.newFactory(
MockUtils.mockServiceGetVertexPartitionOwner(1), conf));
- server =
- new NettyServer<IntWritable, IntWritable, IntWritable, IntWritable>(
- conf, serverData);
+ server = new NettyServer(conf,
+ new WorkerRequestServerHandler.Factory(serverData));
server.start();
- client =
- new NettyClient<IntWritable, IntWritable, IntWritable, IntWritable>
- (context);
+ client = new NettyClient(context);
client.connectAllAddresses(Collections.singleton(server.getMyAddress()));
// Send the request 2x, but should only be processed once
- WritableRequest<IntWritable, IntWritable, IntWritable,
- IntWritable> request1 = getRequest();
- WritableRequest<IntWritable, IntWritable, IntWritable,
- IntWritable> request2 = getRequest();
+ WritableRequest request1 = getRequest();
+ WritableRequest request2 = getRequest();
client.sendWritableRequest(-1, server.getMyAddress(), request1);
client.sendWritableRequest(-1, server.getMyAddress(), request2);
client.waitAllRequests();
@@ -219,20 +206,15 @@ public class RequestFailureTest {
new ServerData<IntWritable, IntWritable, IntWritable, IntWritable>
(conf, SimpleMessageStore.newFactory(
MockUtils.mockServiceGetVertexPartitionOwner(1), conf));
- server =
- new NettyServer<IntWritable, IntWritable, IntWritable, IntWritable>(
- conf, serverData);
+ server = new NettyServer(conf,
+ new WorkerRequestServerHandler.Factory(serverData));
server.start();
- client =
- new NettyClient<IntWritable, IntWritable, IntWritable, IntWritable>
- (context);
+ client = new NettyClient(context);
client.connectAllAddresses(Collections.singleton(server.getMyAddress()));
// Send the request 2x, but should only be processed once
- WritableRequest<IntWritable, IntWritable, IntWritable,
- IntWritable> request1 = getRequest();
- WritableRequest<IntWritable, IntWritable, IntWritable,
- IntWritable> request2 = getRequest();
+ WritableRequest request1 = getRequest();
+ WritableRequest request2 = getRequest();
client.sendWritableRequest(-1, server.getMyAddress(), request1);
client.sendWritableRequest(-1, server.getMyAddress(), request2);
client.waitAllRequests();
Modified: giraph/trunk/src/test/java/org/apache/giraph/comm/RequestTest.java
URL: http://svn.apache.org/viewvc/giraph/trunk/src/test/java/org/apache/giraph/comm/RequestTest.java?rev=1377792&r1=1377791&r2=1377792&view=diff
==============================================================================
--- giraph/trunk/src/test/java/org/apache/giraph/comm/RequestTest.java (original)
+++ giraph/trunk/src/test/java/org/apache/giraph/comm/RequestTest.java Mon Aug 27 18:54:23 2012
@@ -60,11 +60,9 @@ public class RequestTest {
private ServerData<IntWritable, IntWritable, IntWritable, IntWritable>
serverData;
/** Server */
- private NettyServer<IntWritable, IntWritable, IntWritable, IntWritable>
- server;
+ private NettyServer server;
/** Client */
- private NettyClient<IntWritable, IntWritable, IntWritable, IntWritable>
- client;
+ private NettyClient client;
/**
* Only for testing.
@@ -99,13 +97,10 @@ public class RequestTest {
new ServerData<IntWritable, IntWritable, IntWritable, IntWritable>
(conf, SimpleMessageStore.newFactory(
MockUtils.mockServiceGetVertexPartitionOwner(1), conf));
- server =
- new NettyServer<IntWritable, IntWritable, IntWritable, IntWritable>(
- conf, serverData);
+ server = new NettyServer(conf,
+ new WorkerRequestServerHandler.Factory(serverData));
server.start();
- client =
- new NettyClient<IntWritable, IntWritable, IntWritable, IntWritable>
- (context);
+ client = new NettyClient(context);
client.connectAllAddresses(Collections.singleton(server.getMyAddress()));
}