Guava-free Same Thread Executor by drcrallen · Pull Request #5413 · apache/druid

drcrallen · 2018-02-22T05:10:16Z

The MoreExecs.sameThreadExecutor() is deprecated as per
https://github.com/google/guava/blob/v18.0/guava/src/com/google/common/util/concurrent/MoreExecutors.java#L266-L267
and the function signature moves around a lot in future guava releases.

This creates a new SameThreadExecutorService using more modern java
features.

The `MoreExecs.sameThreadExecutor()` is deprecated as per https://github.com/google/guava/blob/v18.0/guava/src/com/google/common/util/concurrent/MoreExecutors.java#L266-L267 and the function signature moves around a lot in future guava releases. This creates a new `SameThreadExecutorService` using more modern java features.

leventov · 2018-02-22T10:37:37Z

    }).writeValueAsString(taskGroups.get(groupId).sequenceOffsets);
    final Map<String, Object> context = spec.getContext() == null
-                                        ? ImmutableMap.of("checkpoints", checkpoints, IS_INCREMENTAL_HANDOFF_SUPPORTED, true)
+                                        ? ImmutableMap.of(


Please replace ternary with if-else

leventov · 2018-02-22T10:39:23Z

@@ -361,11 +362,17 @@ public void testBasics() throws Exception

    for (DataSegment publishedSegment : publishedSegments) {
      Optional<Map.Entry<SegmentDescriptor, Pair<Executor, Runnable>>> optional = handOffCallbacks.entrySet().stream()


Please break line before .entrySet() to make reduce indentation below. Same below in this class.

or thought I did, checking

haha, pushed to wrong repo, fixed now

leventov · 2018-02-22T10:39:57Z

-        SupervisorManager.class)));
+    tac = new LocalTaskActionClientFactory(
+        taskStorage,
+        new TaskActionToolbox(taskLockbox, mdc, emitter, EasyMock.createMock(


Not formatted properly

leventov · 2018-02-22T11:43:55Z

+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class SameThreadExecutorService extends AbstractExecutorService


Is this code just copied from Guava? Could you refer to a file on Github (under google/guava) that contains it, in javadoc comment?

Could you add "TODO: replace with ... when updated to Guava X", and a corresponding issue?

It is not copied from guava

drcrallen · 2018-02-22T16:30:06Z

@leventov addressed comments

b-slim

Overall LGTM, can you also ban those functions as well to avoid future invocation?

leventov · 2018-02-22T20:13:26Z

              notices.add(new RunNotice());
            }
-          }, MoreExecutors.sameThreadExecutor()
+          }, Execs.sameThreadExecutor()


Execs.sameThreadExecutor() should be on the next line.

leventov · 2018-02-22T20:17:30Z

    public Runnable commit()
    {
-      return () -> {};
+      return () -> {


It's Runnables.getNoopRunnable().

leventov · 2018-02-22T20:17:53Z

    public Runnable commit()
    {
-      return () -> {};
+      return () -> {


leventov · 2018-02-22T20:25:20Z

    return DummyExecutorService.INSTANCE;
  }

+  public static ListeningExecutorService sameThreadExecutor()


It was a mistake in Guava's API to call this method this way, because it gives full impression that some static constant is returned from it (I thought that until now), while in reality this method (and the removed method in Guava) creates a new Object each time. (Guava developers fixed this mistake by calling the new method newDirectExecutorService().)

However, I think that Druid actually never tries to terminate this service, so it would be better to keep the method name as is, actually cache the executor service, and simplify it's implementation, by either throwing an UnsupportedOperationException in shutdown() and shutdownNow() or making them no-ops, and always returning false from isTerminated(), and removing all state and synchronization from the class.

That's reasonable

drcrallen · 2018-02-24T03:03:36Z

I'm waiting on #5413 (review) because doing it will collide with #5414

leventov · 2018-02-26T14:27:39Z

-                                                                                                      publishedSegment.getShardSpec().getPartitionNum()
-                                                                                                  )))
-                                                                                                  .findFirst();
+      Optional<Map.Entry<SegmentDescriptor, Pair<Executor, Runnable>>> optional = handOffCallbacks


This construction looks like handOffCallbacks.get()...

I'd rather not change any logic. This just retains the prior code

leventov · 2018-02-26T14:27:46Z

-                                                                                                      publishedSegment.getShardSpec().getPartitionNum()
-                                                                                                  )))
-                                                                                                  .findFirst();
+      Optional<Map.Entry<SegmentDescriptor, Pair<Executor, Runnable>>> optional = handOffCallbacks


I'm not going to change logic in this PR

