You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by to...@apache.org on 2015/07/15 01:27:26 UTC
incubator-usergrid git commit: First pass POC for running import in
parallel
Repository: incubator-usergrid
Updated Branches:
refs/heads/import-entities-parallel [created] d42015605
First pass POC for running import in parallel
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/d4201560
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/d4201560
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/d4201560
Branch: refs/heads/import-entities-parallel
Commit: d42015605e63016941fd69eeecd67bea16187d95
Parents: c8694f6
Author: Todd Nine <tn...@apigee.com>
Authored: Tue Jul 14 17:27:22 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Tue Jul 14 17:27:22 2015 -0600
----------------------------------------------------------------------
.../apache/usergrid/persistence/Results.java | 4 ++
.../usergrid/services/AbstractService.java | 69 ++++++++++++++++++--
2 files changed, 67 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d4201560/stack/core/src/main/java/org/apache/usergrid/persistence/Results.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/Results.java b/stack/core/src/main/java/org/apache/usergrid/persistence/Results.java
index a16a0f8..e9a3251 100644
--- a/stack/core/src/main/java/org/apache/usergrid/persistence/Results.java
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/Results.java
@@ -518,6 +518,10 @@ public class Results implements Iterable<Entity> {
level = Level.CORE_PROPERTIES;
}
+ public void setEntity( final int index, final Entity entity){
+ this.entities.set( index, entity );
+ }
+
public Results withEntity( Entity resultEntity ) {
setEntity( resultEntity );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d4201560/stack/services/src/main/java/org/apache/usergrid/services/AbstractService.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/AbstractService.java b/stack/services/src/main/java/org/apache/usergrid/services/AbstractService.java
index 569a58c..396b846 100644
--- a/stack/services/src/main/java/org/apache/usergrid/services/AbstractService.java
+++ b/stack/services/src/main/java/org/apache/usergrid/services/AbstractService.java
@@ -48,6 +48,10 @@ import org.apache.usergrid.services.exceptions.ServiceInvocationException;
import org.apache.usergrid.services.exceptions.ServiceResourceNotFoundException;
import org.apache.usergrid.services.exceptions.UnsupportedServiceOperationException;
+import rx.Observable;
+import rx.Subscriber;
+import rx.schedulers.Schedulers;
+
import static org.apache.usergrid.security.shiro.utils.SubjectUtils.getPermissionFromPath;
import static org.apache.usergrid.services.ServiceParameter.filter;
import static org.apache.usergrid.services.ServiceParameter.mergeQueries;
@@ -400,14 +404,67 @@ public abstract class AbstractService implements Service {
List<Entity> entities = results.getEntities();
if ( entities != null ) {
- for ( Entity entity : entities ) {
- Entity imported = importEntity( request, entity );
- if ( imported != entity ) {
- logger.debug( "Import returned new entity instace for {} replacing in results set",
- entity.getUuid() );
- results.replace( imported );
+ importEntitiesParallel(request, results);
+ }
+ }
+
+
+ /**
+ * Import entities in parallel
+ * @param request
+ * @param results
+ */
+ private void importEntitiesParallel(final ServiceRequest request, final Results results ) {
+
+ //create our tuples
+ final Observable<EntityTuple> tuples = Observable.create( new Observable.OnSubscribe<EntityTuple>() {
+ @Override
+ public void call( final Subscriber<? super EntityTuple> subscriber ) {
+ subscriber.onStart();
+
+ final List<Entity> entities = results.getEntities();
+ final int size = entities.size();
+ for ( int i = 0; i < size && !subscriber.isUnsubscribed(); i++ ) {
+ subscriber.onNext( new EntityTuple( i, entities.get( i ) ) );
}
+
+ subscriber.onCompleted();
}
+ } );
+
+ //now process them in parallel up to 10 threads
+
+ tuples.flatMap( tuple -> {
+ //map the entity into the tuple
+ return Observable.just( tuple ).doOnNext( parallelTuple -> {
+ //import the entity and set it at index
+ try {
+
+ final Entity imported = importEntity( request, parallelTuple.entity );
+
+ if(imported != null) {
+ results.setEntity( parallelTuple.index, imported );
+ }
+ }
+ catch ( Exception e ) {
+ throw new RuntimeException(e);
+ }
+ } ).subscribeOn( Schedulers.io() );
+ }, 10 ).toBlocking().lastOrDefault( null );
+ }
+
+
+ /**
+ * Simple tuple representing and entity and it's location within the results
+ */
+ private static final class EntityTuple {
+ private final int index;
+ private final Entity entity;
+
+
+ private EntityTuple( final int index, final Entity entity ) {
+ this.index = index;
+ this.entity = entity;
}
}