Skip to content

Commit 2b47d67

Browse files
jasontedorjasontedor
authored
Remove the index thread pool (#29556)
Now that single-document indexing requests are executed on the bulk thread pool the index thread pool is no longer needed. This commit removes this thread pool from Elasticsearch.
1 parent 9d11c7a commit 2b47d67

File tree

17 files changed

+42
-102
lines changed

17 files changed

+42
-102
lines changed

‎docs/reference/cat/thread_pool.asciidoc

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ node-0 flush 0 0 0
2222
node-0 force_merge 0 0 0
2323
node-0 generic 0 0 0
2424
node-0 get 0 0 0
25-
node-0 index 0 0 0
2625
node-0 listener 0 0 0
2726
node-0 management 1 0 0
2827
node-0 refresh 0 0 0
@@ -52,7 +51,6 @@ flush
5251
force_merge
5352
generic
5453
get
55-
index
5654
listener
5755
management
5856
refresh

‎docs/reference/migration/migrate_7_0/settings.asciidoc

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,4 +5,12 @@
55
==== Percolator
66

77
* The deprecated `index.percolator.map_unmapped_fields_as_string` setting has been removed in favour of
8-
the `index.percolator.map_unmapped_fields_as_text` setting.
8+
the `index.percolator.map_unmapped_fields_as_text` setting.
9+
10+
==== Index thread pool
11+
12+
* Internally, single-document index/delete/update requests are executed as bulk
13+
requests with a single-document payload. This means that these requests are
14+
executed on the bulk thread pool. As such, the indexing thread pool is no
15+
longer needed and has been removed. As such, the settings
16+
`thread_pool.index.size` and `thread_pool.index.queue_size` have been removed.

‎docs/reference/modules/threadpool.asciidoc

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -13,12 +13,6 @@ There are several thread pools, but the important ones include:
1313
For generic operations (e.g., background node discovery).
1414
Thread pool type is `scaling`.
1515

16-
`index`::
17-
For index/delete operations. Thread pool type is `fixed`
18-
with a size of `# of available processors`,
19-
queue_size of `200`. The maximum size for this pool
20-
is `1 + # of available processors`.
21-
2216
`search`::
2317
For count/search/suggest operations. Thread pool type is
2418
`fixed_auto_queue_size` with a size of
@@ -55,13 +49,13 @@ There are several thread pools, but the important ones include:
5549
Mainly for java client executing of action when listener threaded is set to true.
5650
Thread pool type is `scaling` with a default max of `min(10, (# of available processors)/2)`.
5751

58-
Changing a specific thread pool can be done by setting its type-specific parameters; for example, changing the `index`
52+
Changing a specific thread pool can be done by setting its type-specific parameters; for example, changing the `bulk`
5953
thread pool to have more threads:
6054

6155
[source,yaml]
6256
--------------------------------------------------
6357
thread_pool:
64-
index:
58+
bulk:
6559
size: 30
6660
--------------------------------------------------
6761

@@ -89,7 +83,7 @@ full, it will abort the request.
8983
[source,yaml]
9084
--------------------------------------------------
9185
thread_pool:
92-
index:
86+
bulk:
9387
size: 30
9488
queue_size: 1000
9589
--------------------------------------------------

‎rest-api-spec/src/main/resources/rest-api-spec/test/cat.thread_pool/10_basic.yml

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@
3333
3434
- do:
3535
cat.thread_pool:
36-
thread_pool_patterns: bulk,management,flush,index,generic,force_merge
36+
thread_pool_patterns: bulk,management,flush,generic,force_merge
3737
h: id,name,active
3838
v: true
3939

@@ -44,7 +44,6 @@
4444
\S+\s+ flush \s+ \d+ \n
4545
\S+\s+ force_merge \s+ \d+ \n
4646
\S+\s+ generic \s+ \d+ \n
47-
\S+\s+ index \s+ \d+ \n
4847
\S+\s+ management \s+ \d+ \n)+ $/
4948
5049
- do:
@@ -72,12 +71,11 @@
7271
7372
- do:
7473
cat.thread_pool:
75-
thread_pool_patterns: bulk,index,search
74+
thread_pool_patterns: bulk,search
7675
size: ""
7776

