You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sf...@apache.org on 2015/02/25 17:33:06 UTC
[2/2] incubator-usergrid git commit: Add batch of batches
Add batch of batches
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/ac55e087
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/ac55e087
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/ac55e087
Branch: refs/heads/USERGRID-273-indexbuffer
Commit: ac55e0878bc5a73deb8e07137f5394cb85b409a0
Parents: 420f500
Author: Shawn Feldman <sf...@apache.org>
Authored: Wed Feb 25 09:32:48 2015 -0700
Committer: Shawn Feldman <sf...@apache.org>
Committed: Wed Feb 25 09:32:48 2015 -0700
----------------------------------------------------------------------
.../persistence/index/EntityIndexBatch.java | 3 +-
.../index/RequestBuilderContainer.java | 9 ++++
.../index/impl/EsEntityIndexBatchImpl.java | 19 +++++----
.../index/impl/EsEntityIndexImpl.java | 2 +-
.../index/impl/IndexBatchBufferImpl.java | 44 ++++++++++++--------
.../persistence/index/impl/EntityIndexTest.java | 9 ++--
6 files changed, 55 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ac55e087/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndexBatch.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndexBatch.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndexBatch.java
index f5b8abc..f3f9100 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndexBatch.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndexBatch.java
@@ -20,6 +20,7 @@ package org.apache.usergrid.persistence.index;/*
import java.util.UUID;
+import org.apache.usergrid.persistence.core.future.BetterFuture;
import org.apache.usergrid.persistence.index.query.CandidateResult;
import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.entity.Id;
@@ -64,7 +65,7 @@ public interface EntityIndexBatch {
/**
* Execute the batch
*/
- public void execute();
+ public BetterFuture execute();
/**
* Execute the batch and force the refresh
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ac55e087/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/RequestBuilderContainer.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/RequestBuilderContainer.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/RequestBuilderContainer.java
index bc5e115..e7928ff 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/RequestBuilderContainer.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/RequestBuilderContainer.java
@@ -30,6 +30,8 @@ public class RequestBuilderContainer{
private final ConcurrentLinkedQueue<ShardReplicationOperationRequestBuilder> builders;
private final BetterFuture<Iterator<ShardReplicationOperationRequestBuilder>> containerFuture;
+ private boolean forceRefresh = false;
+
public RequestBuilderContainer(){
final RequestBuilderContainer parent = this;
builders = new ConcurrentLinkedQueue<>();
@@ -53,4 +55,11 @@ public class RequestBuilderContainer{
public BetterFuture<Iterator<ShardReplicationOperationRequestBuilder>> getFuture(){
return containerFuture;
}
+
+ public boolean isForceRefresh() {
+ return forceRefresh;
+ }
+ public void setForceRefresh(boolean forceRefresh) {
+ this.forceRefresh = forceRefresh;
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ac55e087/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexBatchImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexBatchImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexBatchImpl.java
index dc8b1af..4fcdeb3 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexBatchImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexBatchImpl.java
@@ -88,19 +88,17 @@ public class EsEntityIndexBatchImpl implements EntityIndexBatch {
private final IndexIdentifier indexIdentifier;
private final IndexBatchBuffer indexBatchBuffer;
- private final FailureMonitor failureMonitor;
private final AliasedEntityIndex entityIndex;
- private final RequestBuilderContainer container;
+ private RequestBuilderContainer container;
public EsEntityIndexBatchImpl(final ApplicationScope applicationScope, final Client client,final IndexBatchBuffer indexBatchBuffer,
- final IndexFig config, final FailureMonitor failureMonitor, final AliasedEntityIndex entityIndex ) {
+ final IndexFig config, final AliasedEntityIndex entityIndex ) {
this.applicationScope = applicationScope;
this.client = client;
this.indexBatchBuffer = indexBatchBuffer;
- this.failureMonitor = failureMonitor;
this.entityIndex = entityIndex;
this.indexIdentifier = IndexingUtils.createIndexIdentifier(config, applicationScope);
this.alias = indexIdentifier.getAlias();
@@ -214,15 +212,20 @@ public class EsEntityIndexBatchImpl implements EntityIndexBatch {
}
@Override
- public void execute() {
- indexBatchBuffer.put(container);
+ public BetterFuture execute() {
+ RequestBuilderContainer tempContainer = container;
+ container = new RequestBuilderContainer();
+ BetterFuture future = indexBatchBuffer.put(tempContainer);
+ return future;
}
@Override
public void executeAndRefresh() {
- BetterFuture future = indexBatchBuffer.put(container);
+ container.setForceRefresh(true);
+ RequestBuilderContainer tempContainer = container;
+ container = new RequestBuilderContainer();
+ BetterFuture future = indexBatchBuffer.put(tempContainer);
future.get();
- entityIndex.refresh();
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ac55e087/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
index 4e5687f..2881ce4 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
@@ -281,7 +281,7 @@ public class EsEntityIndexImpl implements AliasedEntityIndex {
@Override
public EntityIndexBatch createBatch() {
return new EsEntityIndexBatchImpl(
- applicationScope, esProvider.getClient(),indexBatchBuffer, config, failureMonitor, this );
+ applicationScope, esProvider.getClient(),indexBatchBuffer, config, this );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ac55e087/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexBatchBufferImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexBatchBufferImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexBatchBufferImpl.java
index 364c20a..b5d9528 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexBatchBufferImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexBatchBufferImpl.java
@@ -40,12 +40,15 @@ import rx.Subscriber;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func1;
+import rx.schedulers.Schedulers;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
/**
@@ -55,24 +58,20 @@ import java.util.concurrent.*;
public class IndexBatchBufferImpl implements IndexBatchBuffer {
private static final Logger log = LoggerFactory.getLogger(IndexBatchBufferImpl.class);
- private final MetricsFactory metricsFactory;
private final Counter indexSizeCounter;
private final Client client;
private final FailureMonitorImpl failureMonitor;
private final IndexFig config;
private final Timer flushTimer;
- private final ArrayBlockingQueue<RequestBuilderContainer> blockingQueue;
private final Counter bufferCounter;
private Observable<List<RequestBuilderContainer>> consumer;
private Producer producer;
@Inject
public IndexBatchBufferImpl(final IndexFig config, final EsProvider provider, final MetricsFactory metricsFactory){
- this.metricsFactory = metricsFactory;
this.flushTimer = metricsFactory.getTimer(IndexBatchBuffer.class, "index.buffer.flush");
this.indexSizeCounter = metricsFactory.getCounter(IndexBatchBuffer.class, "index.buffer.size");
this.config = config;
- this.blockingQueue = new ArrayBlockingQueue<>(config.getIndexBatchSize());
this.failureMonitor = new FailureMonitorImpl(config,provider);
this.producer = new Producer();
this.client = provider.getClient();
@@ -81,18 +80,25 @@ public class IndexBatchBufferImpl implements IndexBatchBuffer {
}
private void consumer() {
+ final AtomicLong queueSize = new AtomicLong();
//batch up sets of some size and send them in batch
this.consumer = Observable.create(producer)
+ .subscribeOn(Schedulers.io())
+ .doOnNext(new Action1<RequestBuilderContainer>() {
+ @Override
+ public void call(RequestBuilderContainer requestBuilderContainer) {
+ queueSize.addAndGet(requestBuilderContainer.getBuilder().size());
+ }
+ })
.buffer(config.getIndexBufferTimeout(), TimeUnit.MILLISECONDS, config.getIndexBufferSize())
.doOnNext(new Action1<List<RequestBuilderContainer>>() {
@Override
public void call(List<RequestBuilderContainer> containerList) {
- for (RequestBuilderContainer container : containerList) {
- blockingQueue.add(container);
- }
flushTimer.time();
indexSizeCounter.dec(containerList.size());
- execute(config.isForcedRefresh());
+ if(containerList.size()>0){
+ execute(containerList);
+ }
}
});
consumer.subscribe();
@@ -109,20 +115,22 @@ public class IndexBatchBufferImpl implements IndexBatchBuffer {
/**
* Execute the request, check for errors, then re-init the batch for future use
*/
- private void execute(final boolean refresh) {
+ private void execute(final List<RequestBuilderContainer> containers) {
- if (blockingQueue.size() == 0) {
+ if (containers == null || containers.size() == 0) {
return;
}
-
- Collection<RequestBuilderContainer> containerCollection = new ArrayList<>(config.getIndexBatchSize());
- blockingQueue.drainTo(containerCollection);
+ final AtomicBoolean isForceRefresh = new AtomicBoolean(config.isForcedRefresh());
//clear the queue or proceed to buffersize
- Observable.from(containerCollection)
+ Observable.from(containers)
+ .subscribeOn(Schedulers.io())
.flatMap(new Func1<RequestBuilderContainer, Observable<ShardReplicationOperationRequestBuilder>>() {
@Override
public Observable<ShardReplicationOperationRequestBuilder> call(RequestBuilderContainer requestBuilderContainer) {
+ if (requestBuilderContainer.isForceRefresh()){
+ isForceRefresh.set(true);
+ }
return Observable.from(requestBuilderContainer.getBuilder())
.map(new Func1<ShardReplicationOperationRequestBuilder, ShardReplicationOperationRequestBuilder>() {
@Override
@@ -136,20 +144,20 @@ public class IndexBatchBufferImpl implements IndexBatchBuffer {
.doOnNext(new Action1<List<ShardReplicationOperationRequestBuilder>>() {
@Override
public void call(List<ShardReplicationOperationRequestBuilder> builders) {
- final BulkRequestBuilder bulkRequest = initRequest(refresh);
- for(ShardReplicationOperationRequestBuilder builder : builders) {
+ final BulkRequestBuilder bulkRequest = initRequest(isForceRefresh.get());
+ for (ShardReplicationOperationRequestBuilder builder : builders) {
if (builder instanceof IndexRequestBuilder) {
bulkRequest.add((IndexRequestBuilder) builder);
}
if (builder instanceof DeleteRequestBuilder) {
bulkRequest.add((DeleteRequestBuilder) builder);
}
- sendRequest(bulkRequest);
}
+ sendRequest(bulkRequest);
}
}).toBlocking().lastOrDefault(null);
- for (RequestBuilderContainer container : containerCollection) {
+ for (RequestBuilderContainer container : containers) {
container.done();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/ac55e087/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java b/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java
index 6cda9f2..e3eeaf4 100644
--- a/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java
+++ b/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/index/impl/EntityIndexTest.java
@@ -467,18 +467,21 @@ public class EntityIndexTest extends BaseIT {
EntityIndexBatch batch = ei.createBatch();
- batch.index( appScope, user ).executeAndRefresh();
+ batch.index( appScope, user );
+ batch.executeAndRefresh();
+
Query query = new Query();
query.addEqualityFilter( "username", "edanuff" );
CandidateResults r = ei.search( appScope, SearchTypes.fromTypes( "edanuff" ), query );
assertEquals( user.getId(), r.get( 0 ).getId() );
- batch.deindex(appScope, user.getId(), user.getVersion() ).executeAndRefresh();
-
+ batch.deindex(appScope, user.getId(), user.getVersion() );
+ batch.executeAndRefresh();
// EntityRef
query = new Query();
query.addEqualityFilter( "username", "edanuff" );
+
r = ei.search(appScope,SearchTypes.fromTypes( "edanuff" ), query );
assertFalse( r.iterator().hasNext() );