flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From trohrm...@apache.org
Subject [33/50] [abbrv] flink git commit: [FLINK-4658] [rpc] Allow RpcService to execute Runnables and Callables in its executor
Date Sun, 02 Oct 2016 21:58:38 GMT
[FLINK-4658] [rpc] Allow RpcService to execute Runnables and Callables in its executor

This closes #2531.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9b22d661
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9b22d661
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9b22d661

Branch: refs/heads/flip-6
Commit: 9b22d6610650cbc14e540b067467a80648ecd982
Parents: 88a05f5
Author: Till Rohrmann <trohrmann@apache.org>
Authored: Wed Sep 21 18:16:27 2016 +0200
Committer: Till Rohrmann <trohrmann@apache.org>
Committed: Sun Oct 2 23:44:45 2016 +0200

----------------------------------------------------------------------
 .../apache/flink/runtime/rpc/RpcService.java    | 31 ++++++++++++++
 .../flink/runtime/rpc/akka/AkkaRpcService.java  | 14 ++++++
 .../flink/runtime/rpc/AsyncCallsTest.java       |  1 -
 .../runtime/rpc/TestingSerialRpcService.java    | 16 +++++++
 .../runtime/rpc/akka/AkkaRpcServiceTest.java    | 45 ++++++++++++++++++++
 5 files changed, 106 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/9b22d661/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java
index a367ff2..437e08b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.rpc;
 import org.apache.flink.runtime.concurrent.Future;
 import org.apache.flink.runtime.rpc.exceptions.RpcConnectionException;
 
+import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
@@ -89,4 +90,34 @@ public interface RpcService {
 	 * @param delay    The delay after which the runnable will be executed
 	 */
 	void scheduleRunnable(Runnable runnable, long delay, TimeUnit unit);
+
+	/**
+	 * Execute the given runnable in the executor of the RPC service. This method can be used
to run
+	 * code outside of the main thread of a {@link RpcEndpoint}.
+	 *
+	 * <p><b>IMPORTANT:</b> This executor does not isolate the method invocations
against
+	 * any concurrent invocations and is therefore not suitable to run completion methods of
futures
+	 * that modify state of an {@link RpcEndpoint}. For such operations, one needs to use the
+	 * {@link RpcEndpoint#getMainThreadExecutor() MainThreadExecutionContext} of that
+	 * {@code RpcEndpoint}.
+	 *
+	 * @param runnable to execute
+	 */
+	void execute(Runnable runnable);
+
+	/**
+	 * Execute the given callable and return its result as a {@link Future}. This method can
be used
+	 * to run code outside of the main thread of a {@link RpcEndpoint}.
+	 *
+	 * <p><b>IMPORTANT:</b> This executor does not isolate the method invocations
against
+	 * any concurrent invocations and is therefore not suitable to run completion methods of
futures
+	 * that modify state of an {@link RpcEndpoint}. For such operations, one needs to use the
+	 * {@link RpcEndpoint#getMainThreadExecutor() MainThreadExecutionContext} of that
+	 * {@code RpcEndpoint}.
+	 *
+	 * @param callable to execute
+	 * @param <T> is the return value type
+	 * @return Future containing the callable's future result
+	 */
+	<T> Future<T> execute(Callable<T> callable);
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/9b22d661/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java
index 36f1115..cee19c4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java
@@ -25,6 +25,7 @@ import akka.actor.ActorSystem;
 import akka.actor.Identify;
 import akka.actor.PoisonPill;
 import akka.actor.Props;
+import akka.dispatch.Futures;
 import akka.dispatch.Mapper;
 
 import akka.pattern.Patterns;
@@ -48,6 +49,7 @@ import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Proxy;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
@@ -223,4 +225,16 @@ public class AkkaRpcService implements RpcService {
 
 		actorSystem.scheduler().scheduleOnce(new FiniteDuration(delay, unit), runnable, actorSystem.dispatcher());
 	}
+
+	@Override
+	public void execute(Runnable runnable) {
+		actorSystem.dispatcher().execute(runnable);
+	}
+
+	@Override
+	public <T> Future<T> execute(Callable<T> callable) {
+		scala.concurrent.Future<T> scalaFuture = Futures.future(callable, actorSystem.dispatcher());
+
+		return new FlinkFuture<>(scalaFuture);
+	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/9b22d661/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java
index 7c6b0ee..e8255d4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java
@@ -19,7 +19,6 @@
 package org.apache.flink.runtime.rpc;
 
 import akka.actor.ActorSystem;
-import akka.util.Timeout;
 
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.core.testutils.OneShotLatch;

http://git-wip-us.apache.org/repos/asf/flink/blob/9b22d661/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/TestingSerialRpcService.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/TestingSerialRpcService.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/TestingSerialRpcService.java
index 957453a..c58ea20 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/TestingSerialRpcService.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/TestingSerialRpcService.java
@@ -65,6 +65,22 @@ public class TestingSerialRpcService implements RpcService {
 	}
 
 	@Override
+	public void execute(Runnable runnable) {
+		runnable.run();
+	}
+
+	@Override
+	public <T> Future<T> execute(Callable<T> callable) {
+		try {
+			T result = callable.call();
+
+			return FlinkCompletableFuture.completed(result);
+		} catch (Exception e) {
+			return FlinkCompletableFuture.completedExceptionally(e);
+		}
+	}
+
+	@Override
 	public Executor getExecutor() {
 		return executorService;
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/9b22d661/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceTest.java
index 4e9e518..5550cb5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceTest.java
@@ -22,13 +22,18 @@ import akka.actor.ActorSystem;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.core.testutils.OneShotLatch;
 import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.concurrent.Future;
 import org.apache.flink.util.TestLogger;
 
 import org.junit.AfterClass;
 import org.junit.Test;
 
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 public class AkkaRpcServiceTest extends TestLogger {
@@ -70,4 +75,44 @@ public class AkkaRpcServiceTest extends TestLogger {
 
 		assertTrue("call was not properly delayed", ((stop - start) / 1000000) >= delay);
 	}
+
+	/**
+	 * Tests that the {@link AkkaRpcService} can execute runnables
+	 */
+	@Test
+	public void testExecuteRunnable() throws Exception {
+		final OneShotLatch latch = new OneShotLatch();
+
+		akkaRpcService.execute(new Runnable() {
+			@Override
+			public void run() {
+				latch.trigger();
+			}
+		});
+
+		latch.await(30L, TimeUnit.SECONDS);
+	}
+
+	/**
+	 * Tests that the {@link AkkaRpcService} can execute callables and returns their result
as
+	 * a {@link Future}.
+	 */
+	@Test
+	public void testExecuteCallable() throws InterruptedException, ExecutionException, TimeoutException
{
+		final OneShotLatch latch = new OneShotLatch();
+		final int expected = 42;
+
+		Future<Integer> result = akkaRpcService.execute(new Callable<Integer>() {
+			@Override
+			public Integer call() throws Exception {
+				latch.trigger();
+				return expected;
+			}
+		});
+
+		int actual = result.get(30L, TimeUnit.SECONDS);
+
+		assertEquals(expected, actual);
+		assertTrue(latch.isTriggered());
+	}
 }


Mime
View raw message