add transport server, client, encode, decode

computer-core/pom.xml

@@ -24,6 +24,10 @@
  <groupId>com.baidu.hugegraph</groupId>
  <artifactId>hugegraph-client</artifactId>
  </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-all</artifactId>
+ </dependency>
  <dependency>
  <groupId>io.etcd</groupId>
  <artifactId>jetcd-core</artifactId>

computer-core/src/main/java/com/baidu/hugegraph/computer/core/common/exception/TransportException.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2017 HugeGraph Authors
+ *
+ * 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 com.baidu.hugegraph.computer.core.common.exception;
+
+import java.io.IOException;
+
+/**
+ * The transport network exception
+ */
+public class TransportException extends IOException {
+
+ private static final long serialVersionUID = -6538118382075299762L;
+
+ public static final int DEFAULT_CODE = 0;
+
+ private final int errorCode;
+
+ public TransportException(String message) {
+ this(DEFAULT_CODE, message);
+ }
+
+ public TransportException(int errorCode, String message) {
+ super(message);
+ this.errorCode = errorCode;
+ }
+
+ public TransportException(String message, Object... args) {
+ this(DEFAULT_CODE, message, args);
+ }
+
+ public TransportException(int errorCode, String message, Object... args) {
+ super(String.format(message, args));
+ this.errorCode = errorCode;
+ }
+
+ public TransportException(String message, Throwable cause) {
+ super(message, cause);
+ this.errorCode = DEFAULT_CODE;
+ }
+
+ public TransportException(String message, Throwable cause, Object... args) {
+ super(String.format(message, args), cause);
+ this.errorCode = DEFAULT_CODE;
+ }
+
+ public Throwable rootCause() {
+ return ComputeException.rootCause(this);
+ }
+
+ public int errorCode() {
+ return this.errorCode;
+ }
+}

computer-core/src/main/java/com/baidu/hugegraph/computer/core/config/ComputerOptions.java

@@ -21,6 +21,7 @@
 
 import static com.baidu.hugegraph.config.OptionChecker.allowValues;
 import static com.baidu.hugegraph.config.OptionChecker.disallowEmpty;
+import static com.baidu.hugegraph.config.OptionChecker.nonNegativeInt;
 import static com.baidu.hugegraph.config.OptionChecker.positiveInt;
 
 import java.util.Set;
@@ -30,6 +31,8 @@
 import com.baidu.hugegraph.computer.core.aggregator.WorkerAggrManager;
 import com.baidu.hugegraph.computer.core.graph.partition.HashPartitioner;
 import com.baidu.hugegraph.computer.core.master.DefaultMasterComputation;
+import com.baidu.hugegraph.computer.core.network.NettyTransportProvider;
+import com.baidu.hugegraph.computer.core.network.TransportConf;
 import com.baidu.hugegraph.config.ConfigOption;
 import com.baidu.hugegraph.config.OptionHolder;
 import com.google.common.collect.ImmutableSet;
@@ -241,26 +244,6 @@ public static synchronized ComputerOptions instance() {
  TimeUnit.SECONDS.toMillis(30L)
  );
 
