Prohibit Futures.addCallback(Future, Callback) #8809

zhenxiao · 2019-11-02T00:52:53Z

leventov · 2019-11-02T08:19:15Z

Mechanically adding Execs.directExecutor() doesn't solve the problem. You should actually thoughtfully check every case and verify that Execs.directExecutor() is an acceptable option in each of them. A case should meet at least one of three criteria, as described in https://github.com/code-review-checklists/java-concurrency#cf-beware-non-async:

The callback is lightweight and non-blocking (as is the case in your PR In DirectDruidClient, don't run Future cancellation listener in HTTP library executor #8700); or
The callback is added from the same executor as the future is completed; or
The callback attachment is preceded with if (future.isDone()) check.

Then, the specific criterion should be identified in a comment, similarly to what I asked you to do here: #8700 (comment).

Otherwise (if the case meets neither of these criteria), directExecutor() should not be used.

leventov · 2019-11-02T08:21:43Z

A forbidden-apis entry should also be added. See an example here: 15fbf59#diff-4cfb6baf06c34e537a8666d9a0f3aebd

zhenxiao · 2019-11-06T01:47:55Z

thank you, @leventov
comments addressed
did a force push, put this work on top of #8700, so that compiles with forbidden-apis
only the 2nd commit for review

leventov · 2019-11-06T06:56:53Z

...main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexPhaseRunner.java

@@ -250,7 +254,8 @@ public void onFailure(Throwable t)
            LOG.error(t, "Error while running a task for subTaskSpec[%s]", spec);
            taskCompleteEvents.offer(SubTaskCompleteEvent.fail(spec, t));
          }
-        }
+        },
+        blockingQueueHandler


taskCompleteEvents.offer(completeEvent); -- BlockingQueue.offer() is actually non-blocking

Correction - BlockingQueue.offer() still needs to enter a critical section, but it may be considered "mostly non-blocking", so directExecutor() should still be OK

leventov · 2019-11-06T06:57:28Z

core/src/main/java/org/apache/druid/java/util/common/concurrent/ListenableFutures.java

@@ -61,15 +61,19 @@ public void onFailure(Throwable t)
          {
            finalFuture.setException(t);
          }
-        });
+        },
+            // The callback is non-blocking and quick, so it's OK to schedule it using directExecutor()


Too much indentation

seems this is the indentation suggested when compile the code. Anyway, let me refactor to make it more readable

leventov · 2019-11-06T06:57:34Z

core/src/main/java/org/apache/druid/java/util/common/concurrent/ListenableFutures.java

      }

      @Override
      public void onFailure(Throwable t)
      {
        finalFuture.setException(t);
      }
-    });
+    },
+        // The callback is non-blocking and quick, so it's OK to schedule it using directExecutor()


leventov · 2019-11-06T06:59:41Z

indexing-service/src/main/java/org/apache/druid/indexing/overlord/RemoteTaskRunner.java

@@ -256,7 +256,9 @@ public void onFailure(Throwable throwable)
                          waitingForMonitor.notifyAll();
                        }
                      }
-                    }
+                    },
+                    // The callback is non-blocking and quick, so it's OK to schedule it using directExecutor()


synchronized (waitingForMonitor) is blocking. You may argue that this lock is (almost) free, for some reasons, so then the callback should still be considered "mostly non-blocking" and OK to schedule on directExecutor(), but in any case it demands a more elaborate comment

leventov · 2019-11-06T07:19:12Z

indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java

@@ -556,7 +559,8 @@ private void handleStatus(final TaskStatus status)
                 .emit();
            }
          }
-        }
+        },
+        statusHandler


It's non-obvious from this point in code to see that attachCallbacks() is called from a different thread pool than statusFuture is completed.

There are options to improve:

Propagate Executor argument and pass statusHandler only in manage(), and, additionally, amend the javadoc comment for manage() noting that it is run from {@link #managerExec}.

Just add a more elaborate comment right here.

In either case, it will be non-obvious for readers why different executors demand a special executor. So the comment should look something like that:
"Using dedicated statusHandler executor instead of directExecutor() because the callback's onSuccess() is not trivial and because the statusFuture is completed in some incapsulated thread pool in TaskRunner; directExecutor() may create operational instability and subtle dependency between components here. See https://github.com/code-review-checklists/java-concurrency#cf-beware-non-async for details."

Something similar should be done in all other places where you use a custom executor instead of directExecutor().

leventov · 2019-11-06T07:21:32Z

...ce/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java

@@ -1008,7 +1014,8 @@ public void onFailure(Throwable t)
            log.error(t, "Error while publishing segments for sequenceNumber[%s]", sequenceMetadata);
            handoffFuture.setException(t);
          }
-        }
+        },
+        sequencePersistExecutor


I withdraw from analyzing execution flows and thread pool relationships here. @jihoonson could you please vet this?