7877
- match:
7978
$body: |
8079
/ #node_name name active queue rejected
8180
^ (\S+ \s+ bulk \s+ \d+ \s+ \d+ \s+ \d+ \n
82-
\S+ \s+ index \s+ \d+ \s+ \d+ \s+ \d+ \n
8381
\S+ \s+ search \s+ \d+ \s+ \d+ \s+ \d+ \n)+ $/

‎server/src/main/java/org/elasticsearch/action/delete/TransportDeleteAction.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ public TransportDeleteAction(Settings settings, TransportService transportServic
4646
ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
4747
TransportBulkAction bulkAction, TransportShardBulkAction shardBulkAction) {
4848
super(settings, DeleteAction.NAME, transportService, clusterService, indicesService, threadPool, shardStateAction,
49-
actionFilters, indexNameExpressionResolver, DeleteRequest::new, DeleteRequest::new, ThreadPool.Names.INDEX,
49+
actionFilters, indexNameExpressionResolver, DeleteRequest::new, DeleteRequest::new, ThreadPool.Names.BULK,
5050
bulkAction, shardBulkAction);
5151
}
5252

‎server/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,7 @@ public TransportIndexAction(Settings settings, TransportService transportService
5454
ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
5555
TransportBulkAction bulkAction, TransportShardBulkAction shardBulkAction) {
5656
super(settings, IndexAction.NAME, transportService, clusterService, indicesService, threadPool, shardStateAction,
57-
actionFilters, indexNameExpressionResolver, IndexRequest::new, IndexRequest::new, ThreadPool.Names.INDEX,
57+
actionFilters, indexNameExpressionResolver, IndexRequest::new, IndexRequest::new, ThreadPool.Names.BULK,
5858
bulkAction, shardBulkAction);
5959
}
6060

