Add TcpChannel to unify Transport implementations by Tim-Brooks · Pull Request #27132 · elastic/elasticsearch

Tim-Brooks · 2017-10-26T21:16:48Z

Right now our different transport implementations must duplicate
functionality in order to stay compliant with the requirements of
TcpTransport. They must all implement common logic to open channels,
close channels, keep track of channels for eventual shutdown, etc.

Additionally, there is a weird and complicated relationship between
Transport and TransportService. We eventually want to start merging
some of the functionality between these classes.

This commit starts moving towards a world where TransportService retains
all the application logic and channel state. Transport implementations
in this world will only be tasked with returning a channel when one is
requested, calling transport service when a channel is accepted from
a server, and starting / stopping itself.

Specifically this commit changes how channels are opened and closed. All
Transport implementations now return a channel type that must comply with
the new TcpChannel interface. This interface has the methods necessary
for TcpTransport to completely manage the lifecycle of a channel. This
includes setting the channel up, waiting for connection, adding close
listeners, and eventually closing.

s1monw

I did an initial pass. I like it. left some comments

s1monw · 2017-10-27T09:03:16Z

core/src/main/java/org/elasticsearch/transport/PlainChannelFuture.java

+
+import org.elasticsearch.action.support.PlainActionFuture;
+
+public class PlainChannelFuture<Channel> extends PlainActionFuture<Channel> {


can we have javadocs for this? and can it be final

s1monw · 2017-10-27T09:04:45Z

core/src/main/java/org/elasticsearch/transport/TcpChannel.java

+
+    ListenableActionFuture<C> closeAsync();
+
+    ListenableActionFuture<C> getCloseFuture();


what is this doing?

At times we need to attach a close listener. And the future allows us to do that. But I presume that you want just a method to add close listeners instead.

s1monw · 2017-10-27T09:04:48Z

core/src/main/java/org/elasticsearch/transport/TcpChannel.java

+
+public interface TcpChannel<C extends TcpChannel<C>> {
+
+    ListenableActionFuture<C> closeAsync();


can we turn this around and accept a listener isntead? I really don't like returning futures it's a design flaw IMO

I'm not sure I understand here. We need something to block on at shutdown. Are you suggesting this?

PlainActionFuture<C> future = PlainActionFuture.newFuture(); channel.addCloseListener(future); channel.closeAsync(); future.actionGet();

Or some callback with a CountDownLatch?

CountDownLatch latch = new CountDownLatch(channels.size()); for (final C channel : channels) { channel.addCloseListener(ActionListener.wrap(latch::countDown)); } latch.await();

I think the simplest options would be this:

void close(ActionListener<Void> onClosed);

s1monw · 2017-10-27T09:08:55Z

core/src/main/java/org/elasticsearch/transport/TcpChannel.java

+
+import java.io.IOException;
+
+public interface TcpChannel<C extends TcpChannel<C>> {


I assume that down the road we will get rid of the generics here and add all the necessary method to it that TcpTransport needs, right?

s1monw · 2017-10-27T09:11:23Z

core/src/main/java/org/elasticsearch/transport/TcpTransport.java


        public List<Channel> getChannels() {
-            return Arrays.asList(channels);
+            return new ArrayList<>(channels);


Can we wrap this in the ctor in a Collections#unmodifiableList and just return the instance instead of a mutable copy

Yep good idea. I should use the unmodifiableList in a few other places (in NodeChannels) too.

s1monw · 2017-10-27T09:13:39Z

core/src/main/java/org/elasticsearch/transport/TcpTransport.java

+                List<PlainChannelFuture<Channel>> pendingChannels = new ArrayList<>(numConnections);
+                for (int i = 0; i < numConnections; ++i) {
+                    try {
+                        PlainChannelFuture<Channel> pending = initiateChannel(node, connectionProfile.getConnectTimeout());


I don't like that this returns a future. I am almost certain it's simpler if we go with a callback here. I havent' seen a useage of future where it was actually helpful or simplifying anything.

I guess same question as above. Are you imaging something where I pass in a CountDownLatch as a callback or a future as a callback? At some point we need to block while waiting for the connections to be completed (without changing our architecture in a pretty major way).

s1monw · 2017-10-27T09:14:58Z

core/src/main/java/org/elasticsearch/transport/TcpTransport.java


-    /** parse a hostname+port range spec into its equivalent addresses */
+    /**
+     * parse a hostname+port range spec into its equivalent addresses


autoformat. I'll fix in next revision.

…nection

Tim-Brooks · 2017-11-03T15:20:40Z

This is ready for another review. I have address @s1monw suggested changes.

s1monw

this looks great! good stuff! I left some comments

s1monw · 2017-11-13T10:37:02Z

core/src/main/java/org/elasticsearch/transport/TcpChannel.java

+
+import java.io.IOException;
+
+public interface TcpChannel<C extends TcpChannel<C>> {


this needs javadocs :)

s1monw · 2017-11-13T10:37:46Z

core/src/main/java/org/elasticsearch/transport/TcpChannel.java

+
+public interface TcpChannel<C extends TcpChannel<C>> {
+
+    void closeAsync();


I wonder if this can simply implement AutoCloseable and we document that is might be async and the close listeners will be called upon closing.

s1monw · 2017-11-13T10:46:30Z

core/src/main/java/org/elasticsearch/action/support/ListenerExecutionContext.java

+
+    @SuppressWarnings("unchecked")
+    private void internalAddListener(ActionListener<V> listener) {
+        listeners.offer(listener);


we have a race here. We might execute this listener twice. I think we should keep it simple and just use synchronize and make a fully copy of the list every time we add a listener. I don't think this is perf critical. Let's make the state only mutable under a mutex otherwise it's too complicated. That way you can also just use ActionListener#onResponse and ActionListener#onFailure

s1monw · 2017-11-13T10:47:19Z

core/src/main/java/org/elasticsearch/transport/TcpChannelUtils.java

+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class TcpChannelUtils {


can we move these static methods on to TcpChannel I don't think we need this util class?

s1monw · 2017-11-13T10:48:22Z

core/src/main/java/org/elasticsearch/transport/TcpChannelUtils.java

+
+public class TcpChannelUtils {
+
+    public static <C extends TcpChannel<C>> void closeChannel(C channel, boolean blocking) {


can we just delegate to closeChannels with a Collections#singletonList or can we even do it on the caller side?

s1monw · 2017-11-13T10:57:50Z

core/src/main/java/org/elasticsearch/transport/TcpTransport.java

            ensureOpen();
            try {
+                Exception connectionException = null;
+                int numConnections = connectionProfile.getNumConnections();


can we assert that this is > 0?

s1monw · 2017-11-13T10:59:36Z

core/src/main/java/org/elasticsearch/transport/TcpTransport.java

+                serverChannels.clear();
+
+                // close all of the incoming channels
+                TcpChannelUtils.closeChannels(new ArrayList<>(acceptedChannels), true);


can you leave a comment why we copy this list / set

s1monw · 2017-11-13T11:01:54Z

core/src/main/java/org/elasticsearch/transport/TcpTransport.java

+    }
+
+    protected void serverAcceptedChannel(Channel channel) {
+        if (acceptedChannels.add(channel)) {


I am not sure why we have this conditional. Do we deduplicate? do we have duplicates? I think we should not do this but rather make sure we assert on it or throw a hard exception? I also wonder if we should install the close listener first otherwise it might not be invoked in the case of a race?

I agree that the channel should not have be attempted to be added twice. So I added an assertion.

I actually disagree on the other point. The only way to avoid a race is to add the close listener afterwards. Imagine that the channel is already closed. We add a close listener that removes the channel and it is immediately executed. Then we add the channel to the accepted set. Nothing removes it at this point.

Close listeners are always invoked once they are added. Even if they are added after a channel is closed (they are invoked immediately). So it is safe to add the close listener after the channel is added to the accepted set, because it will ensure that the channel is actually removed.

s1monw · 2017-11-13T11:02:21Z

core/src/main/java/org/elasticsearch/transport/TcpTransport.java

        pendingHandshakes.put(requestId, handler);
        boolean success = false;
        try {
-            if (isOpen(channel) == false) {


I love it! :) this is really much better

s1monw · 2017-11-13T11:02:54Z

core/src/main/java/org/elasticsearch/transport/TcpTransportChannel.java

 import java.util.concurrent.atomic.AtomicBoolean;

-public final class TcpTransportChannel<Channel> implements TransportChannel {
+public final class TcpTransportChannel<Channel extends TcpChannel<Channel>> implements TransportChannel {


at some point we need to get rid of the generics here. I wonder why we still need it but this is a followup...

…nection

Tim-Brooks · 2017-11-13T20:12:24Z

@s1monw I have made changes based on your feedback. I was able to remove some of the parameterization. But more should be able to be removed in followup PRs.

I did disagree with this:

we have a race here. We might execute this listener twice. I think we should keep it simple and just use synchronize and make a fully copy of the list every time we add a listener. I don't think this is perf critical. Let's make the state only mutable under a mutex otherwise it's too complicated. That way you can also just use ActionListener#onResponse and ActionListener#onFailure

Can you articulate the race? I do not think there is one. This pattern is pretty common and I use it a number of places. I attempt to enqueue something with a concurrent queue. After enqueuing I check a volatile variable to see if this context has been completed. If it has not been completed this listener will be dealt with by the completing thread. If the context has been completed this thread will attempt to remove the listener. If the listener is removed, we deal with it and the completing thread will never see it. If the completing thread removes it from the queue first, it will deal with the listener and we can move on. Only one thread can remove it from the queue.

Pretty much all of the complicated concurrency work is provided by the java std lib concurrent queue.

Additionally, unless you execute all of the listeners in a synchronized block the logic is still pretty complicated using synchronization. (As an aside, after this PR we are not really using the PlainListenableActionFuture anymore. We could probably remove that as I think you prefer to not use the future api?)

And you say that this is not a performance sensitive area of the code but this is called on the network thread. Using synchronization risks the network thread being parked while waiting for another thread to attach a listener. I would prefer to avoid that risk with using two java std lib utilities (atomic reference and concurrent queue) and a reasonably easy to understand concurrency pattern.

I did add a comment to the code explaining what is going on.

Thoughts?

Tim-Brooks · 2017-11-13T21:57:00Z

@s1monw Actually it looks like CompletableFuture provides the thing that I was looking for (multiple listeners without synchronization 😀) and is in the std lib. So I just deleted the ListenerExecutionContext and used that.

s1monw

LGTM

s1monw · 2017-11-14T09:51:08Z

core/src/main/java/org/elasticsearch/action/ActionListener.java

+                listener.onResponse(tcpChannel);
+            } else {
+                if (throwable instanceof Exception) {
+                    listener.onFailure((Exception) throwable);


I don't think we should hide the non-exception. this will be an error and in this case we need to rethrow it? @jasontedor WDYT?

s1monw · 2017-11-14T09:52:19Z

core/src/main/java/org/elasticsearch/action/ActionListener.java

+     * @param <Response> the type of the response
+     * @return a listener that listens for responses and invokes the runnable when received
+     */
+    static <Response> ActionListener<Response> wrap(Runnable runnable) {


maybe just return wrap(runnable:run, e -> runnable.run());?

s1monw · 2017-11-14T09:59:44Z

Can you articulate the race? I do not think there is one. This pattern is pretty common and I use it a number of places. I attempt to enqueue something with a concurrent queue. After enqueuing I check a volatile variable to see if this context has been completed. If it has not been completed this listener will be dealt with by the completing thread. If the context has been completed this thread will attempt to remove the listener. If the listener is removed, we deal with it and the completing thread will never see it. If the completing thread removes it from the queue first, it will deal with the listener and we can move on. Only one thread can remove it from the queue.

you are right this is correct. I didn't anticipate / overlooked that poll removes. Yet, I like the solution you have now better :)

jasontedor

@s1monw did the heavy lifting in the review phase here but this LGTM too.

…nection

Right now our different transport implementations must duplicate functionality in order to stay compliant with the requirements of TcpTransport. They must all implement common logic to open channels, close channels, keep track of channels for eventual shutdown, etc. Additionally, there is a weird and complicated relationship between Transport and TransportService. We eventually want to start merging some of the functionality between these classes. This commit starts moving towards a world where TransportService retains all the application logic and channel state. Transport implementations in this world will only be tasked with returning a channel when one is requested, calling transport service when a channel is accepted from a server, and starting / stopping itself. Specifically this commit changes how channels are opened and closed. All Transport implementations now return a channel type that must comply with the new TcpChannel interface. This interface has the methods necessary for TcpTransport to completely manage the lifecycle of a channel. This includes setting the channel up, waiting for connection, adding close listeners, and eventually closing.

This commit is a follow up to the work completed in elastic#27132. Essentially it transitions two more methods (sendMessage and getLocalAddress) from Transport to TcpChannel. With this change, there is no longer a need for TcpTransport to be aware of the specific type of channel a transport returns. So that class is no longer parameterized by channel type.

This is a follow up to elastic#27132. As that PR greatly simplified the connection logic inside a low level transport implementation, much of the functionality provided by the NioClient class is no longer necessary. This commit removes that class.

* master: Stop skipping REST test after backport of #27056 Fix default value of ignore_unavailable for snapshot REST API (#27056) Add composite aggregator (#26800) Fix `ShardSplittingQuery` to respect nested documents. (#27398) [Docs] Restore section about multi-level parent/child relation in parent-join (#27392) Add TcpChannel to unify Transport implementations (#27132) Add note on plugin distributions in plugins folder Remove implementations of `TransportChannel` (#27388) Update Google SDK to version 1.23 (#27381) Fix Gradle 4.3.1 compatibility for logging (#27382) [Test] Change Elasticsearch startup timeout to 120s in packaging tests Docs/windows installer (#27369)

This is a follow up to #27132. As that PR greatly simplified the connection logic inside a low level transport implementation, much of the functionality provided by the NioClient class is no longer necessary. This commit removes that class.

This commit is a follow up to the work completed in #27132. Essentially it transitions two more methods (sendMessage and getLocalAddress) from Transport to TcpChannel. With this change, there is no longer a need for TcpTransport to be aware of the specific type of channel a transport returns. So that class is no longer parameterized by channel type.

Tim-Brooks added 14 commits October 23, 2017 15:55

WIP

d8cdf08

WIP

a5a30ca

Fix mock transport tests

e5d4c21

WIP

044f6bf

DWIP

068bde0

WIP

dbc2c85

Fix nio client tests

701e3a8

WIP

ac4cfc3

Merge branch 'master' into introduce_tcp_connection

b63eb9b

WIP

8f85ba5

Cleanup futures

c887e7d

Cleanup channel closing

4c5d0d6

Work on handling accepting

57a506d

Cleanup

8fa658d

Tim-Brooks added :Distributed/Network Http and internode communication implementations >non-issue review v6.1.0 v7.0.0 labels Oct 26, 2017

Tim-Brooks requested review from jasontedor, jaymode and s1monw October 26, 2017 21:16

Add license

1028126

s1monw suggested changes Oct 27, 2017

View reviewed changes

Tim-Brooks added 5 commits October 27, 2017 15:32

Work on cleaning up close futures

172f9eb

Cleanups

cea0253

Add back method

3565108

Merge remote-tracking branch 'upstream/master' into introduce_tcp_con…

9dc88dc

…nection

Remove futures

10fb412

Merge branch 'master' into introduce_tcp_connection

9e7903d

s1monw suggested changes Nov 13, 2017

View reviewed changes

Tim-Brooks added 5 commits November 13, 2017 09:25

Drop parameterization

8e32b7b

Fix test

1f1fd43

make changes based on review

8d8df72

Cleanup code

c57d919

Merge remote-tracking branch 'upstream/master' into introduce_tcp_con…

22bef9c

…nection

Tim-Brooks changed the title ~~Add TcpChannel<C> to unify Transport impls~~ Add TcpChannel to unify Transport impls Nov 13, 2017

Tim-Brooks changed the title ~~Add TcpChannel to unify Transport impls~~ Add TcpChannel to unify Transport implementations Nov 13, 2017

use completablefuture

c664c07

Make getter public

d58678c

s1monw approved these changes Nov 14, 2017

View reviewed changes

Tim-Brooks added 2 commits November 14, 2017 09:07

Changes based on review

3a79a16

Don't wrap throwable

bb89a97

jasontedor approved these changes Nov 15, 2017

View reviewed changes

Merge remote-tracking branch 'upstream/master' into introduce_tcp_con…

9fd9984

…nection

Tim-Brooks merged commit ca11085 into elastic:master Nov 15, 2017

Tim-Brooks mentioned this pull request Nov 15, 2017

Remove parameterization from TcpTransport #27407

Merged

Tim-Brooks mentioned this pull request Nov 15, 2017

Remove unnecessary nio client #27408

Merged


		import org.elasticsearch.action.support.PlainActionFuture;

		public class PlainChannelFuture<Channel> extends PlainActionFuture<Channel> {


		ListenableActionFuture<C> closeAsync();

		ListenableActionFuture<C> getCloseFuture();


		public interface TcpChannel<C extends TcpChannel<C>> {

		ListenableActionFuture<C> closeAsync();


		import java.io.IOException;

		public interface TcpChannel<C extends TcpChannel<C>> {


		public interface TcpChannel<C extends TcpChannel<C>> {

		void closeAsync();


		public class TcpChannelUtils {

		public static <C extends TcpChannel<C>> void closeChannel(C channel, boolean blocking) {

Conversation

Tim-Brooks commented Oct 26, 2017

Uh oh!

s1monw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Tim-Brooks Oct 27, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Tim-Brooks commented Nov 3, 2017

Uh oh!

s1monw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Tim-Brooks commented Nov 13, 2017

Uh oh!

Tim-Brooks commented Nov 13, 2017

Uh oh!

s1monw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

s1monw commented Nov 14, 2017

Uh oh!

jasontedor left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Tim-Brooks Oct 27, 2017 •

edited

Loading