Sure, I will take a look soon.

This still needs to be reviewed by maintainers of SeekableStreamIndexTaskRunner.

leventov · 2019-11-06T07:26:38Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerTaskManager.java

@@ -251,7 +251,9 @@ public void onFailure(Throwable t)
          {
            submitNoticeToExec(new StatusNotice(task, TaskStatus.failure(task.getId())));
          }
-        }
+        },
+        // The callback is non-blocking and quick, so it's OK to schedule it using directExecutor()


Please add a javadoc comment to submitNoticeToExec() like "This method is mostly non-blocking because exec is a ThreadPoolExecutor with unbounded queue used by default."

Correspondingly, this comment also should say "mostly non-blocking", not just "non-blocking".

leventov · 2019-11-06T07:29:54Z

...xing-service/src/main/java/org/apache/druid/indexing/worker/http/TaskManagementResource.java

@@ -192,7 +196,8 @@ public void onFailure(Throwable th)
              log.debug(ex, "Request timed out or closed already.");
            }
          }
-        }
+        },


Some weird asynchrony going on in this method (at least it looks so), so it demands more comments

leventov · 2019-11-06T07:32:36Z

server/src/main/java/org/apache/druid/server/http/SegmentListerResource.java

@@ -205,7 +209,8 @@ public void onFailure(Throwable th)
              log.debug(ex, "Request timed out or closed already.");
            }
          }
-        }
+        },


Same. This code looks similar to TaskManagementResource. Maybe the common logic should be extracted. Or, there should be comments to both of them noting that these classes are written in a similar way so changes, fixes, refactorings should be applied to both classes at the same time.

extract and refactor TaskManagementResource, SegmentListerResource needs more work, will file separate PR to do it
in this PR, will add comments to both classes at the same time

Ok, but the comments that you left right here don't make anything clearer. It's ok to omit them in this PR and instead write better comments as part of that PR that you planned, though.

leventov · 2019-11-16T11:30:24Z

indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java

@@ -560,6 +560,7 @@ private void handleStatus(final TaskStatus status)
            }
          }
        },
+        //Using dedicated statusHandler executor instead of directExecutor() because the callback's onSuccess() is not trivial and because the statusFuture is completed in some incapsulated thread pool in TaskRunner; directExecutor() may create operational instability and subtle dependency between components here. See https://github.com/code-review-checklists/java-concurrency#cf-beware-non-async for details


Line longer than 120 columns

leventov · 2019-11-16T11:31:30Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerTaskManager.java

@@ -177,6 +177,9 @@ public void stop() throws Exception
    return completedTasks;
  }