‎server/src/main/java/org/elasticsearch/action/update/TransportUpdateAction.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -86,7 +86,7 @@ public TransportUpdateAction(Settings settings, ThreadPool threadPool, ClusterSe
8686

8787
@Override
8888
protected String executor() {
89-
return ThreadPool.Names.INDEX;
89+
return ThreadPool.Names.BULK;
9090
}
9191

9292
@Override

‎server/src/main/java/org/elasticsearch/threadpool/ExecutorBuilder.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ protected static String settingsKey(final String prefix, final String key) {
4848
}
4949

5050
protected int applyHardSizeLimit(final Settings settings, final String name) {
51-
if (name.equals(ThreadPool.Names.BULK) || name.equals(ThreadPool.Names.INDEX)) {
51+
if (name.equals(ThreadPool.Names.BULK)) {
5252
return 1 + EsExecutors.numberOfProcessors(settings);
5353
} else {
5454
return Integer.MAX_VALUE;

‎server/src/main/java/org/elasticsearch/threadpool/FixedExecutorBuilder.java

Lines changed: 3 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -49,20 +49,7 @@ public final class FixedExecutorBuilder extends ExecutorBuilder<FixedExecutorBui
4949
* @param queueSize the size of the backing queue, -1 for unbounded
5050
*/
5151
FixedExecutorBuilder(final Settings settings, final String name, final int size, final int queueSize) {
52-
this(settings, name, size, queueSize, false);
53-
}
54-
55-
/**
56-
* Construct a fixed executor builder; the settings will have the key prefix "thread_pool." followed by the executor name.
57-
*
58-
* @param settings the node-level settings
59-
* @param name the name of the executor
60-
* @param size the fixed number of threads
61-
* @param queueSize the size of the backing queue, -1 for unbounded
62-
* @param deprecated whether or not the thread pool is deprecated
63-
*/
64-
FixedExecutorBuilder(final Settings settings, final String name, final int size, final int queueSize, final boolean deprecated) {
65-
this(settings, name, size, queueSize, "thread_pool." + name, deprecated);
52+
this(settings, name, size, queueSize, "thread_pool." + name);
6653
}
6754

6855
/**
@@ -75,41 +62,16 @@ public final class FixedExecutorBuilder extends ExecutorBuilder<FixedExecutorBui
7562
* @param prefix the prefix for the settings keys
7663
*/
7764
public FixedExecutorBuilder(final Settings settings, final String name, final int size, final int queueSize, final String prefix) {
78-
this(settings, name, size, queueSize, prefix, false);
79-
}
80-
81-
/**
82-
* Construct a fixed executor builder.
83-
*
84-
* @param settings the node-level settings
85-
* @param name the name of the executor
86-
* @param size the fixed number of threads
87-
* @param queueSize the size of the backing queue, -1 for unbounded
88-
* @param prefix the prefix for the settings keys
89-
*/
90-
private FixedExecutorBuilder(
91-
final Settings settings,
92-
final String name,
93-
final int size,
94-
final int queueSize,
95-
final String prefix,
96-
final boolean deprecated) {
9765
super(name);
9866
final String sizeKey = settingsKey(prefix, "size");
99-
final Setting.Property[] properties;
100-
if (deprecated) {
101-
properties = new Setting.Property[]{Setting.Property.NodeScope, Setting.Property.Deprecated};
102-
} else {
103-
properties = new Setting.Property[]{Setting.Property.NodeScope};
104-
}
10567
this.sizeSetting =
10668
new Setting<>(
10769
sizeKey,
10870
s -> Integer.toString(size),
10971
s -> Setting.parseInt(s, 1, applyHardSizeLimit(settings, name), sizeKey),
110-
properties);
72+
Setting.Property.NodeScope);
11173
final String queueSizeKey = settingsKey(prefix, "queue_size");
112-
this.queueSizeSetting = Setting.intSetting(queueSizeKey, queueSize, properties);
74+
this.queueSizeSetting = Setting.intSetting(queueSizeKey, queueSize, Setting.Property.NodeScope);
11375
}
11476

11577
@Override

‎server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,6 @@ public static class Names {
6969
public static final String LISTENER = "listener";
7070
public static final String GET = "get";
7171
public static final String ANALYZE = "analyze";
72-
public static final String INDEX = "index";
7372
public static final String BULK = "bulk";
7473
public static final String SEARCH = "search";
7574
public static final String MANAGEMENT = "management";
@@ -126,7 +125,6 @@ public static ThreadPoolType fromType(String type) {
126125
map.put(Names.LISTENER, ThreadPoolType.FIXED);
127126
map.put(Names.GET, ThreadPoolType.FIXED);
128127
map.put(Names.ANALYZE, ThreadPoolType.FIXED);
129-
map.put(Names.INDEX, ThreadPoolType.FIXED);
130128
map.put(Names.BULK, ThreadPoolType.FIXED);
131129
map.put(Names.SEARCH, ThreadPoolType.FIXED_AUTO_QUEUE_SIZE);
132130
map.put(Names.MANAGEMENT, ThreadPoolType.SCALING);
@@ -172,7 +170,6 @@ public ThreadPool(final Settings settings, final ExecutorBuilder<?>... customBui
172170
final int halfProcMaxAt10 = halfNumberOfProcessorsMaxTen(availableProcessors);
173171
final int genericThreadPoolMax = boundedBy(4 * availableProcessors, 128, 512);
174172
builders.put(Names.GENERIC, new ScalingExecutorBuilder(Names.GENERIC, 4, genericThreadPoolMax, TimeValue.timeValueSeconds(30)));
175-
builders.put(Names.INDEX, new FixedExecutorBuilder(settings, Names.INDEX, availableProcessors, 200, true));
176173
builders.put(Names.BULK, new FixedExecutorBuilder(settings, Names.BULK, availableProcessors, 200)); // now that we reuse bulk for index/delete ops
177174
builders.put(Names.GET, new FixedExecutorBuilder(settings, Names.GET, availableProcessors, 1000));
178175
builders.put(Names.ANALYZE, new FixedExecutorBuilder(settings, Names.ANALYZE, 1, 16));

‎server/src/test/java/org/elasticsearch/action/RejectionActionIT.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,8 +45,8 @@ protected Settings nodeSettings(int nodeOrdinal) {
4545
.put(super.nodeSettings(nodeOrdinal))
4646
.put("thread_pool.search.size", 1)
4747
.put("thread_pool.search.queue_size", 1)
48-
.put("thread_pool.index.size", 1)
49-
.put("thread_pool.index.queue_size", 1)
48+
.put("thread_pool.bulk.size", 1)
49+
.put("thread_pool.bulk.queue_size", 1)
5050
.put("thread_pool.get.size", 1)
5151
.put("thread_pool.get.queue_size", 1)
5252
.build();

‎server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionIngestTests.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -124,7 +124,7 @@ class TestSingleItemBulkWriteAction extends TransportSingleItemBulkWriteAction<I
124124
super(Settings.EMPTY, IndexAction.NAME, TransportBulkActionIngestTests.this.transportService,
125125
TransportBulkActionIngestTests.this.clusterService,
126126
null, null, null, new ActionFilters(Collections.emptySet()), null,
127-
IndexRequest::new, IndexRequest::new, ThreadPool.Names.INDEX, bulkAction, null);
127+
IndexRequest::new, IndexRequest::new, ThreadPool.Names.BULK, bulkAction, null);
128128
}
129129

130130
@Override

‎server/src/test/java/org/elasticsearch/index/replication/ESIndexLevelReplicationTestCase.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -542,7 +542,7 @@ public void onFailure(Exception e) {
542542
listener.onFailure(e);
543543
}
544544
},
545-
ThreadPool.Names.INDEX, request);
545+
ThreadPool.Names.BULK, request);
546546
}
547547

548548
@Override

‎server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -284,14 +284,14 @@ public void testClosesPreventsNewOperations() throws InterruptedException, Execu
284284
closeShards(indexShard);
285285
assertThat(indexShard.getActiveOperationsCount(), equalTo(0));
286286
try {
287-
indexShard.acquirePrimaryOperationPermit(null, ThreadPool.Names.INDEX, "");
287+
indexShard.acquirePrimaryOperationPermit(null, ThreadPool.Names.BULK, "");
288288
fail("we should not be able to increment anymore");
289289
} catch (IndexShardClosedException e) {
290290
// expected
291291
}
292292
try {
293293
indexShard.acquireReplicaOperationPermit(indexShard.getPrimaryTerm(), SequenceNumbers.UNASSIGNED_SEQ_NO, null,
294-
ThreadPool.Names.INDEX, "");
294+
ThreadPool.Names.BULK, "");
295295
fail("we should not be able to increment anymore");
296296
} catch (IndexShardClosedException e) {
297297
// expected
@@ -302,7 +302,7 @@ public void testRejectOperationPermitWithHigherTermWhenNotStarted() throws IOExc
302302
IndexShard indexShard = newShard(false);
303303
expectThrows(IndexShardNotStartedException.class, () ->
304304
indexShard.acquireReplicaOperationPermit(indexShard.getPrimaryTerm() + randomIntBetween(1, 100),
305-
SequenceNumbers.UNASSIGNED_SEQ_NO, null, ThreadPool.Names.INDEX, ""));
305+
SequenceNumbers.UNASSIGNED_SEQ_NO, null, ThreadPool.Names.BULK, ""));
306306
closeShards(indexShard);
307307
}
308308

