honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use atomic collections to make sure all of the memory contents are visible from writing to reading thread.

This commit is contained in:
Martijn van Groningen 2013-08-30 00:06:45 +02:00
parent 7113731022
commit a7b2b7847a
1 changed files with 33 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
src/main/java/org/elasticsearch/action/percolate/TransportMultiPercolateAction.java
View File

@ -33,6 +33,7 @@ import org.elasticsearch.cluster.routing.ShardIterator;
import org.elasticsearch.common.inject.Inject; import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.concurrent.AtomicArray; import org.elasticsearch.common.util.concurrent.AtomicArray;
import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
import org.elasticsearch.index.engine.DocumentMissingException; import org.elasticsearch.index.engine.DocumentMissingException;
import org.elasticsearch.index.shard.ShardId; import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.percolator.PercolatorService; import org.elasticsearch.percolator.PercolatorService;
@ -42,7 +43,9 @@ import org.elasticsearch.transport.TransportChannel;
import org.elasticsearch.transport.TransportService; import org.elasticsearch.transport.TransportService;
import java.util.*; import java.util.*;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerArray;
import java.util.concurrent.atomic.AtomicReferenceArray; import java.util.concurrent.atomic.AtomicReferenceArray;
/** /**
@ -72,20 +75,24 @@ public class TransportMultiPercolateAction extends TransportAction<MultiPercolat
protected void doExecute(final MultiPercolateRequest request, final ActionListener<MultiPercolateResponse> listener) { protected void doExecute(final MultiPercolateRequest request, final ActionListener<MultiPercolateResponse> listener) {
final ClusterState clusterState = clusterService.state(); final ClusterState clusterState = clusterService.state();
clusterState.blocks().globalBlockedRaiseException(ClusterBlockLevel.READ); clusterState.blocks().globalBlockedRaiseException(ClusterBlockLevel.READ);
@SuppressWarnings("unchecked")
final List<Object> percolateRequests = (List) request.requests();
final TIntArrayList slots = new TIntArrayList(); final AtomicReferenceArray<Object> percolateRequests = new AtomicReferenceArray<Object>(request.requests().size());
final List<GetRequest> existingDocsRequests = new ArrayList<GetRequest>(); TIntArrayList getRequestSlots = new TIntArrayList();
List<GetRequest> existingDocsRequests = new ArrayList<GetRequest>();
for (int i = 0; i < request.requests().size(); i++) { for (int i = 0; i < request.requests().size(); i++) {
PercolateRequest percolateRequest = request.requests().get(i); PercolateRequest percolateRequest = request.requests().get(i);
percolateRequest.startTime = System.currentTimeMillis(); percolateRequest.startTime = System.currentTimeMillis();
percolateRequests.set(i, percolateRequest);
if (percolateRequest.getRequest() != null) { if (percolateRequest.getRequest() != null) {
existingDocsRequests.add(percolateRequest.getRequest()); existingDocsRequests.add(percolateRequest.getRequest());
slots.add(i); getRequestSlots.add(i);
} }
} }
// Can have a mixture of percolate requests. (normal percolate requests & percolate existing doc),
// so we need to keep track for what percolate request we had a get request
final AtomicIntegerArray percolateRequestSlotsWithGet = new AtomicIntegerArray(getRequestSlots.toArray());
if (!existingDocsRequests.isEmpty()) { if (!existingDocsRequests.isEmpty()) {
final MultiGetRequest multiGetRequest = new MultiGetRequest(); final MultiGetRequest multiGetRequest = new MultiGetRequest();
for (GetRequest getRequest : existingDocsRequests) { for (GetRequest getRequest : existingDocsRequests) {
@ -101,7 +108,7 @@ public class TransportMultiPercolateAction extends TransportAction<MultiPercolat
public void onResponse(MultiGetResponse multiGetItemResponses) { public void onResponse(MultiGetResponse multiGetItemResponses) {
for (int i = 0; i < multiGetItemResponses.getResponses().length; i++) { for (int i = 0; i < multiGetItemResponses.getResponses().length; i++) {
MultiGetItemResponse itemResponse = multiGetItemResponses.getResponses()[i]; MultiGetItemResponse itemResponse = multiGetItemResponses.getResponses()[i];
int slot = slots.get(i); int slot = percolateRequestSlotsWithGet.get(i);
if (!itemResponse.isFailed()) { if (!itemResponse.isFailed()) {
GetResponse getResponse = itemResponse.getResponse(); GetResponse getResponse = itemResponse.getResponse();
if (getResponse.isExists()) { if (getResponse.isExists()) {
@ -127,18 +134,21 @@ public class TransportMultiPercolateAction extends TransportAction<MultiPercolat
} }
private void multiPercolate(MultiPercolateRequest multiPercolateRequest, final List<Object> percolateRequests, private void multiPercolate(MultiPercolateRequest multiPercolateRequest, final AtomicReferenceArray<Object> percolateRequests,
final ActionListener<MultiPercolateResponse> listener, ClusterState clusterState) { final ActionListener<MultiPercolateResponse> listener, ClusterState clusterState) {
final AtomicInteger[] expectedOperationsPerItem = new AtomicInteger[percolateRequests.size()]; final AtomicInteger[] expectedOperationsPerItem = new AtomicInteger[percolateRequests.length()];
final AtomicReferenceArray<AtomicReferenceArray> responsesByItemAndShard = new AtomicReferenceArray<AtomicReferenceArray>(multiPercolateRequest.requests().size()); final AtomicReferenceArray<AtomicReferenceArray> responsesByItemAndShard = new AtomicReferenceArray<AtomicReferenceArray>(multiPercolateRequest.requests().size());
final AtomicArray<Object> reducedResponses = new AtomicArray<Object>(percolateRequests.size()); final AtomicArray<Object> reducedResponses = new AtomicArray<Object>(percolateRequests.length());
// Keep track what slots belong to what shard, in case a request to a shard fails on all copies
final ConcurrentMap<ShardId, AtomicIntegerArray> shardToSlots = ConcurrentCollections.newConcurrentMap();
// Resolving concrete indices and routing and grouping the requests by shard // Resolving concrete indices and routing and grouping the requests by shard
final Map<ShardId, TransportShardMultiPercolateAction.Request> requestsByShard = new HashMap<ShardId, TransportShardMultiPercolateAction.Request>(); Map<ShardId, TransportShardMultiPercolateAction.Request> requestsByShard = new HashMap<ShardId, TransportShardMultiPercolateAction.Request>();
final Map<ShardId, TIntArrayList> shardToSlots = new HashMap<ShardId, TIntArrayList>(); Map<ShardId, TIntArrayList> shardToSlotsBuilder = new HashMap<ShardId, TIntArrayList>();
int expectedResults = 0; int expectedResults = 0;
for (int i = 0; i < percolateRequests.size(); i++) { for (int i = 0; i < percolateRequests.length(); i++) {
Object element = percolateRequests.get(i); Object element = percolateRequests.get(i);
assert element != null; assert element != null;
if (element instanceof PercolateRequest) { if (element instanceof PercolateRequest) {
@ -160,9 +170,9 @@ public class TransportMultiPercolateAction extends TransportAction<MultiPercolat
} }
requests.add(new TransportShardMultiPercolateAction.Request.Item(i, new PercolateShardRequest(shardId, percolateRequest))); requests.add(new TransportShardMultiPercolateAction.Request.Item(i, new PercolateShardRequest(shardId, percolateRequest)));
TIntArrayList items = shardToSlots.get(shardId); TIntArrayList items = shardToSlotsBuilder.get(shardId);
if (items == null) { if (items == null) {
shardToSlots.put(shardId, items = new TIntArrayList()); shardToSlotsBuilder.put(shardId, items = new TIntArrayList());
} }
items.add(i); items.add(i);
} }
@ -179,6 +189,10 @@ public class TransportMultiPercolateAction extends TransportAction<MultiPercolat
return; return;
} }
for (Map.Entry<ShardId, TIntArrayList> entry : shardToSlotsBuilder.entrySet()) {
shardToSlots.put(entry.getKey(), new AtomicIntegerArray(entry.getValue().toArray()));
}
final AtomicInteger expectedOperations = new AtomicInteger(expectedResults); final AtomicInteger expectedOperations = new AtomicInteger(expectedResults);
for (Map.Entry<ShardId, TransportShardMultiPercolateAction.Request> entry : requestsByShard.entrySet()) { for (Map.Entry<ShardId, TransportShardMultiPercolateAction.Request> entry : requestsByShard.entrySet()) {
final ShardId shardId = entry.getKey(); final ShardId shardId = entry.getKey();
@ -219,8 +233,8 @@ public class TransportMultiPercolateAction extends TransportAction<MultiPercolat
public void onFailure(Throwable e) { public void onFailure(Throwable e) {
logger.debug("Shard multi percolate failure", e); logger.debug("Shard multi percolate failure", e);
try { try {
TIntArrayList slots = shardToSlots.get(shardId); AtomicIntegerArray slots = shardToSlots.get(shardId);
for (int i = 0; i < slots.size(); i++) { for (int i = 0; i < slots.length(); i++) {
int slot = slots.get(i); int slot = slots.get(i);
AtomicReferenceArray shardResults = responsesByItemAndShard.get(slot); AtomicReferenceArray shardResults = responsesByItemAndShard.get(slot);
if (shardResults == null) { if (shardResults == null) {
@ -244,7 +258,7 @@ public class TransportMultiPercolateAction extends TransportAction<MultiPercolat
} }
private void reduce(int slot, private void reduce(int slot,
List<Object> percolateRequests, AtomicReferenceArray<Object> percolateRequests,
AtomicInteger expectedOperations, AtomicInteger expectedOperations,
AtomicArray<Object> reducedResponses, AtomicArray<Object> reducedResponses,
ActionListener<MultiPercolateResponse> listener, ActionListener<MultiPercolateResponse> listener,
@ -253,7 +267,7 @@ public class TransportMultiPercolateAction extends TransportAction<MultiPercolat
AtomicReferenceArray shardResponses = responsesByItemAndShard.get(slot); AtomicReferenceArray shardResponses = responsesByItemAndShard.get(slot);
PercolateResponse reducedResponse = TransportPercolateAction.reduce((PercolateRequest) percolateRequests.get(slot), shardResponses, percolatorService); PercolateResponse reducedResponse = TransportPercolateAction.reduce((PercolateRequest) percolateRequests.get(slot), shardResponses, percolatorService);
reducedResponses.set(slot, reducedResponse); reducedResponses.set(slot, reducedResponse);
assert expectedOperations.get() >= 1; assert expectedOperations.get() >= 1 : "slot[" + slot + "] expected options should be >= 1 but is " + expectedOperations.get();
if (expectedOperations.decrementAndGet() == 0) { if (expectedOperations.decrementAndGet() == 0) {
finish(reducedResponses, listener); finish(reducedResponses, listener);
} }
@ -263,7 +277,7 @@ public class TransportMultiPercolateAction extends TransportAction<MultiPercolat
MultiPercolateResponse.Item[] finalResponse = new MultiPercolateResponse.Item[reducedResponses.length()]; MultiPercolateResponse.Item[] finalResponse = new MultiPercolateResponse.Item[reducedResponses.length()];
for (int i = 0; i < reducedResponses.length(); i++) { for (int i = 0; i < reducedResponses.length(); i++) {
Object element = reducedResponses.get(i); Object element = reducedResponses.get(i);
assert element != null; assert element != null : "Element[" + i + "] shouldn't be null";
if (element instanceof PercolateResponse) { if (element instanceof PercolateResponse) {
finalResponse[i] = new MultiPercolateResponse.Item((PercolateResponse) element); finalResponse[i] = new MultiPercolateResponse.Item((PercolateResponse) element);
} else if (element instanceof Throwable) { } else if (element instanceof Throwable) {