Extract batch executor out of cluster service #24102

ywelsch · 2017-04-14T07:20:06Z

Code refactoring that extracts the executor functionality from ClusterService and makes it a reusable component that can be tested in isolation. This will allow this component to be used in a follow-up separating ClusterService into MasterService and ClusterApplierService, which will both make use of the executor components.

jasontedor

I think that it's a nice, and long overdue cleanup. I left some comments about the naming.

jasontedor · 2017-04-20T02:00:21Z

core/src/main/java/org/elasticsearch/cluster/service/ClusterService.java

@@ -244,6 +240,35 @@ protected synchronized void doStop() {
    protected synchronized void doClose() {
    }

+    class ClusterServiceTaskBatching extends TaskBatching {


Sorry to get hung up on the naming, I know it's annoying, but should it be ClusterServiceTaskBatcher and TaskBatcher?

Nit: +1 to Jason's suggestion

jasontedor · 2017-04-20T02:01:17Z

core/src/main/java/org/elasticsearch/cluster/service/TaskBatching.java

+ * Batching support for {@link PrioritizedEsThreadPoolExecutor}
+ * Tasks that share the same batching key are batched (see {@link BatchingTask#batchingKey})
+ */
+public abstract class TaskBatching {


Should it be TaskBatcher?

jasontedor · 2017-04-20T02:01:37Z

core/src/main/java/org/elasticsearch/cluster/service/TaskBatching.java

+     * Represents a runnable task that supports batching.
+     * Implementors of TaskBatching can subclass this to add a payload to the task.
+     */
+    protected abstract class BatchingTask extends SourcePrioritizedRunnable {


Should it be BatchableTask?

My preference goes to BatchedTask and have the taskIdentity become just task (next to the batching key). Obviously this falls deep into bikeshedding territory so feel totally free to reject and keep what you have.

jasontedor · 2017-04-20T02:03:52Z

core/src/main/java/org/elasticsearch/cluster/service/TaskBatching.java

+        }
+    }
+
+    public interface BatchingKey<T> {


I think BatchKey?

jasontedor · 2017-04-20T02:14:33Z

core/src/main/java/org/elasticsearch/cluster/service/ClusterService.java

@@ -1189,7 +1074,7 @@ public void clusterChanged(ClusterChangedEvent event) {
            if (master && !event.localNodeMaster()) {
                master = false;
                for (LocalNodeMasterListener listener : listeners) {
-                    Executor executor = threadPool.executor(listener.executorName());
+                    java.util.concurrent.Executor executor = threadPool.executor(listener.executorName());


I think that you can just import this and there's no need for the FQN here?

jasontedor · 2017-04-20T02:14:38Z

core/src/main/java/org/elasticsearch/cluster/service/ClusterService.java

@@ -1180,7 +1065,7 @@ public void clusterChanged(ClusterChangedEvent event) {
            if (!master && event.localNodeMaster()) {
                master = true;
                for (LocalNodeMasterListener listener : listeners) {
-                    Executor executor = threadPool.executor(listener.executorName());
+                    java.util.concurrent.Executor executor = threadPool.executor(listener.executorName());


I think that you can just import this and there's no need for the FQN here?

bleskes

This looks great. I left a bunch of nits + two real comments (re thread pool and the BatchingKey interface)

bleskes · 2017-04-18T10:49:06Z

core/src/main/java/org/elasticsearch/cluster/service/SingleTaskExecutor.java

+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public abstract class SingleTaskExecutor {


Java docs would be great.

bleskes · 2017-04-20T08:25:45Z

core/src/main/java/org/elasticsearch/cluster/service/ClusterService.java

@@ -244,6 +240,35 @@ protected synchronized void doStop() {
    protected synchronized void doClose() {
    }

+    class ClusterServiceTaskBatching extends TaskBatching {


Nit: +1 to Jason's suggestion

bleskes · 2017-04-20T08:36:09Z

core/src/main/java/org/elasticsearch/cluster/service/TaskBatching.java

+     * Represents a runnable task that supports batching.
+     * Implementors of TaskBatching can subclass this to add a payload to the task.
+     */
+    protected abstract class BatchingTask extends SourcePrioritizedRunnable {


My preference goes to BatchedTask and have the taskIdentity become just task (next to the batching key). Obviously this falls deep into bikeshedding territory so feel totally free to reject and keep what you have.

bleskes · 2017-04-20T08:42:21Z

core/src/main/java/org/elasticsearch/cluster/service/ClusterService.java

+
+        @Override
+        protected void run(Object batchingKey, List<? extends BatchingTask> tasks, String tasksSummary) {
+            ClusterStateTaskExecutor<Object> taskExecutor = (ClusterStateTaskExecutor<Object>) batchingKey;


I'm doubting if we should introduce generics for the BatchingTask and batchingKey. I'm fine with keeping as is if you don't like the generic version.

bleskes · 2017-04-20T08:44:25Z

core/src/main/java/org/elasticsearch/cluster/service/TaskBatching.java

+                }
+            }
+            if (toRemove.isEmpty() == false) {
+                Object batchingExecutor = toRemove.get(0).batchingKey;


can we assert all of them have the same batching key?

bleskes · 2017-04-20T08:48:29Z

core/src/main/java/org/elasticsearch/cluster/service/TaskBatching.java

+                        }
+                    }
+                }
+                for (BatchingTask task : toRemove) {


can we maybe change onTimeout(task, timeout) to be onTimeout(List<..> tasks, timeout, batchingKey) ? than we can leave the spawning to the implementation. This means we don't need to capture a thread pool for this class. Also I'm wondering if we should fold threadPool.scheduler() into PrioritizedEsThreadPoolExecutor as I don't see the need for people to use multiple schedulers.

bleskes · 2017-04-20T08:50:31Z

core/src/main/java/org/elasticsearch/cluster/service/TaskBatching.java

+    }
+
+    /**
+     * Action to be implemented by the specific batching implementation


can we add a comment that all the tasks are guaranteed to have to supplied batching key?

bleskes · 2017-04-20T08:52:43Z

core/src/main/java/org/elasticsearch/cluster/service/TaskBatching.java

+         * This method can be called multiple times with different lists before execution.
+         * This allows groupd task description but the submitting source.
+         */
+        default String describeTasks(List<T> tasks) {


I'm not sure we need this interface. It feels weird to me that the batchingKey knows how to describe the tasks. I would prefer to have an abstract method describeTasks(List<? extends BatchedTask>, batchKey) which can be overridden by the subclasses. This will be consistent with run and onTimeout (see other comment) and will allows us to remove this interface.

bleskes · 2017-04-20T08:54:22Z

core/src/test/java/org/elasticsearch/cluster/service/TaskBatchingTests.java

+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasToString;
+
+public class TaskBatchingTests extends TaskExecutorTests {


since this is the only sub class that inherits from TaskExecutorTests, can we fold them together? or do you already see a clear need for a base class?

The separation into two classes allows the tests in TaskExecutorTests to be run both with and without batching support, i.e. it shows that batching does not have an adverse effect on these tests.

I see. I'm not sure it merits the base class, but I'm willing to go with it and see how it goes.

ywelsch · 2017-04-20T10:21:13Z

@jasontedor @bleskes thanks for the feedback. I've pushed a1e1ec4 which should address all comments.

bleskes

LGTM.

jasontedor

Thanks for making the name change, can you also change the variable names throughout to reflect? (Again, I'm so sorry for such an annoying request, I think it's a really good PR.)

jasontedor · 2017-04-20T11:52:11Z

core/src/main/java/org/elasticsearch/cluster/service/ClusterService.java

@@ -114,14 +110,14 @@
    private TimeValue slowTaskLoggingThreshold;

    private volatile PrioritizedEsThreadPoolExecutor threadPoolExecutor;
+    private volatile ClusterServiceTaskBatcher taskBatching;


Can you update variable names to reflect the class name changes too?

Good catch. I've pushed 98a4117 which should hopefully fix that (and others)

jasontedor

LGTM.

jasontedor · 2017-04-20T14:37:29Z

Thank you @ywelsch.

Refactoring that extracts the task batching functionality from ClusterService and makes it a reusable component that can be tested in isolation.

…ecutor (#24237) Changes in #24102 exposed the following oddity: PrioritizedEsThreadPoolExecutor.getPending() can return Pending entries where pending.task == null. This can happen for example when tasks are added to the pending list while they are in the clean up phase, i.e. TieBreakingPrioritizedRunnable#runAndClean has run already, but afterExecute has not removed the task yet. Instead of safeguarding consumers of the API (as was done before #24102) this changes the executor to not count these tasks as pending at all.

* master: (61 commits) Build: Move plugin cli and tests to distribution tool (elastic#24220) Peer Recovery: remove maxUnsafeAutoIdTimestamp hand off (elastic#24243) Adds version 5.3.2 and backwards compatibility indices for 5.3.1 Add utility method to parse named XContent objects with typed prefix (elastic#24240) MultiBucketsAggregation.Bucket should not extend Writeable (elastic#24216) Don't expose cleaned-up tasks as pending in PrioritizedEsThreadPoolExecutor (elastic#24237) Adds declareNamedObjects methods to ConstructingObjectParser (elastic#24219) ESIntegTestCase.indexRandom should not introduce types. (elastic#24202) Tests: Extend InternalStatsTests (elastic#24212) IndicesQueryCache should delegate the scorerSupplier method. (elastic#24209) Speed up parsing of large `terms` queries. (elastic#24210) [TEST] make sure that the random query_string query generator defines a default_field or a list of fields token_count type : add an option to count tokens (fix elastic#23227) (elastic#24175) Query string default field (elastic#24214) Make Aggregations an abstract class rather than an interface (elastic#24184) [TEST] ensure expected sequence no and version are set when index/delete engine operation has a document failure Extract batch executor out of cluster service (elastic#24102) Add 5.3.1 to bwc versions Added "release-state" support to plugin docs Added examples to cross cluster search of using cluster settings ...

…ecutor (elastic#24237) Changes in elastic#24102 exposed the following oddity: PrioritizedEsThreadPoolExecutor.getPending() can return Pending entries where pending.task == null. This can happen for example when tasks are added to the pending list while they are in the clean up phase, i.e. TieBreakingPrioritizedRunnable#runAndClean has run already, but afterExecute has not removed the task yet. Instead of safeguarding consumers of the API (as was done before elastic#24102) this changes the executor to not count these tasks as pending at all.

Extract batch executor out of cluster service

4653e82

ywelsch added :Cluster >non-issue v5.4.0 v6.0.0-alpha1 labels Apr 14, 2017

ywelsch requested review from bleskes and jasontedor April 14, 2017 07:20

ywelsch added the v5.5.0 label Apr 14, 2017

ywelsch added 3 commits April 19, 2017 19:47

Remove SingleTaskExecutor

c9bd097

javadoc and smaller fixes

58f0235

make some variables private

abe574f

jasontedor reviewed Apr 20, 2017

View reviewed changes

bleskes reviewed Apr 20, 2017

View reviewed changes

address review comments

a1e1ec4

bleskes approved these changes Apr 20, 2017

View reviewed changes

jasontedor reviewed Apr 20, 2017

View reviewed changes

fix variable naming

98a4117

jasontedor approved these changes Apr 20, 2017

View reviewed changes

ywelsch merged commit 22e0795 into elastic:master Apr 20, 2017

ywelsch added a commit that referenced this pull request Apr 20, 2017

Extract batch executor out of cluster service (#24102)

86f2e1e

Refactoring that extracts the task batching functionality from ClusterService and makes it a reusable component that can be tested in isolation.

ywelsch added a commit that referenced this pull request Apr 20, 2017

Extract batch executor out of cluster service (#24102)

8db5a6d

Refactoring that extracts the task batching functionality from ClusterService and makes it a reusable component that can be tested in isolation.

ywelsch mentioned this pull request Apr 21, 2017

Don't expose cleaned-up tasks as pending in PrioritizedEsThreadPoolExecutor #24237

Merged

clintongormley added :Distributed Indexing/Distributed A catch all label for anything in the Distributed Area. Please avoid if you can. and removed :Cluster labels Feb 13, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Extract batch executor out of cluster service #24102

Extract batch executor out of cluster service #24102

ywelsch commented Apr 14, 2017

jasontedor left a comment

jasontedor Apr 20, 2017

bleskes Apr 20, 2017

jasontedor Apr 20, 2017

jasontedor Apr 20, 2017

bleskes Apr 20, 2017

jasontedor Apr 20, 2017

jasontedor Apr 20, 2017

jasontedor Apr 20, 2017

bleskes left a comment

bleskes Apr 18, 2017

bleskes Apr 20, 2017

bleskes Apr 20, 2017

bleskes Apr 20, 2017

bleskes Apr 20, 2017

bleskes Apr 20, 2017

bleskes Apr 20, 2017

bleskes Apr 20, 2017

bleskes Apr 20, 2017

ywelsch Apr 20, 2017

bleskes Apr 20, 2017

ywelsch commented Apr 20, 2017

bleskes left a comment

jasontedor left a comment •

edited

Loading

jasontedor Apr 20, 2017

ywelsch Apr 20, 2017

jasontedor left a comment

jasontedor commented Apr 20, 2017

Extract batch executor out of cluster service #24102

Extract batch executor out of cluster service #24102

Conversation

ywelsch commented Apr 14, 2017

jasontedor left a 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

bleskes left a 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

ywelsch commented Apr 20, 2017

bleskes left a comment

Choose a reason for hiding this comment

jasontedor left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jasontedor left a comment

Choose a reason for hiding this comment

jasontedor commented Apr 20, 2017

jasontedor left a comment •

edited

Loading