@@ -342,7 +342,7 @@ public void onFailure(Exception e) {
342342
throw new RuntimeException(e);
343343
}
344344
},
345-
ThreadPool.Names.INDEX, id);
345+
ThreadPool.Names.BULK, id);
346346
});
347347
thread.start();
348348
threads.add(thread);
@@ -393,7 +393,7 @@ public void onFailure(Exception e) {
393393
throw new RuntimeException(e);
394394
}
395395
},
396-
ThreadPool.Names.INDEX, id);
396+
ThreadPool.Names.BULK, id);
397397
});
398398
thread.start();
399399
delayedThreads.add(thread);
@@ -589,7 +589,7 @@ public void testOperationPermitsOnPrimaryShards() throws InterruptedException, E
589589
assertEquals(0, indexShard.getActiveOperationsCount());
590590
if (indexShard.routingEntry().isRelocationTarget() == false) {
591591
try {
592-
indexShard.acquireReplicaOperationPermit(primaryTerm, indexShard.getGlobalCheckpoint(), null, ThreadPool.Names.INDEX, "");
592+
indexShard.acquireReplicaOperationPermit(primaryTerm, indexShard.getGlobalCheckpoint(), null, ThreadPool.Names.BULK, "");
593593
fail("shard shouldn't accept operations as replica");
594594
} catch (IllegalStateException ignored) {
595595

@@ -608,14 +608,14 @@ public void testOperationPermitsOnPrimaryShards() throws InterruptedException, E
608608

609609
private Releasable acquirePrimaryOperationPermitBlockingly(IndexShard indexShard) throws ExecutionException, InterruptedException {
610610
PlainActionFuture<Releasable> fut = new PlainActionFuture<>();
611-
indexShard.acquirePrimaryOperationPermit(fut, ThreadPool.Names.INDEX, "");
611+
indexShard.acquirePrimaryOperationPermit(fut, ThreadPool.Names.BULK, "");
612612
return fut.get();
613613
}
614614

615615
private Releasable acquireReplicaOperationPermitBlockingly(IndexShard indexShard, long opPrimaryTerm)
616616
throws ExecutionException, InterruptedException {
617617
PlainActionFuture<Releasable> fut = new PlainActionFuture<>();
618-
indexShard.acquireReplicaOperationPermit(opPrimaryTerm, indexShard.getGlobalCheckpoint(), fut, ThreadPool.Names.INDEX, "");
618+
indexShard.acquireReplicaOperationPermit(opPrimaryTerm, indexShard.getGlobalCheckpoint(), fut, ThreadPool.Names.BULK, "");
619619
return fut.get();
620620
}
621621

@@ -663,7 +663,7 @@ public void testOperationPermitOnReplicaShards() throws Exception {
663663
if (shardRouting.primary() == false) {
664664
final IllegalStateException e =
665665
expectThrows(IllegalStateException.class,
666-
() -> indexShard.acquirePrimaryOperationPermit(null, ThreadPool.Names.INDEX, ""));
666+
() -> indexShard.acquirePrimaryOperationPermit(null, ThreadPool.Names.BULK, ""));
667667
assertThat(e, hasToString(containsString("shard " + shardRouting + " is not a primary")));
668668
}
669669

@@ -700,7 +700,7 @@ public void onFailure(Exception e) {
700700
};
701701

702702
indexShard.acquireReplicaOperationPermit(primaryTerm - 1, SequenceNumbers.UNASSIGNED_SEQ_NO, onLockAcquired,
703-
ThreadPool.Names.INDEX, "");
703+
ThreadPool.Names.BULK, "");
704704

705705
assertFalse(onResponse.get());
706706
assertTrue(onFailure.get());
@@ -1020,7 +1020,7 @@ public void onFailure(Exception e) {
10201020
latch.countDown();
10211021
}
10221022
},
1023-
ThreadPool.Names.INDEX, "");
1023+
ThreadPool.Names.BULK, "");
10241024
};
10251025

