Make index and delete operation execute as a single bulk item #21964

areek · 2016-12-05T04:58:52Z

Performance testing by @danielmitterdorfer revealed single
index/delete operations have similar performance (indexing
throughput) to equivalent single item bulk request.
This PR reduces the code paths to executing single write
operations, by reusing the logic in (shard) bulk action for
executing single operation as a single-item bulk request.

bleskes · 2016-12-05T08:09:51Z

@areek thanks. I quickly looked at it and I have a question. What happens when people issue an index request now via the transport client? I believe they still go through the the transportIndexAction mechanics? we need to make sure it's translated and we never send a reroute request to the primaries (I expected some change in the execute method of that action). Also I think we need to change the way the rest layer work (we can do it as a follow up if you prefer)

areek · 2016-12-06T05:10:35Z

Thanks for the feedback @bleskes!

What happens when people issue an index request now via the transport client? I believe they still go through the the transportIndexAction mechanics?

when a transport client issues an index request it still goes through transportIndexAction (just like it does for node clients). The only difference in this PR is we use shardBulk action to execute the actual operation.

we need to make sure it's translated and we never send a reroute request to the primaries (I expected some change in the execute method of that action)

I took this approach initially, i.e. delegate to bulk action on the transportIndexAction's doExecute. But tests started failing as bulk action is not a replicated action and as a result behaves differently (e.g. handling cluster state global blocks, in bulk action we get an exception outright (TransportBulkAction#209) whereas for replication action we retry on global blocks (TransportReplicationAction#737). Should we change bulk action to have similar behaviour for 'reroute' phase?

areek · 2016-12-07T05:33:33Z

@bleskes I have updated the PR to use single-item bulk action to execute index and delete operation. Now we delegate single write operation requests to bulk action in doExecute method, this simplified TransportIndexAction & TransportDeleteAction s to HandledTransportAction. There are a few functional changes in the PR, namely:

TransportBulkAction now retries retryable cluster blocks (to be consistent with TransportReplicationAction)
The toString representation of IndexRequest and DeleteRequest has changed to be the same as a single item ShardBulkRequest.toString, should we special case this? This might lead to confusion while debugging
Temporarily I increased the threadpool size for Bulk to make tests pass from 50 to 200 (default size for index threadpool). As a side, now only update action uses the Index threadpool
All tests pass except IndexWithShadowReplicasIT, am still investigating the cause.

As TransportIndexAction and TransportDeleteActions are handled transport actions, TransportShardBulkAction is the only implementation of TransportWriteAction. Do we really need a base TransportWriteAction class anymore?

@danielmitterdorfer

Performance testing by @danielmitterdorfer revealed single index/delete operations have similar performance (indexing throughput) to equivalent single item bulk request. This PR reduces the code paths to executing single write operations, by reusing the logic in (shard) bulk action for executing single operation as a single-item bulk request.

martijnvg · 2016-12-07T08:20:08Z

@areek I think with this change, we don't need the logic for IndexRequests in IngestActionFilter? (switch case for index action in apply() method, processIndexRequest() method and PipelineExecutionService#executeIndexRequest(...) method). If it isn't that much work, maybe we can remove it as part of this change?

martijnvg · 2016-12-07T08:43:53Z

core/src/main/java/org/elasticsearch/threadpool/ThreadPool.java

        final int halfProcMaxAt10 = halfNumberOfProcessorsMaxTen(availableProcessors);
        final int genericThreadPoolMax = boundedBy(4 * availableProcessors, 128, 512);
        builders.put(Names.GENERIC, new ScalingExecutorBuilder(Names.GENERIC, 4, genericThreadPoolMax, TimeValue.timeValueSeconds(30)));
        builders.put(Names.INDEX, new FixedExecutorBuilder(settings, Names.INDEX, availableProcessors, 200));


Is there a reason to keep INDEX TP around?

INDEX TP is currently only used by TransportUpdateAction. I agree that we should remove it or at least rename it to better reflect where it is used.

I see, right, renaming it to update would be better.

I think we can migrate TransportUpdateAction to redirect to a single item bulk as well (as a follow up). At that point we can rename bulk to index :)

