Skip to content

Commit 4a3978d

Browse files
kimchykimchy
committed
Optimize dynamic mapping updates on master by processing latest one per index/node
Instead of processing all the bulk of update mappings we have per index/node, we can only update the last ordered one out of those (cause they are incremented on the node/index level). This will improve the processing time of an index that have large updates of mappings. closes elastic#4373
1 parent dd95895 commit 4a3978d

File tree

5 files changed

+194
-44
lines changed

5 files changed

+194
-44
lines changed

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

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
import org.elasticsearch.cluster.block.ClusterBlockLevel;
4343
import org.elasticsearch.cluster.metadata.IndexMetaData;
4444
import org.elasticsearch.cluster.metadata.MappingMetaData;
45+
import org.elasticsearch.cluster.node.DiscoveryNode;
4546
import org.elasticsearch.cluster.routing.ShardIterator;
4647
import org.elasticsearch.common.bytes.BytesReference;
4748
import org.elasticsearch.common.collect.Tuple;
@@ -567,11 +568,18 @@ private void updateMappingOnMaster(final String index, final String type) {
567568
if (documentMapper == null) { // should not happen
568569
return;
569570
}
570-
documentMapper.refreshSource();
571-
572571
IndexMetaData metaData = clusterService.state().metaData().index(index);
572+
if (metaData == null) {
573+
return;
574+
}
575+
576+
// we generate the order id before we get the mapping to send and refresh the source, so
577+
// if 2 happen concurrently, we know that the later order will include the previous one
578+
long orderId = mappingUpdatedAction.generateNextMappingUpdateOrder();
579+
documentMapper.refreshSource();
573580

574-
final MappingUpdatedAction.MappingUpdatedRequest request = new MappingUpdatedAction.MappingUpdatedRequest(index, metaData.uuid(), type, documentMapper.mappingSource());
581+
DiscoveryNode node = clusterService.localNode();
582+
final MappingUpdatedAction.MappingUpdatedRequest request = new MappingUpdatedAction.MappingUpdatedRequest(index, metaData.uuid(), type, documentMapper.mappingSource(), orderId, node != null ? node.id() : null);
575583
mappingUpdatedAction.execute(request, new ActionListener<MappingUpdatedAction.MappingUpdatedResponse>() {
576584
@Override
577585
public void onResponse(MappingUpdatedAction.MappingUpdatedResponse mappingUpdatedResponse) {

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

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import org.elasticsearch.cluster.metadata.IndexMetaData;
3737
import org.elasticsearch.cluster.metadata.MappingMetaData;
3838
import org.elasticsearch.cluster.metadata.MetaData;
39+
import org.elasticsearch.cluster.node.DiscoveryNode;
3940
import org.elasticsearch.cluster.routing.ShardIterator;
4041
import org.elasticsearch.common.inject.Inject;
4142
import org.elasticsearch.common.settings.Settings;
@@ -272,9 +273,13 @@ private void updateMappingOnMaster(final IndexRequest request, IndexMetaData ind
272273
if (documentMapper == null) { // should not happen
273274
return;
274275
}
276+
// we generate the order id before we get the mapping to send and refresh the source, so
277+
// if 2 happen concurrently, we know that the later order will include the previous one
278+
long orderId = mappingUpdatedAction.generateNextMappingUpdateOrder();
275279
documentMapper.refreshSource();
280+
DiscoveryNode node = clusterService.localNode();
276281
final MappingUpdatedAction.MappingUpdatedRequest mappingRequest =
277-
new MappingUpdatedAction.MappingUpdatedRequest(request.index(), indexMetaData.uuid(), request.type(), documentMapper.mappingSource());
282+
new MappingUpdatedAction.MappingUpdatedRequest(request.index(), indexMetaData.uuid(), request.type(), documentMapper.mappingSource(), orderId, node != null ? node.id() : null);
278283
logger.trace("Sending mapping updated to master: {}", mappingRequest);
279284
mappingUpdatedAction.execute(mappingRequest, new ActionListener<MappingUpdatedAction.MappingUpdatedResponse>() {
280285
@Override

‎src/main/java/org/elasticsearch/cluster/action/index/MappingUpdatedAction.java

Lines changed: 34 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -41,13 +41,15 @@
4141
import org.elasticsearch.transport.TransportService;
4242

4343
import java.io.IOException;
44+
import java.util.concurrent.atomic.AtomicLong;
4445

4546
/**
4647
* Called by shards in the cluster when their mapping was dynamically updated and it needs to be updated
4748
* in the cluster state meta data (and broadcast to all members).
4849
*/
4950
public class MappingUpdatedAction extends TransportMasterNodeOperationAction<MappingUpdatedAction.MappingUpdatedRequest, MappingUpdatedAction.MappingUpdatedResponse> {
5051

52+
private final AtomicLong mappingUpdateOrderGen = new AtomicLong();
5153
private final MetaDataMappingService metaDataMappingService;
5254

5355
@Inject
@@ -57,6 +59,10 @@ public MappingUpdatedAction(Settings settings, TransportService transportService
5759
this.metaDataMappingService = metaDataMappingService;
5860
}
5961

62+
public long generateNextMappingUpdateOrder() {
63+
return mappingUpdateOrderGen.incrementAndGet();
64+
}
65+
6066
@Override
6167
protected String transportAction() {
6268
return "cluster/mappingUpdated";
@@ -80,7 +86,7 @@ protected MappingUpdatedResponse newResponse() {
8086

8187
@Override
8288
protected void masterOperation(final MappingUpdatedRequest request, final ClusterState state, final ActionListener<MappingUpdatedResponse> listener) throws ElasticSearchException {
83-
metaDataMappingService.updateMapping(request.index(), request.indexUUID(), request.type(), request.mappingSource(), new ClusterStateUpdateListener() {
89+
metaDataMappingService.updateMapping(request.index(), request.indexUUID(), request.type(), request.mappingSource(), request.order, request.nodeId, new ClusterStateUpdateListener() {
8490
@Override
8591
public void onResponse(ClusterStateUpdateResponse response) {
8692
listener.onResponse(new MappingUpdatedResponse());
@@ -112,15 +118,19 @@ public static class MappingUpdatedRequest extends MasterNodeOperationRequest<Map
112118
private String indexUUID = IndexMetaData.INDEX_UUID_NA_VALUE;
113119
private String type;
114120
private CompressedString mappingSource;
121+
private long order = -1; // -1 means not set...
122+
private String nodeId = null; // null means not set
115123

116124
MappingUpdatedRequest() {
117125
}
118126

119-
public MappingUpdatedRequest(String index, String indexUUID, String type, CompressedString mappingSource) {
127+
public MappingUpdatedRequest(String index, String indexUUID, String type, CompressedString mappingSource, long order, String nodeId) {
120128
this.index = index;
121129
this.indexUUID = indexUUID;
122130
this.type = type;
123131
this.mappingSource = mappingSource;
132+
this.order = order;
133+
this.nodeId = nodeId;
124134
}
125135

126136
public String index() {
@@ -139,6 +149,20 @@ public CompressedString mappingSource() {
139149
return mappingSource;
140150
}
141151

152+
/**
153+
* Returns -1 if not set...
154+
*/
155+
public long order() {
156+
return this.order;
157+
}
158+
159+
/**
160+
* Returns null for not set.
161+
*/
162+
public String nodeId() {
163+
return this.nodeId;
164+
}
165+
142166
@Override
143167
public ActionRequestValidationException validate() {
144168
return null;
@@ -153,6 +177,10 @@ public void readFrom(StreamInput in) throws IOException {
153177
if (in.getVersion().onOrAfter(Version.V_0_90_6)) {
154178
indexUUID = in.readString();
155179
}
180+
if (in.getVersion().after(Version.V_0_90_7)) {
181+
order = in.readLong();
182+
nodeId = in.readOptionalString();
183+
}
156184
}
157185

158186
@Override
@@ -164,6 +192,10 @@ public void writeTo(StreamOutput out) throws IOException {
164192
if (out.getVersion().onOrAfter(Version.V_0_90_6)) {
165193
out.writeString(indexUUID);
166194
}
195+
if (out.getVersion().after(Version.V_0_90_7)) {
196+
out.writeLong(order);
197+
out.writeOptionalString(nodeId);
198+
}
167199
}
168200

169201
@Override

‎src/main/java/org/elasticsearch/cluster/metadata/MetaDataMappingService.java

Lines changed: 105 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,6 @@
3838
import org.elasticsearch.common.inject.Inject;
3939
import org.elasticsearch.common.settings.Settings;
4040
import org.elasticsearch.common.unit.TimeValue;
41-
import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
4241
import org.elasticsearch.index.Index;
4342
import org.elasticsearch.index.mapper.DocumentMapper;
4443
import org.elasticsearch.index.mapper.MapperService;
@@ -49,9 +48,9 @@
4948
import org.elasticsearch.indices.InvalidTypeNameException;
5049
import org.elasticsearch.indices.TypeMissingException;
5150
import org.elasticsearch.percolator.PercolatorService;
51+
import org.elasticsearch.threadpool.ThreadPool;
5252

5353
import java.util.*;
54-
import java.util.concurrent.BlockingQueue;
5554

5655
import static com.google.common.collect.Maps.newHashMap;
5756
import static org.elasticsearch.index.mapper.DocumentMapper.MergeFlags.mergeFlags;
@@ -61,15 +60,20 @@
6160
*/
6261
public class MetaDataMappingService extends AbstractComponent {
6362

63+
private final ThreadPool threadPool;
6464
private final ClusterService clusterService;
65-
6665
private final IndicesService indicesService;
6766

68-
private final BlockingQueue<MappingTask> refreshOrUpdateQueue = ConcurrentCollections.newBlockingQueue();
67+
// the mutex protect all the refreshOrUpdate variables!
68+
private final Object refreshOrUpdateMutex = new Object();
69+
private final List<MappingTask> refreshOrUpdateQueue = new ArrayList<MappingTask>();
70+
private long refreshOrUpdateInsertOrder;
71+
private long refreshOrUpdateProcessedInsertOrder;
6972

7073
@Inject
71-
public MetaDataMappingService(Settings settings, ClusterService clusterService, IndicesService indicesService) {
74+
public MetaDataMappingService(Settings settings, ThreadPool threadPool, ClusterService clusterService, IndicesService indicesService) {
7275
super(settings);
76+
this.threadPool = threadPool;
7377
this.clusterService = clusterService;
7478
this.indicesService = indicesService;
7579
}
@@ -96,12 +100,16 @@ static class RefreshTask extends MappingTask {
96100
static class UpdateTask extends MappingTask {
97101
final String type;
98102
final CompressedString mappingSource;
103+
final long order; // -1 for unknown
104+
final String nodeId; // null fr unknown
99105
final ClusterStateUpdateListener listener;
100106

101-
UpdateTask(String index, String indexUUID, String type, CompressedString mappingSource, ClusterStateUpdateListener listener) {
107+
UpdateTask(String index, String indexUUID, String type, CompressedString mappingSource, long order, String nodeId, ClusterStateUpdateListener listener) {
102108
super(index, indexUUID);
103109
this.type = type;
104110
this.mappingSource = mappingSource;
111+
this.order = order;
112+
this.nodeId = nodeId;
105113
this.listener = listener;
106114
}
107115
}
@@ -111,9 +119,26 @@ static class UpdateTask extends MappingTask {
111119
* as possible so we won't create the same index all the time for example for the updates on the same mapping
112120
* and generate a single cluster change event out of all of those.
113121
*/
114-
ClusterState executeRefreshOrUpdate(final ClusterState currentState) throws Exception {
115-
List<MappingTask> allTasks = new ArrayList<MappingTask>();
116-
refreshOrUpdateQueue.drainTo(allTasks);
122+
ClusterState executeRefreshOrUpdate(final ClusterState currentState, final long insertionOrder) throws Exception {
123+
final List<MappingTask> allTasks = new ArrayList<MappingTask>();
124+
125+
synchronized (refreshOrUpdateMutex) {
126+
if (refreshOrUpdateQueue.isEmpty()) {
127+
return currentState;
128+
}
129+
130+
// we already processed this task in a bulk manner in a previous cluster event, simply ignore
131+
// it so we will let other tasks get in and processed ones, we will handle the queued ones
132+
// later on in a subsequent cluster state event
133+
if (insertionOrder < refreshOrUpdateProcessedInsertOrder) {
134+
return currentState;
135+
}
136+
137+
allTasks.addAll(refreshOrUpdateQueue);
138+
refreshOrUpdateQueue.clear();
139+
140+
refreshOrUpdateProcessedInsertOrder = refreshOrUpdateInsertOrder;
141+
}
117142

118143
if (allTasks.isEmpty()) {
119144
return currentState;
@@ -132,32 +157,61 @@ ClusterState executeRefreshOrUpdate(final ClusterState currentState) throws Exce
132157
tasksPerIndex.put(task.index, indexTasks);
133158
}
134159
indexTasks.add(task);
135-
136160
}
137161

138162
boolean dirty = false;
139163
MetaData.Builder mdBuilder = MetaData.builder(currentState.metaData());
140164
for (Map.Entry<String, List<MappingTask>> entry : tasksPerIndex.entrySet()) {
141165
String index = entry.getKey();
142-
List<MappingTask> tasks = entry.getValue();
166+
final IndexMetaData indexMetaData = mdBuilder.get(index);
167+
if (indexMetaData == null) {
168+
// index got deleted on us, ignore...
169+
logger.debug("[{}] ignoring tasks - index meta data doesn't exist", index);
170+
continue;
171+
}
172+
// the tasks lists to iterate over, filled with the list of mapping tasks, trying to keep
173+
// the latest (based on order) update mapping one per node
174+
List<MappingTask> allIndexTasks = entry.getValue();
175+
List<MappingTask> tasks = new ArrayList<MappingTask>();
176+
for (MappingTask task : allIndexTasks) {
177+
if (!indexMetaData.isSameUUID(task.indexUUID)) {
178+
logger.debug("[{}] ignoring task [{}] - index meta data doesn't match task uuid", index, task);
179+
continue;
180+
}
181+
boolean add = true;
182+
// if its an update task, make sure we only process the latest ordered one per node
183+
if (task instanceof UpdateTask) {
184+
UpdateTask uTask = (UpdateTask) task;
185+
// we can only do something to compare if we have the order && node
186+
if (uTask.order != -1 && uTask.nodeId != null) {
187+
for (int i = 0; i < tasks.size(); i++) {
188+
MappingTask existing = tasks.get(i);
189+
if (existing instanceof UpdateTask) {
190+
UpdateTask eTask = (UpdateTask) existing;
191+
// if we have the order, and the node id, then we can compare, and replace if applicable
192+
if (eTask.order != -1 && eTask.nodeId != null) {
193+
if (eTask.nodeId.equals(uTask.nodeId) && uTask.order > eTask.order) {
194+
// a newer update task, we can replace so we execute it one!
195+
tasks.set(i, uTask);
196+
add = false;
197+
break;
198+
}
199+
}
200+
}
201+
}
202+
}
203+
}
204+
205+
if (add) {
206+
tasks.add(task);
207+
}
208+
}
209+
143210
boolean removeIndex = false;
144211
// keep track of what we already refreshed, no need to refresh it again...
145212
Set<String> processedRefreshes = Sets.newHashSet();
146213
try {
147214
for (MappingTask task : tasks) {
148-
final IndexMetaData indexMetaData = mdBuilder.get(index);
149-
if (indexMetaData == null) {
150-
// index got deleted on us, ignore...
151-
logger.debug("[{}] ignoring task [{}] - index meta data doesn't exist", index, task);
152-
continue;
153-
}
154-
155-
if (!indexMetaData.isSameUUID(task.indexUUID)) {
156-
// index got deleted on us, ignore...
157-
logger.debug("[{}] ignoring task [{}] - index meta data doesn't match task uuid", index, task);
158-
continue;
159-
}
160-
161215
if (task instanceof RefreshTask) {
162216
RefreshTask refreshTask = (RefreshTask) task;
163217
try {
@@ -249,13 +303,24 @@ ClusterState executeRefreshOrUpdate(final ClusterState currentState) throws Exce
249303
if (removeIndex) {
250304
indicesService.removeIndex(index, "created for mapping processing");
251305
}
252-
for (Object task : tasks) {
306+
}
307+
}
308+
309+
// fork sending back updates, so we won't wait to send them back on the cluster state, there
310+
// might be a few of those...
311+
threadPool.generic().execute(new Runnable() {
312+
@Override
313+
public void run() {
314+
for (Object task : allTasks) {
253315
if (task instanceof UpdateTask) {
254-
((UpdateTask) task).listener.onResponse(new ClusterStateUpdateResponse(true));
316+
UpdateTask uTask = (UpdateTask) task;
317+
ClusterStateUpdateResponse response = new ClusterStateUpdateResponse(true);
318+
uTask.listener.onResponse(response);
255319
}
256320
}
257321
}
258-
}
322+
});
323+
259324

260325
if (!dirty) {
261326
return currentState;
@@ -267,7 +332,11 @@ ClusterState executeRefreshOrUpdate(final ClusterState currentState) throws Exce
267332
* Refreshes mappings if they are not the same between original and parsed version
268333
*/
269334
public void refreshMapping(final String index, final String indexUUID, final String... types) {
270-
refreshOrUpdateQueue.add(new RefreshTask(index, indexUUID, types));
335+
final long insertOrder;
336+
synchronized (refreshOrUpdateMutex) {
337+
insertOrder = ++refreshOrUpdateInsertOrder;
338+
refreshOrUpdateQueue.add(new RefreshTask(index, indexUUID, types));
339+
}
271340
clusterService.submitStateUpdateTask("refresh-mapping [" + index + "][" + Arrays.toString(types) + "]", Priority.HIGH, new ClusterStateUpdateTask() {
272341
@Override
273342
public void onFailure(String source, Throwable t) {
@@ -276,13 +345,17 @@ public void onFailure(String source, Throwable t) {
276345

277346
@Override
278347
public ClusterState execute(ClusterState currentState) throws Exception {
279-
return executeRefreshOrUpdate(currentState);
348+
return executeRefreshOrUpdate(currentState, insertOrder);
280349
}
281350
});
282351
}
283352

284-
public void updateMapping(final String index, final String indexUUID, final String type, final CompressedString mappingSource, final ClusterStateUpdateListener listener) {
285-
refreshOrUpdateQueue.add(new UpdateTask(index, indexUUID, type, mappingSource, listener));
353+
public void updateMapping(final String index, final String indexUUID, final String type, final CompressedString mappingSource, final long order, final String nodeId, final ClusterStateUpdateListener listener) {
354+
final long insertOrder;
355+
synchronized (refreshOrUpdateMutex) {
356+
insertOrder = ++refreshOrUpdateInsertOrder;
357+
refreshOrUpdateQueue.add(new UpdateTask(index, indexUUID, type, mappingSource, order, nodeId, listener));
358+
}
286359
clusterService.submitStateUpdateTask("update-mapping [" + index + "][" + type + "]", Priority.HIGH, new ClusterStateUpdateTask() {
287360
@Override
288361
public void onFailure(String source, Throwable t) {
@@ -291,7 +364,7 @@ public void onFailure(String source, Throwable t) {
291364

292365
@Override
293366
public ClusterState execute(final ClusterState currentState) throws Exception {
294-
return executeRefreshOrUpdate(currentState);
367+
return executeRefreshOrUpdate(currentState, insertOrder);
295368
}
296369
});
297370
}

0 commit comments

Comments
 (0)