usergrid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From g...@apache.org
Subject [18/37] incubator-usergrid git commit: add batches of batches
Date Tue, 03 Mar 2015 20:00:57 GMT
add batches 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/420f5009
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/420f5009
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/420f5009

Branch: refs/heads/USERGRID-422
Commit: 420f50093e90c2d764d7415a48b1bfb204a1e1f4
Parents: 3231534
Author: Shawn Feldman <sfeldman@apache.org>
Authored: Tue Feb 24 18:57:15 2015 -0700
Committer: Shawn Feldman <sfeldman@apache.org>
Committed: Tue Feb 24 18:57:15 2015 -0700

----------------------------------------------------------------------
 .../persistence/index/IndexBatchBuffer.java     | 11 +--
 .../index/RequestBuilderContainer.java          | 56 ++++++++++++
 .../index/impl/EsEntityIndexBatchImpl.java      | 40 ++-------
 .../index/impl/IndexBatchBufferImpl.java        | 90 ++++++++------------
 4 files changed, 100 insertions(+), 97 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/420f5009/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexBatchBuffer.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexBatchBuffer.java
b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexBatchBuffer.java
index b24d37b..1d71bcd 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexBatchBuffer.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexBatchBuffer.java
@@ -34,16 +34,11 @@ public interface IndexBatchBuffer {
     /**
      * put request into buffer, retu
      *
-     * @param builder
+     * @param container
      */
-    public BetterFuture put(IndexRequestBuilder builder);
+    public BetterFuture put(RequestBuilderContainer container);
+
 
-    /**
-     * put request into buffer
-     *
-     * @param builder
-     */
-    public BetterFuture put(DeleteRequestBuilder builder);
 
 
 }

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/420f5009/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
new file mode 100644
index 0000000..bc5e115
--- /dev/null
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/RequestBuilderContainer.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.persistence.index;
+
+import org.apache.usergrid.persistence.core.future.BetterFuture;
+import org.elasticsearch.action.support.replication.ShardReplicationOperationRequestBuilder;
+
+import java.util.Iterator;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * Classy class class.
+ */
+public  class RequestBuilderContainer{
+    private final ConcurrentLinkedQueue<ShardReplicationOperationRequestBuilder> builders;
+    private final BetterFuture<Iterator<ShardReplicationOperationRequestBuilder>>
containerFuture;
+
+    public RequestBuilderContainer(){
+        final RequestBuilderContainer parent = this;
+        builders = new ConcurrentLinkedQueue<>();
+        this.containerFuture = new BetterFuture<>(new Callable<Iterator<ShardReplicationOperationRequestBuilder>>()
{
+            @Override
+            public Iterator<ShardReplicationOperationRequestBuilder> call() throws
Exception {
+                return parent.getBuilder().iterator();
+            }
+        });
+    }
+
+    public void add(ShardReplicationOperationRequestBuilder builder){
+        builders.add(builder);
+    }
+    public ConcurrentLinkedQueue<ShardReplicationOperationRequestBuilder> getBuilder(){
+        return builders;
+    }
+    public void done(){
+        containerFuture.done();
+    }
+    public BetterFuture<Iterator<ShardReplicationOperationRequestBuilder>> getFuture(){
+        return containerFuture;
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/420f5009/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 e008707..dc8b1af 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
@@ -91,7 +91,7 @@ public class EsEntityIndexBatchImpl implements EntityIndexBatch {
     private final FailureMonitor failureMonitor;
 
     private final AliasedEntityIndex entityIndex;
-    private final ConcurrentLinkedQueue<BetterFuture> promises;
+    private final RequestBuilderContainer container;
 
 
     public EsEntityIndexBatchImpl(final ApplicationScope applicationScope, final Client client,final
IndexBatchBuffer indexBatchBuffer,
@@ -106,7 +106,7 @@ public class EsEntityIndexBatchImpl implements EntityIndexBatch {
         this.alias = indexIdentifier.getAlias();
         this.refresh = config.isForcedRefresh();
         //constrained
-        this.promises = new ConcurrentLinkedQueue<>();
+        this.container = new RequestBuilderContainer();
     }
 
 
@@ -140,7 +140,7 @@ public class EsEntityIndexBatchImpl implements EntityIndexBatch {
         final String entityType = entity.getId().getType();
         IndexRequestBuilder builder =
                 client.prepareIndex(alias.getWriteAlias(), entityType, indexId).setSource(
entityAsMap );
-        promises.add(indexBatchBuffer.put(builder));
+        container.add(builder);
         return this;
     }
 
@@ -186,7 +186,7 @@ public class EsEntityIndexBatchImpl implements EntityIndexBatch {
                    public Object call(String index) {
                        try {
                            DeleteRequestBuilder builder = client.prepareDelete(index, entityType,
indexId).setRefresh(refresh);
-                           promises.add(indexBatchBuffer.put(builder));
+                           container.add(builder);
                        }catch (Exception e){
                            log.error("failed to deindex",e);
                            throw e;
@@ -197,14 +197,12 @@ public class EsEntityIndexBatchImpl implements EntityIndexBatch {
 
         log.debug("Deindexed Entity with index id " + indexId);
 
-
         return this;
     }
 
 
     @Override
     public EntityIndexBatch deindex( final IndexScope indexScope, final Entity entity ) {
-
         return deindex( indexScope, entity.getId(), entity.getVersion() );
     }
 
@@ -217,40 +215,16 @@ public class EsEntityIndexBatchImpl implements EntityIndexBatch {
 
     @Override
     public void execute() {
-       flushFutures();
+        indexBatchBuffer.put(container);
     }
 
     @Override
     public void executeAndRefresh() {
-        flushFutures();
+        BetterFuture future = indexBatchBuffer.put(container);
+        future.get();
         entityIndex.refresh();
     }
 
-    private void flushFutures() {
-        ObservableIterator<BetterFuture> iterator = new ObservableIterator<BetterFuture>("futures")
{
-            @Override
-            protected Iterator<BetterFuture> getIterator() {
-                return promises.iterator();
-            }
-        };
-        Observable.create(iterator)
-                .doOnNext(new Action1<BetterFuture>() {
-                    @Override
-                    public void call(BetterFuture betterFuture) {
-                        betterFuture.get();
-                    }
-                })
-                .buffer(100)
-                .doOnNext(new Action1<List<BetterFuture>>() {
-                    @Override
-                    public void call(List<BetterFuture> betterFutures) {
-                        promises.removeAll(betterFutures);
-                    }
-                })
-                .toBlocking()
-                .lastOrDefault(null);
-    }
-
     /**
      * Set the entity as a map with the context
      *

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/420f5009/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 ad14920..364c20a 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
@@ -24,6 +24,7 @@ import org.apache.usergrid.persistence.core.future.BetterFuture;
 import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
 import org.apache.usergrid.persistence.index.IndexBatchBuffer;
 import org.apache.usergrid.persistence.index.IndexFig;
+import org.apache.usergrid.persistence.index.RequestBuilderContainer;
 import org.elasticsearch.action.WriteConsistencyLevel;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
@@ -38,9 +39,11 @@ import rx.Observable;
 import rx.Subscriber;
 import rx.functions.Action0;
 import rx.functions.Action1;
+import rx.functions.Func1;
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.*;
 
@@ -96,81 +99,56 @@ public class IndexBatchBufferImpl implements IndexBatchBuffer {
     }
 
     @Override
-    public BetterFuture<ShardReplicationOperationRequestBuilder> put(IndexRequestBuilder
builder){
-        RequestBuilderContainer container = new RequestBuilderContainer(builder);
+    public BetterFuture put(RequestBuilderContainer container){
         bufferCounter.inc();
         producer.put(container);
         return container.getFuture();
     }
 
-    @Override
-    public BetterFuture<ShardReplicationOperationRequestBuilder> put(DeleteRequestBuilder
builder){
-        RequestBuilderContainer container = new RequestBuilderContainer(builder);
-        bufferCounter.inc();
-        producer.put(container);
-        return container.getFuture();
-    }
-
-
 
-    private static class RequestBuilderContainer{
-        private final ShardReplicationOperationRequestBuilder builder;
-        private final BetterFuture<ShardReplicationOperationRequestBuilder> containerFuture;
-
-        public RequestBuilderContainer(ShardReplicationOperationRequestBuilder builder){
-            final RequestBuilderContainer parent = this;
-            this.builder = builder;
-            this.containerFuture = new BetterFuture<>(new Callable<ShardReplicationOperationRequestBuilder>()
{
-                @Override
-                public ShardReplicationOperationRequestBuilder call() throws Exception {
-                    return parent.getBuilder();
-                }
-            });
-        }
-
-        public ShardReplicationOperationRequestBuilder getBuilder(){
-            return builder;
-        }
-        private void done(){
-            containerFuture.done();
-        }
-        public BetterFuture<ShardReplicationOperationRequestBuilder> getFuture(){
-            return containerFuture;
-        }
-    }
     /**
      * Execute the request, check for errors, then re-init the batch for future use
      */
-    private void execute( boolean refresh) {
+    private void execute(final boolean refresh) {
 
         if (blockingQueue.size() == 0) {
             return;
         }
 
-        BulkRequestBuilder bulkRequest = initRequest(refresh);
 
         Collection<RequestBuilderContainer> containerCollection = new ArrayList<>(config.getIndexBatchSize());
         blockingQueue.drainTo(containerCollection);
-        int count = 0;
         //clear the queue or proceed to buffersize
-        for (RequestBuilderContainer container : containerCollection) {
-
-            ShardReplicationOperationRequestBuilder builder = container.getBuilder();
-            //only handle two types of requests for now, annoyingly there is no base class
implementation on BulkRequest
-            if (builder instanceof IndexRequestBuilder) {
-                bulkRequest.add((IndexRequestBuilder) builder);
-            }
-            if (builder instanceof DeleteRequestBuilder) {
-                bulkRequest.add((DeleteRequestBuilder) builder);
-            }
-
-            if (count++ == config.getIndexBatchSize()) {
-                sendRequest(bulkRequest);
-                bulkRequest = initRequest(refresh);
+        Observable.from(containerCollection)
+                .flatMap(new Func1<RequestBuilderContainer, Observable<ShardReplicationOperationRequestBuilder>>()
{
+                    @Override
+                    public Observable<ShardReplicationOperationRequestBuilder> call(RequestBuilderContainer
requestBuilderContainer) {
+                        return Observable.from(requestBuilderContainer.getBuilder())
+                                .map(new Func1<ShardReplicationOperationRequestBuilder,
ShardReplicationOperationRequestBuilder>() {
+                                    @Override
+                                    public ShardReplicationOperationRequestBuilder call(ShardReplicationOperationRequestBuilder
builder) {
+                                        return builder;
+                                    }
+                                });
+                    }
+                })
+                .buffer(config.getIndexBatchSize())
+                .doOnNext(new Action1<List<ShardReplicationOperationRequestBuilder>>()
{
+                    @Override
+                    public void call(List<ShardReplicationOperationRequestBuilder>
builders) {
+                        final BulkRequestBuilder bulkRequest = initRequest(refresh);
+                        for(ShardReplicationOperationRequestBuilder builder : builders) {
+                            if (builder instanceof IndexRequestBuilder) {
+                                bulkRequest.add((IndexRequestBuilder) builder);
+                            }
+                            if (builder instanceof DeleteRequestBuilder) {
+                                bulkRequest.add((DeleteRequestBuilder) builder);
+                            }
+                            sendRequest(bulkRequest);
+                        }
+                    }
+                }).toBlocking().lastOrDefault(null);
 
-            }
-        }
-        sendRequest(bulkRequest);
         for (RequestBuilderContainer container : containerCollection) {
             container.done();
         }


Mime
View raw message