areek · 2016-12-07T15:29:24Z

thanks @martijnvg for the comment, I will remove the logic index requests logic in IngestActionFilter

areek · 2016-12-09T06:04:39Z

@bleskes Can you take a look, I reverted index and delete transport actions to be TransportWriteActions and now it delegates the shardOperationOnPrimary/Replica to reuse the shard bulk logic for performing single-item shard bulks. This was done to ensure bwc, such that incoming primary/replica single-operation-write-requests from pre-6.0 nodes are executed as expected, while index/delete requests from 6.0 nodes are delegated to TransportBulk and avoids the replication reroute phase entirely.

bleskes

This looks great. Left very minor comments and questions. I also asked @jaymode and @martijnvg for some validation.

bleskes · 2016-12-20T13:18:40Z

core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

    }

+    public <Request extends ReplicatedWriteRequest<Request>, Response extends ReplicationResponse & WriteResponse>
+    WritePrimaryResult<Request, Response> executeSingleItemBulkRequestOnPrimary(


wondering if we can put this something else .. maybe a static method on TransportIndexAction ? It's a shame to cluster this class with BWC code..

I have moved these BWC method to static methods to SingleWriteOperationUtility, (suggestion for better name welcomed). The only downside is now shardOperationOnPrimary and shardOperationOnReplica for bulk shard action is made public, as they are called from transport index and delete actions

bleskes · 2016-12-20T13:19:20Z

core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

+            Request request, IndexShard primary) throws Exception {
+        BulkItemRequest[] itemRequests = new BulkItemRequest[1];
+        WriteRequest.RefreshPolicy refreshPolicy = request.getRefreshPolicy();
+        request.setRefreshPolicy(WriteRequest.RefreshPolicy.NONE);


feels weird to mutate the incoming request here. Why is it needed?

thanks for pointing it out. This was left over from using the bulk action directly, I removed mutating the request

bleskes · 2016-12-20T13:27:01Z

core/src/main/java/org/elasticsearch/action/delete/TransportDeleteAction.java

 * Performs the delete operation.
 */
-public class TransportDeleteAction extends TransportWriteAction<DeleteRequest, DeleteRequest,DeleteResponse> {
+public class TransportDeleteAction extends TransportWriteAction<DeleteRequest, DeleteRequest, DeleteResponse> {


wondering - should we mark this as deprecated, so we want use internally? not sure how much it buys us.

I marked both transport index/delete action as deprecated, now the only other usage of these actions are in TransportUpdateAction. should we use the bulk action directly as a follow up?

bleskes · 2016-12-20T13:31:31Z

core/src/main/java/org/elasticsearch/action/delete/TransportDeleteAction.java


    @Override
    protected void doExecute(Task task, final DeleteRequest request, final ActionListener<DeleteResponse> listener) {
-        ClusterState state = clusterService.state();


Since we can't override the execute method itself, I think it will be good to get a sanity check from @jaymode and @martijnvg about potential implications of running the request filter chain twice

bleskes · 2016-12-20T13:33:07Z

core/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java

+        bulkAction.execute(task, bulkRequest, new ActionListener<BulkResponse>() {
+            @Override
+            public void onResponse(BulkResponse bulkItemResponses) {
+                assert bulkItemResponses.getItems().length == 1: "expected only one item in bulk request";


since this is also the same code here as in delete, I wonder if we should add a generic BWC class that will do all this shared munging (and the code from the bulk action as well).

I converted them to static helper functions and moved them to SingleWriteOperationUtilty

bleskes · 2016-12-20T13:34:22Z

core/src/main/java/org/elasticsearch/action/support/replication/TransportWriteAction.java

     * Result of taking the action on the primary.
     */
-    protected class WritePrimaryResult extends PrimaryResult implements RespondingWriteResult {
+    protected static class WritePrimaryResult<ReplicaRequest extends ReplicatedWriteRequest<ReplicaRequest>,


wondering - why was this needed (the change to static and the explicit generic references)?

this is needed because we deal with two types of WritePrimaryResult in indexAction/deleteAction.shardOperationOnPrimary/Replica. WritePrimaryResult with type BulkShardRequest/BulkShardResponse and Index/delete request and response.

bleskes · 2016-12-20T13:37:28Z

core/src/main/java/org/elasticsearch/threadpool/ThreadPool.java

        final int halfProcMaxAt10 = halfNumberOfProcessorsMaxTen(availableProcessors);
        final int genericThreadPoolMax = boundedBy(4 * availableProcessors, 128, 512);
        builders.put(Names.GENERIC, new ScalingExecutorBuilder(Names.GENERIC, 4, genericThreadPoolMax, TimeValue.timeValueSeconds(30)));
        builders.put(Names.INDEX, new FixedExecutorBuilder(settings, Names.INDEX, availableProcessors, 200));


I think we can migrate TransportUpdateAction to redirect to a single item bulk as well (as a follow up). At that point we can rename bulk to index :)

bleskes · 2016-12-20T13:39:16Z

core/src/test/java/org/elasticsearch/action/index/TransportIndexActionIngestTests.java

-import static org.mockito.Mockito.verifyZeroInteractions;
-import static org.mockito.Mockito.when;
-
-public class TransportIndexActionIngestTests extends ESTestCase {


do we have any test that makes sure that single item index ops work with ingest?

There is already test for ingest with bulk action (TransportBulkActionIngestTests) which covers ingest with single item ops

I think we still need a way to make sure igest via single item index works?

bleskes · 2016-12-20T13:44:00Z

core/src/test/java/org/elasticsearch/cluster/NoMasterNodeIT.java

+        checkWriteAction(timeout,
                client().prepareIndex("test", "type1", "1").setSource(XContentFactory.jsonBuilder().startObject().endObject()).setTimeout(timeout));

-        checkWriteAction(autoCreateIndex, timeout,


why can we remove autoCreateIndex? I mean it can still be created? what am I missing?

This is actually due to TransportUpdateAction behaving differently (on how it handles cluster blocks when autoCreateIndex is true, instead of throwing cluster block exception, transport update throws master not discovered exception). IMO, update action should be fixed to be consistent here and throw a cluster block exception instead. I re-worked the tests, so the distinction is clear.

bleskes · 2016-12-20T13:45:20Z

core/src/test/java/org/elasticsearch/index/mapper/DynamicMappingDisabledTests.java

-    private IndexNameExpressionResolver indexNameExpressionResolver;
-    private AutoCreateIndex autoCreateIndex;
-    private Settings settings;
+    private TransportIndexAction transportIndexAction;


we should stop using it :) - maybe have a test utility to index a single item using a bulk action?

I changed it to using bulk action instead

areek · 2016-12-21T06:04:52Z

Thanks @bleskes for the feedback, I addressed your comments. After running bwc tests backward-5.0 for many iterations, I found a bug where the listener hangs indefinitely when the refresh flag is set to WAIT_FOR (one of rest-tests testing refresh for bulk, index and delete seemed to fail). I couldn't reproduce this running the rest-test on only 6.0 cluster. I am investigating the code flow in TransportWriteAction.AsyncAfterWriteAction but still no luck.

bleskes · 2016-12-21T12:21:34Z

thx @areek . I played with this a bit and came to this: areek/elasticsearch@enhancement/use_shard_bulk_for_single_ops...bleskes:use_shard_bulk_for_single_ops

What do you think?

Also - I think the wait for refresh flag issues are unrelated. They have been showing up on normal CI. I started chasing them too

bleskes · 2016-12-22T08:57:42Z

retest this please

bleskes

LGTM. @martijnvg do you mind taking a look at the ingest tests?

bleskes · 2016-12-22T09:56:09Z

@danielmitterdorfer has gracefully agreed to benchmark this PR to validate that performance is comparable to his initial research

martijnvg

Looks good, left one minor comment.

martijnvg · 2016-12-22T10:42:01Z

core/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionIngestTests.java

+        completionHandler.getValue().accept(null);
+        assertTrue(action.isExecuted);
+        assertFalse(responseCalled.get()); // listener would only be called by real index action, not our mocked one
+        verifyZeroInteractions(transportService);


I think we should also verify that interaction happened with ingestService mock here?

@bleskes:

Index: core/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionIngestTests.java IDEA additional info: Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP <+>UTF-8 =================================================================== --- core/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionIngestTests.java (date 1482389279000) +++ core/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionIngestTests.java (revision ) @@ -28,15 +28,12 @@ import org.elasticsearch.cluster.ClusterChangedEvent; import org.elasticsearch.cluster.ClusterState; import org.elasticsearch.cluster.ClusterStateApplier; -import org.elasticsearch.cluster.action.shard.ShardStateAction; -import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver; import org.elasticsearch.cluster.node.DiscoveryNode; import org.elasticsearch.cluster.node.DiscoveryNodes; import org.elasticsearch.cluster.service.ClusterService; import org.elasticsearch.common.collect.ImmutableOpenMap; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.util.concurrent.AtomicArray; -import org.elasticsearch.indices.IndicesService; import org.elasticsearch.ingest.IngestService; import org.elasticsearch.ingest.PipelineExecutionService; import org.elasticsearch.tasks.Task; @@ -54,7 +51,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.BiConsumer; import java.util.function.Consumer; -import java.util.function.Supplier; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.sameInstance; @@ -64,6 +60,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyZeroInteractions; import static org.mockito.Mockito.when; @@ -224,6 +221,7 @@ verify(executionService).executeBulkRequest(bulkDocsItr.capture(), failureHandler.capture(), completionHandler.capture()); completionHandler.getValue().accept(exception); assertTrue(failureCalled.get()); + verify(executionService, times(1)).executeBulkRequest(any(), any(), any()); // now check success Iterator<DocWriteRequest> req = bulkDocsItr.getValue().iterator(); @@ -311,6 +309,7 @@ } else { assertSame(remoteNode1, node.getValue()); } + verifyZeroInteractions(executionService); } public void testSingleItemBulkActionIngestForward() throws Exception {

danielmitterdorfer · 2017-01-05T17:16:28Z

@areek, @bleskes I have benchmarked the single index operation with async translog fsync *) of the latest version on Areek's branch ("contender") against the merge base (baseline 8aca504). The overhead is basically not measurable:

Metric	Baseline Value	Contender Value	Unit
Min Throughput	9410	9345	docs/s
Median Throughput	9439	9433	docs/s
Max Throughput	9454	9447	docs/s

I used a custom track (ltaxis) which is a stripped down version of the standard nyc_taxis track ("only" 15 million records). The main reason is that this benchmark is doing so much requests that it would not be feasible to run the full nyc_taxis track.

*) async translog fsync is enabled to avoid accidentally bottlenecking on the fsync operation.

@danielmitterdorfer

…rt of elastic#21964) Performance testing by @danielmitterdorfer revealed single index/delete operations have similar performance (indexing throughput) to equivalent single item bulk request. This PR reduces the code paths to executing single write operations, by reusing the logic in (shard) bulk action for executing single operation as a single-item bulk request. relates to elastic#21964

@danielmitterdorfer

… of #21964) (#22812) * Make index and delete operation execute as a single bulk item (backport of #21964) Performance testing by @danielmitterdorfer revealed single index/delete operations have similar performance (indexing throughput) to equivalent single item bulk request. This PR reduces the code paths to executing single write operations, by reusing the logic in (shard) bulk action for executing single operation as a single-item bulk request. relates to #21964 * remove awaitfix for IndexWithShadowReplica tests

Currently, update action internally uses deprecated index and delete transport actions. As of elastic#21964, these tranport actions were deprecated in favour of using single item bulk request. In this commit, update action uses single item bulk action.

Currently, update action internally uses deprecated index and delete transport actions. As of #21964, these tranport actions were deprecated in favour of using single item bulk request. In this commit, update action uses single item bulk action.

@bleskes

As @bleskes pointed out in elastic#23069 there were inconsistencies in version handling on 5.x and 5.3 from master due to backport of elastic#21964. This change ensures versions are handled uniformly and fixes minor issues in shard bulk action to be similar to master fixes elastic#23069

@bleskes

As @bleskes pointed out in #23069 there were inconsistencies in version handling on 5.x and 5.3 from master due to backport of #21964. This change ensures versions are handled uniformly and fixes minor issues in shard bulk action to be similar to master fixes #23069

@bleskes

As @bleskes pointed out in #23069 there were inconsistencies in version handling on 5.x and 5.3 from master due to backport of #21964. This change ensures versions are handled uniformly and fixes minor issues in shard bulk action to be similar to master fixes #23069

areek added >enhancement resiliency review v5.2.0 v6.0.0-alpha1 labels Dec 5, 2016

areek force-pushed the enhancement/use_shard_bulk_for_single_ops branch 3 times, most recently from 7989e01 to c8fe192 Compare December 7, 2016 05:13

areek force-pushed the enhancement/use_shard_bulk_for_single_ops branch from c8fe192 to c5b09ad Compare December 7, 2016 05:43

martijnvg reviewed Dec 7, 2016

View reviewed changes

clintongormley added the :Distributed Indexing/CRUD A catch all label for issues around indexing, updating and getting a doc by id. Not search. label Dec 7, 2016

areek added 3 commits December 8, 2016 17:17

Add bwc for index/delete requests from pre-6.0 nodes

4231aa4

Merge branch 'master' into enhancement/use_shard_bulk_for_single_ops

f766dfc

Merge branch 'master' into enhancement/use_shard_bulk_for_single_ops

85d9c74

areek added 2 commits December 16, 2016 12:03

Merge branch 'master' into enhancement/use_shard_bulk_for_single_ops

38060b2

Merge branch 'master' into enhancement/use_shard_bulk_for_single_ops

eb5cc1e

bleskes suggested changes Dec 20, 2016

View reviewed changes

areek added 2 commits December 21, 2016 00:26

incorporate feedback

de44584

Merge branch 'master' into enhancement/use_shard_bulk_for_single_ops

180ceef

areek force-pushed the enhancement/use_shard_bulk_for_single_ops branch from 3136ae8 to 180ceef Compare December 21, 2016 06:04

better code sharding?

0eaaee1

bleskes approved these changes Dec 22, 2016

View reviewed changes

martijnvg approved these changes Dec 22, 2016

View reviewed changes

dakrone added 4 commits January 9, 2017 11:44

Merge branch 'master' into enhancement/use_shard_bulk_for_single_ops

bd65ad5

Merge branch 'master' into enhancement/use_shard_bulk_for_single_ops

7da6e0f

Merge branch 'master' into enhancement/use_shard_bulk_for_single_ops

692ddac

Merge branch 'master' into enhancement/use_shard_bulk_for_single_ops

e93fdb8

dakrone merged commit e93fdb8 into elastic:master Jan 13, 2017

bleskes added v5.3.0 and removed v5.2.0 labels Jan 16, 2017

areek mentioned this pull request Jan 26, 2017

Make index and delete operation execute as single bulk item (backport of #21964) #22812

Merged

areek mentioned this pull request Feb 1, 2017

Use bulk action internally for update #22915

Merged

bleskes mentioned this pull request Feb 9, 2017

Delete operation on primary don't do version handling correctly, on 5.3 & 5.x #23069

Closed

areek mentioned this pull request Feb 9, 2017

Fix backport executing ops as single item bulk #23083

Merged

Make index and delete operation execute as a single bulk item #21964

Make index and delete operation execute as a single bulk item #21964

Uh oh!

Conversation

areek commented Dec 5, 2016

Uh oh!

bleskes commented Dec 5, 2016

Uh oh!

areek commented Dec 6, 2016

Uh oh!

areek commented Dec 7, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

martijnvg commented Dec 7, 2016

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!

areek commented Dec 7, 2016

Uh oh!

areek commented Dec 9, 2016

Uh oh!

bleskes 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!

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!

areek commented Dec 21, 2016

Uh oh!

bleskes commented Dec 21, 2016

Uh oh!

bleskes commented Dec 22, 2016

Uh oh!

bleskes left a comment

Choose a reason for hiding this comment

Uh oh!

bleskes commented Dec 22, 2016

areek commented Dec 7, 2016 •

edited

Loading