+  /**
+   * This method is mostly non-blocking because exec is a ThreadPoolExecutor with unbounded queue used by default.


Please make exec and ThreadPoolExecutor javadoc links

leventov · 2019-11-16T11:37:00Z

indexing-service/src/main/java/org/apache/druid/indexing/overlord/RemoteTaskRunner.java

@@ -257,7 +257,7 @@ public void onFailure(Throwable throwable)
                        }
                      }
                    },
-                    // The callback is non-blocking and quick, so it's OK to schedule it using directExecutor()
+                    // The callback is mostly non-blocking and quick, so it's OK to schedule it using directExecutor()


In my comment, I wrote "you may argue that... this code is mostly non-blocking", I didn't state that (and I actually don't know). I. e. the fact that the synchronized block in onSuccess() "is mostly non-blocking" requires some proof, expressed in a comment. There is not enough information to make such proof (or disproof) just in this code excerpt in Github's interface. A wider context (specifically, the code which synchronizes on waitingForMonitor and waits) should be considered.

get it. the synchronization of waitingForMonitor involves a while loop. Will use dedicated thread

leventov · 2019-11-16T11:39:32Z

server/src/main/java/org/apache/druid/server/http/SegmentListerResource.java

@@ -205,7 +209,8 @@ public void onFailure(Throwable th)
              log.debug(ex, "Request timed out or closed already.");
            }
          }
-        }
+        },


Ok, but the comments that you left right here don't make anything clearer. It's ok to omit them in this PR and instead write better comments as part of that PR that you planned, though.

zhenxiao · 2019-11-17T09:23:49Z

thank you, @leventov
get comments addressed

zhenxiao · 2019-12-03T22:17:22Z

Hi @leventov I rebased this PR to resolve code conflicts. Are you free to take a look?

leventov · 2019-12-06T17:09:50Z

indexing-service/src/main/java/org/apache/druid/indexing/overlord/RemoteTaskRunner.java

@@ -175,6 +175,7 @@

  private final ConcurrentMap<String, ScheduledFuture> removedWorkerCleanups = new ConcurrentHashMap<>();
  private final ProvisioningStrategy<WorkerTaskRunner> provisioningStrategy;
+  private final ExecutorService monitorSyncHandler;


This extra service must be explicitly shutdown in stop()/close(), see https://github.com/code-review-checklists/java-concurrency#explicit-shutdown. Applies to other classes in this PR as well

thank you, fixed.

leventov · 2019-12-06T17:13:10Z

server/src/main/java/org/apache/druid/segment/realtime/appenderator/AppenderatorPlumber.java

@@ -483,7 +485,8 @@ public void onFailure(Throwable e)
            log.warn(e, "Failed to push [%,d] segments.", segmentsToPush.size());
            errorHandler.apply(e);
          }
-        }
+        },
+        scheduledExecutor


This still needs to be reviewed by maintainers of AppenderatorPlumber

leventov · 2019-12-06T17:14:35Z

...ce/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java

@@ -1008,7 +1014,8 @@ public void onFailure(Throwable t)
            log.error(t, "Error while publishing segments for sequenceNumber[%s]", sequenceMetadata);
            handoffFuture.setException(t);
          }
-        }
+        },
+        sequencePersistExecutor


This still needs to be reviewed by maintainers of SeekableStreamIndexTaskRunner.

leventov · 2019-12-07T11:04:14Z

indexing-service/src/main/java/org/apache/druid/indexing/overlord/RemoteTaskRunner.java

@@ -350,6 +350,7 @@ public void stop()
      return;
    }
    try {
+      monitorSyncHandler.shutdown();


These two things must be closed using the closer below

leventov · 2019-12-07T11:05:48Z

indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java

@@ -219,6 +219,7 @@ public void stop()
      tasks.clear();
      taskFutures.clear();
      active = false;
+      statusHandler.shutdownNow();


Please shutdown these three Executors using a Closer

leventov · 2019-12-07T11:12:37Z

...ce/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java

@@ -219,6 +221,7 @@
  private final List<ListenableFuture<SegmentsAndMetadata>> handOffWaitList = new ArrayList<>();

  private final LockGranularity lockGranularityToUse;
+  private final ExecutorService sequencePersistExecutor;


In classes that don't have close()/stop() methods yet, like SeekableStreamIndexTaskRunner, they should be introduced if you want to manage an executor.

If the class is pretty ephemeral (e. g. many instances of this class routinely start and stop), it should be strongly preferred to create an executor one level up the stack and reuse it for all instances of the class (or all instances of the class within a certain domain e. g. all SeekableStreamIndexTaskRunners associated with a single datasource, or something like that).

Note again, I don't actually tell you that this class SeekableStreamIndexTaskRunner falls into this category - please figure it out yourself, or consult the maintainers of this subsystem.

github-actions · 2023-08-03T00:17:44Z

This pull request has been marked as stale due to 60 days of inactivity.
It will be closed in 4 weeks if no further activity occurs. If you think
that's incorrect or this pull request should instead be reviewed, please simply
write any comment. Even if closed, you can still revive the PR at any time or
discuss it on the dev@druid.apache.org list.
Thank you for your contributions.

github-actions · 2023-09-01T00:16:27Z

This pull request/issue has been closed due to lack of activity. If you think that
is incorrect, or the pull request requires review, you can revive the PR at any time.

leventov added the Improvement label Nov 2, 2019

zhenxiao force-pushed the callback branch from e649d74 to 16401e7 Compare November 6, 2019 01:44

leventov requested changes Nov 6, 2019

View reviewed changes

zhenxiao force-pushed the callback branch from dbf1720 to 699c538 Compare November 7, 2019 08:50

leventov requested changes Nov 16, 2019

View reviewed changes

zhenxiao added 3 commits December 3, 2019 14:14

Prohibit Futures.addCallback(Future, Callback)

b42e38d

address comments and code refactoring

23b9ecd

comments addressed

20bd314

zhenxiao force-pushed the callback branch from 88b630e to 20bd314 Compare December 3, 2019 22:15

leventov requested changes Dec 6, 2019

View reviewed changes

leventov added the Area - Automation/Static Analysis label Dec 6, 2019

shutdown thread in stop/close

38ecf82

leventov requested changes Dec 7, 2019

View reviewed changes

leventov reviewed Dec 7, 2019

View reviewed changes

github-actions bot added the stale label Aug 3, 2023

github-actions bot closed this Sep 1, 2023

Prohibit Futures.addCallback(Future, Callback) #8809

Prohibit Futures.addCallback(Future, Callback) #8809

Conversation

zhenxiao commented Nov 2, 2019

leventov commented Nov 2, 2019

leventov commented Nov 2, 2019

zhenxiao commented Nov 6, 2019

Choose a reason for hiding this comment

leventov Nov 6, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leventov Nov 16, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhenxiao commented Nov 17, 2019

zhenxiao commented Dec 3, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Aug 3, 2023

github-actions bot commented Sep 1, 2023

leventov Nov 6, 2019 •

edited

Loading

leventov Nov 16, 2019 •

edited

Loading