10261026
final long firstIncrement = 1 + (randomBoolean() ? 0 : 1);
@@ -1381,7 +1381,7 @@ public void onResponse(Releasable releasable) {
13811381
super.onResponse(releasable);
13821382
}
13831383
};
1384-
shard.acquirePrimaryOperationPermit(onLockAcquired, ThreadPool.Names.INDEX, "i_" + i);
1384+
shard.acquirePrimaryOperationPermit(onLockAcquired, ThreadPool.Names.BULK, "i_" + i);
13851385
onLockAcquiredActions.add(onLockAcquired);
13861386
}
13871387

‎server/src/test/java/org/elasticsearch/indices/flush/SyncedFlushSingleNodeTests.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -113,7 +113,7 @@ public void testSyncFailsIfOperationIsInFlight() throws InterruptedException, Ex
113113
SyncedFlushService flushService = getInstanceFromNode(SyncedFlushService.class);
114114
final ShardId shardId = shard.shardId();
115115
PlainActionFuture<Releasable> fut = new PlainActionFuture<>();
116-
shard.acquirePrimaryOperationPermit(fut, ThreadPool.Names.INDEX, "");
116+
shard.acquirePrimaryOperationPermit(fut, ThreadPool.Names.BULK, "");
117117
try (Releasable operationLock = fut.get()) {
118118
SyncedFlushUtil.LatchedListener<ShardsSyncedFlushResult> listener = new SyncedFlushUtil.LatchedListener<>();
119119
flushService.attemptSyncedFlush(shardId, listener);

‎server/src/test/java/org/elasticsearch/threadpool/FixedThreadPoolTests.java

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -85,10 +85,6 @@ public void testRejectedExecutionCounter() throws InterruptedException {
8585

8686
assertThat(counter, equalTo(rejections));
8787
assertThat(stats(threadPool, threadPoolName).getRejected(), equalTo(rejections));
88-
89-
if (threadPoolName.equals(ThreadPool.Names.INDEX)) {
90-
assertSettingDeprecationsAndWarnings(new String[]{"thread_pool.index.queue_size", "thread_pool.index.size"});
91-
}
9288
} finally {
9389
terminateThreadPoolIfNeeded(threadPool);
9490
}

0 commit comments

Comments
 (0)