- public static final ConfigOption<Integer> WORKER_DATA_PORT_START =
- new ConfigOption<>(
- "worker.data_port_start",
- "The start of range [data_port_start, data_port_end]. " +
- "The worker will choose one from small to large of the " +
- "range for data transportation.",
- positiveInt(),
- 11000
- );
-
- public static final ConfigOption<Integer> WORKER_DATA_PORT_END =
- new ConfigOption<>(
- "worker.data_port_end",
- "The end of range [data_port_start, data_port_end]. " +
- "The worker will choose one from small to large of the " +
- "range for data transportation.",
- positiveInt(),
- 12000
- );
-
  public static final ConfigOption<Class<?>> WORKER_PARTITIONER =
  new ConfigOption<>(
  "worker.partitioner",
@@ -333,4 +316,177 @@ public static synchronized ComputerOptions instance() {
  disallowEmpty(),
  "hugegraph"
  );
+
+ public static final ConfigOption<String> TRANSPORT_SERVER_HOST =
+ new ConfigOption<>(
+ "transport.server_host",
+ "The server bind host.",
+ disallowEmpty(),
+ "127.0.0.1"
+ );
+
+ public static final ConfigOption<Integer> TRANSPORT_SERVER_PORT =
+ new ConfigOption<>(
+ "transport.server_port",
+ "The server bind port, if it is zero " +
+ "will let the system pick up an ephemeral port.",
+ nonNegativeInt(),
+ 0
+ );
+
+ public static final ConfigOption<Integer> TRANSPORT_SERVER_THREADS =
+ new ConfigOption<>(
+ "transport.server_threads",
+ "The number of transport threads for server, the default " +
+ "value is CPUs.",
+ positiveInt(),
+ TransportConf.NUMBER_CPU_CORES
+ );
+
+ public static final ConfigOption<Integer> TRANSPORT_CLIENT_THREADS =
+ new ConfigOption<>(
+ "transport.client_threads",
+ "The number of transport threads for client, the default " +
+ "value is CPUs.",
+ positiveInt(),
+ TransportConf.NUMBER_CPU_CORES
+ );
+
+ public static final ConfigOption<Class<?>> TRANSPORT_PROVIDER_CLASS =
+ new ConfigOption<>(
+ "transport.provider_class",
+ "The transport provider, currently only supports Netty.",
+ disallowEmpty(),
+ NettyTransportProvider.class
+ );
+
+ public static final ConfigOption<String> TRANSPORT_IO_MODE =
+ new ConfigOption<>(
+ "transport.io_mode",
+ "The network IO Mode, either 'NIO', 'EPOLL', 'AUTO', the " +
+ "'AUTO' means selecting the property mode automatically.",
+ allowValues("NIO", "EPOLL", "AUTO"),
+ "AUTO"
+ );
+
+ public static final ConfigOption<Boolean> TRANSPORT_EPOLL_LT =
+ new ConfigOption<>(
+ "transport.transport_epoll_lt",
+ "Whether enable EPOLL level-trigger.",
+ allowValues(true, false),
+ false
+ );
+
+ public static final ConfigOption<Integer> TRANSPORT_SEND_BUFFER_SIZE =
+ new ConfigOption<>(
+ "transport.send_buffer_size",
+ "The network send buffer size, 0 means using system " +
+ "defaults.",
+ nonNegativeInt(),
+ 0
+ );
+
+ public static final ConfigOption<Integer> TRANSPORT_RECEIVE_BUFFER_SIZE =
+ new ConfigOption<>(
+ "transport.receive_buffer_size",
+ "The network receive buffer size, 0 means using system " +
+ "defaults.",
+ nonNegativeInt(),
+ 0
+ );
+
+ public static final ConfigOption<Integer> TRANSPORT_BACKLOG =
+ new ConfigOption<>(
+ "transport.backlog",
+ "The server connection backlog, 0 means using system " +
+ "defaults.",
+ nonNegativeInt(),
+ 0
+ );
+
+ public static final ConfigOption<Long> TRANSPORT_CLIENT_CONNECT_TIMEOUT =
+ new ConfigOption<>(
+ "transport.client_connect_timeout",
+ "The timeout(in ms) of client connect to server.",
+ positiveInt(),
+ 3000L
+ );
+
+ public static final ConfigOption<Long> TRANSPORT_CLOSE_TIMEOUT =
+ new ConfigOption<>(
+ "transport.close_timeout",
+ "The timeout(in ms) of close server or close client.",
+ positiveInt(),
+ 10_000L
+ );
+
+ public static final ConfigOption<Long> TRANSPORT_SYNC_REQUEST_TIMEOUT =
+ new ConfigOption<>(
+ "transport.sync_request_timeout",
+ "The timeout(in ms) to wait response after " +
+ "sending sync-request.",
+ positiveInt(),
+ 5_000L
+ );
+
+ public static final ConfigOption<Integer> TRANSPORT_NETWORK_RETRIES =
+ new ConfigOption<>(
+ "transport.network_retries",
+ "The number of retry attempts for network communication," +
+ "if network unstable.",
+ nonNegativeInt(),
+ 3
+ );
+
+ public static final ConfigOption<Integer> TRANSPORT_MAX_PENDING_REQUESTS =
+ new ConfigOption<>(
+ "transport.max_pending_requests",
+ "The max number of client unreceived ack, " +
+ "if the number of unreceived ack greater than it, " +
+ "it will block the client from calling send.",
+ positiveInt(),
+ 50
+ );
+
+ public static final ConfigOption<Integer> TRANSPORT_MIN_PENDING_REQUESTS =
+ new ConfigOption<>(
+ "transport.min_pending_requests",
+ "The minimum number of client unreceived ack, " +
+ "if the number of unreceived ack less than it, " +
+ "it will wake the client from calling send.",
+ positiveInt(),
+ 5
+ );
+
+ public static final ConfigOption<Long> TRANSPORT_MIN_ACK_INTERVAL =
+ new ConfigOption<>(
+ "transport.min_ack_interval",
+ "The minimum interval(in ms) of server reply ack.",
+ positiveInt(),
+ 200L
+ );
+
+ public static final ConfigOption<Integer> TRANSPORT_HEARTBEAT_INTERVAL =
+ new ConfigOption<>(
+ "transport.heartbeat_interval_seconds",
+ "Time minimum interval(in seconds) of send heartbeat.",
+ positiveInt(),
+ 60
+ );
+
+ public static final ConfigOption<Integer> TRANSPORT_HEARTBEAT_TIMEOUT =
+ new ConfigOption<>(
+ "transport.heartbeat_timeout_seconds",
+ "The max timeout(in seconds) of heartbeat.",
+ positiveInt(),
+ 120
+ );
+
+ public static final ConfigOption<Boolean> TRANSPORT_TCP_KEEP_ALIVE =
+ new ConfigOption<>(
+ "transport.transport_tcp_keep_alive",
+ "Whether enable TCP keep-alive.",
+ allowValues(true, false),
+ true
+ );
 }