leventov · 2018-02-26T14:27:54Z

-                                                                                                      publishedSegment.getShardSpec().getPartitionNum()
-                                                                                                  )))
-                                                                                                  .findFirst();
+      Optional<Map.Entry<SegmentDescriptor, Pair<Executor, Runnable>>> optional = handOffCallbacks


logic is retained as the prior on purpose

leventov · 2018-02-26T14:28:56Z

+/**
+ * A simple class that implements the ExecutorService interface, but runs the code on a call to submit
+ */
+public class SameThreadExecutorService extends AbstractExecutorService


Please make this class a singleton

I made it singleton in Execs

Execs doesn't prohibit instantiation of new SameThreadExecutorService objects.

Yes that is true. Why force singleton?

To prohibit accidential instantiation.

Referencing Execs in the javadoc comment would be useful too.

Made the class constructor package private and added some more comments.

leventov · 2018-02-26T14:30:33Z

+    final long nanos = TimeUnit.NANOSECONDS.convert(timeout, unit);
+    final long millis = TimeUnit.MILLISECONDS.convert(timeout, unit);
+    final int sleepNanos = (int) (nanos - millis * 1_000_000L);
+    Thread.sleep(millis, sleepNanos);


I don't think sleeping makes much sense here, either return immediately or throw an UnsupportedOperationException.

The contract says it will wait, so I wait

awaitTermination() should be called only after shutdown() or shutdownNow(), which throw UnsupportedOperationException, so it's even more reasonable to throw it from awaitTermination()

The interface is unambiguous : https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html#awaitTermination-long-java.util.concurrent.TimeUnit-

Blocks until all tasks have completed execution after a shutdown request, or the timeout occurs, or the current thread is interrupted, whichever happens first.

Which is what this implementation does.

Javadocs of shutdown() and shutdownNow() also don't give an option to throw an UnsupportedOperationException. Yet when awaitTermination() is called on SameThreadExecutorService, it should be a programming bug, so better to report it rather than silence.

UnsupportedOperationException is an unchecked exception, it does not need to be declared

Yes, and that is why there is no problem with throwing it from the method, although the interface doesn't specify. I mean, if we throw UOE from shutdown() and shutdownNow(), rather than make those methods no-ops, it makes just as much sense to throw UOE from awaitTermination() as well. And so to keep them consistent.

@drcrallen the PR looks good to me, except this thing, but it's not blocking for merge. However I still think it should be changed.

* SameTheadExecutorService constructor now package private