computer-core/src/main/java/com/baidu/hugegraph/computer/core/network/ClientFactory.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2017 HugeGraph Authors
+ *
+ * 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 com.baidu.hugegraph.computer.core.network;
+
+
+import com.baidu.hugegraph.computer.core.common.exception.TransportException;
+
+public interface ClientFactory {
+
+ /**
+ * Initialize the {@link ClientFactory}
+ */
+ void init();
+
+ /**
+ * Create a TransportClient.
+ * @param connectionId {@link ConnectionId}
+ * @param handler
+ * @return {@link TransportClient}
+ */
+ TransportClient createClient(ConnectionId connectionId,
+ ClientHandler handler)
+ throws TransportException;
+
+ /**
+ * Close the {@link ClientFactory}
+ */
+ void close();
+}

computer-core/src/main/java/com/baidu/hugegraph/computer/core/network/Transport4Server.java → computer-core/src/main/java/com/baidu/hugegraph/computer/core/network/ClientHandler.java

@@ -19,22 +19,11 @@
 
 package com.baidu.hugegraph.computer.core.network;
 
-import com.baidu.hugegraph.computer.core.config.Config;
-
-/**
- * This is used for worker that receives data.
- */
-public interface Transport4Server {
-
- /**
- * Startup server, return the port listened. The port range in config is
- * [{@link @ComputerOptions.WORKER_DATA_PORT_START #STATIC_FIELD},
- * {@link @ComputerOptions.WORKER_DATA_PORT_END #STATIC_FIELD}].
- */
- int listen(Config config, MessageHandler handler);
+public interface ClientHandler extends TransportHandler {
 
  /**
- * Stop the server
+ * Invoked when the channel associated with the given connectionId channel
+ * is able to send data immediately.
  */
- void stop();
+ void sendAvailable(ConnectionId connectionId);
 }