leventov · 2018-02-27T11:53:57Z

 */
 public class SameThreadExecutorService extends AbstractExecutorService
 {
+  // Use io.druid.java.util.common.concurrent.Execs#sameThreadExecutor()


Could you please make this a Javadoc comment

Or just move to class-level comment

Added class level comment

…dExecutor

drcrallen · 2018-03-12T14:27:40Z

@leventov any other comments here?

leventov · 2018-03-13T14:52:31Z

+
+/**
+ * A simple class that implements the ExecutorService interface, but runs the code on a call to submit
+ * Use io.druid.java.util.common.concurrent.Execs#sameThreadExecutor() to get the instance


Please make it a javadoc link

…dExecutor

drcrallen · 2018-03-19T17:49:15Z

@leventov anything else here?

…dExecutor

drcrallen · 2018-03-20T16:34:56Z

ick, looks like io.druid.curator.announcement.Announcer#Announcer shuts down the announcer it was passed! which causes tests to fail. I'm going to remove the failure on shutdown

…dExecutor

drcrallen · 2018-03-20T16:49:48Z

had to revert "only one instance" for the same thread executor service :(

leventov · 2018-03-20T21:00:23Z

Why?

drcrallen · 2018-03-23T16:43:22Z

@leventov some of the tests close out the executor service. Announcer comes to mind. Fixing that is a bit much for this pr.

drcrallen · 2018-03-23T16:50:42Z

@leventov see #5527

leventov · 2018-03-23T17:18:55Z

In this case, why wouldn't we temporarily implement all shutdown(), shutdownNow() and isShutdown() as no-ops? And switch to throwing UnsupportedOperationException once #5527 is fixed?

drcrallen · 2018-03-23T17:41:30Z

@leventov because that is only one use case, I'm not sure if there are others since I didn't try to get as many tests to pass as possible. I don't want to make assumptions about how the code is using executor services if it is already shown to do odd things with who-controls-this-executor.

As such keeping behavior close to the executor service interface is desired. This PR keeps behavior closer to the executor service interface

leventov · 2018-03-23T18:15:25Z

Having those methods implemented as no-ops (hence awaitTermination() returns immediately) is guaranteed not to cause errors in runtime. It may not catch extra bugs, yes, but we know that we have bugs already. Once those bugs are fixed, it will be desired to switch back to singleton model, so why changing code back and forth? If it's made singleton now, all that is going to be needed is not replace no-op bodies with bodies throwing exceptions.

drcrallen · 2018-03-26T17:19:54Z

The constructor is package private, all instances must come through Execs.sameThreadExecutor() moving to singleton later will be straight forward

leventov · 2018-03-28T14:20:00Z

I'm concerned about potential performance impact of this solution. Now the executor that is supposed to be a cost-free abstraction (just execute the Runnable's run or Callable's call, passed in) does two operations with a Phaser. As the very minimum, it's a flush or read/write CPU core pipelines. And if some instance of sameThreadExecutor() is used widely across Threads, assuming it has no state, it could suddenly become a concurrency bottleneck. For no real business logic / correctness purpose.

It seems to me that the no-op solution doesn't have any risks and couldn't theoretically have any performance impact.

drcrallen · 2018-03-28T17:03:31Z

@leventov com.google.common.util.concurrent.MoreExecutors.SameThreadExecutorService uses a state lock whenever tasks are submitted. Are you suggesting the performance of a Phaser would be worse?

b-slim · 2018-04-12T16:47:12Z

@drcrallen can we take this to the finish line?

leventov · 2018-04-12T17:07:40Z

@drcrallen the current SameThreadExecutorService implementation with Phaser has logical issues:

It should check values returned from register(), correctly processing termination case
Initial register() and final arriveAndDeregister() is potentially done from different threads, that could lead to errors (this bug has the same root cause).

I didn't mention that before because I wanted to persuade you not to use Phaser using different arguments. But now considering that the current implementation of SameThreadExecutorService is relatively complex and prone to some bugs or inconsistencies, while the proposed no-op approach is absolutely trivial, would you agree to switch to stateless singleton with no-op methods?

drcrallen · 2019-01-05T01:41:47Z

This is too old at this point. closing

Charles Allen added 4 commits February 21, 2018 20:48

Add copyright

37179e6

Linter fixes

0ac9e1a

More linter

5784383

leventov reviewed Feb 22, 2018

View reviewed changes

Charles Allen added 2 commits February 22, 2018 07:46

Manual formattting

28d2342

Prettier formatting

8ad1173

drcrallen added the Compatibility label Feb 22, 2018

b-slim approved these changes Feb 22, 2018

View reviewed changes

leventov reviewed Feb 22, 2018

View reviewed changes

Charles Allen added 2 commits February 23, 2018 18:57

Simply the same thread executor service

9c6f4b5

Address code comments

d5a0233

leventov reviewed Feb 26, 2018

View reviewed changes

leventov mentioned this pull request Feb 26, 2018

Strange code in AppenderatorDriverRealtimeIndexTaskTest that looks like it should be replaced with map.get() #5425

Closed

More comments and druid-forbidden-apis.txt

40ae3cd

* SameTheadExecutorService constructor now package private

leventov reviewed Feb 27, 2018

View reviewed changes

Charles Allen added 2 commits March 12, 2018 07:25

Merge remote-tracking branch 'upstream/master' into guava/noSameThrea…

71b2143

…dExecutor

Move around some comments

c559a88

leventov reviewed Mar 13, 2018

View reviewed changes

Charles Allen added 2 commits March 19, 2018 10:48

Merge remote-tracking branch 'upstream/master' into guava/noSameThrea…

a1688a4

…dExecutor

Add javadoc link

730a7e0

Merge remote-tracking branch 'upstream/master' into guava/noSameThrea…

327ee54

…dExecutor

Charles Allen added 2 commits March 20, 2018 09:47

Allow instances of SameThreadExecutorService

0f4cf1f

Merge remote-tracking branch 'upstream/master' into guava/noSameThrea…

5162e0c

…dExecutor

drcrallen closed this Jan 5, 2019

drcrallen mentioned this pull request Jan 7, 2019

Use Guava Compatible immediate executor service #6815

Merged

		@@ -361,11 +362,17 @@ public void testBasics() throws Exception

		for (DataSegment publishedSegment : publishedSegments) {
		Optional<Map.Entry<SegmentDescriptor, Pair<Executor, Runnable>>> optional = handOffCallbacks.entrySet().stream()

Conversation

drcrallen commented Feb 22, 2018

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!

drcrallen commented Feb 22, 2018

Uh oh!

b-slim 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!

drcrallen commented Feb 24, 2018

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!

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!