[1.7.x] Support paralleledInBatch in promotion tool (#1209)

* Support paralleledInBatch in promotion tool

* Move batchsize to config and fix commented issues

Author: ruhan1

addons/promote/common/src/main/java/org/commonjava/indy/promote/conf/PromoteConfig.java

@@ -39,6 +39,8 @@ public class PromoteConfig
 
     private static final String AUTOLOCK_HOSTED_REPOS = "autolock.hosted.repos";
 
+    public static final int DEFAULT_PARALLELED_BATCH_SIZE = 100;
+
     public static final long DEFAULT_LOCK_TIMEOUT_SECONDS = 30;
 
     public static final boolean DEFAULT_AUTOLOCK = true;
@@ -53,6 +55,8 @@ public class PromoteConfig
 
     private Long lockTimeoutSeconds;
 
+    private Integer paralleledBatchSize;
+
     public PromoteConfig()
     {
     }
@@ -131,4 +135,19 @@ public InputStream getDefaultConfig()
                      .getResourceAsStream( "default-promote.conf" );
     }
 
+    @ConfigName( "paralleled.batch.size" )
+    public void setParalleledBatchSize( Integer paralleledBatchSize )
+    {
+        this.paralleledBatchSize = paralleledBatchSize;
+    }
+
+    public int getParalleledBatchSize()
+    {
+        int ret = paralleledBatchSize == null ? DEFAULT_PARALLELED_BATCH_SIZE : paralleledBatchSize;
+        if ( ret <= 0 )
+        {
+            ret = DEFAULT_PARALLELED_BATCH_SIZE;
+        }
+        return ret;
+    }
 }

addons/promote/common/src/main/java/org/commonjava/indy/promote/validate/PromotionValidationTools.java

@@ -27,6 +27,7 @@
 import org.commonjava.indy.data.ArtifactStoreQuery;
 import org.commonjava.indy.model.core.ArtifactStore;
 import org.commonjava.indy.model.core.StoreKey;
+import org.commonjava.indy.promote.conf.PromoteConfig;
 import org.commonjava.indy.promote.validate.model.ValidationRequest;
 import org.commonjava.indy.util.LocationUtils;
 import org.commonjava.atlas.maven.graph.rel.ProjectRelationship;
@@ -112,6 +113,9 @@ public class PromotionValidationTools
     @Inject
     private ContentDigester contentDigester;
 
+    @Inject
+    private PromoteConfig promoteConfig;
+
     @Inject
     @WeftManaged
     @ExecutorConfig( named = "promote-validation-rules-executor", threads = 8 )
@@ -124,7 +128,8 @@ protected PromotionValidationTools()
     public PromotionValidationTools( final ContentManager manager, final StoreDataManager storeDataManager,
                                      final MavenPomReader pomReader, final MavenMetadataReader metadataReader,
                                      final MavenModelProcessor modelProcessor, final TypeMapper typeMapper,
-                                     final TransferManager transferManager, final ContentDigester contentDigester )
+                                     final TransferManager transferManager, final ContentDigester contentDigester,
+                                     final Executor ruleParallelExecutor, final PromoteConfig config )
     {
         contentManager = manager;
         this.storeDataManager = storeDataManager;
@@ -134,6 +139,8 @@ public PromotionValidationTools( final ContentManager manager, final StoreDataMa
         this.typeMapper = typeMapper;
         this.transferManager = transferManager;
         this.contentDigester = contentDigester;
+        this.ruleParallelExecutor = ruleParallelExecutor;
+        this.promoteConfig = config;
     }
 
     public StoreKey[] getValidationStoreKeys( final ValidationRequest request, final boolean includeSource )
@@ -561,6 +568,72 @@ public <K, V> void paralleledEach( Map<K, V> map, Closure closure )
         runParallelAndWait( entries, closure, logger );
     }
 
+    public <T> void paralleledInBatch( Collection<T> collection, Closure closure )
+    {
+        int batchSize = promoteConfig.getParalleledBatchSize();
+        logger.trace( "Exe parallel on collection {} with closure {} in batch {}", collection, closure, batchSize );
+        Collection<Collection<T>> batches = batch( collection, batchSize );
+        runParallelInBatchAndWait( batches, closure, logger );
+    }
+
+    public <T> void paralleledInBatch( T[] array, Closure closure )
+    {
+        int batchSize = promoteConfig.getParalleledBatchSize();
+        logger.trace( "Exe parallel on array {} with closure {} in batch {}", array, closure, batchSize );
+        Collection<Collection<T>> batches = batch( Arrays.asList( array ), batchSize );
+        runParallelInBatchAndWait( batches, closure, logger );
+    }
+
+    public <K, V> void paralleledInBatch( Map<K, V> map, Closure closure )
+    {
+        int batchSize = promoteConfig.getParalleledBatchSize();
+        Set<Map.Entry<K, V>> entries = map.entrySet();
+        logger.trace( "Exe parallel on map {} with closure {} in batch {}", entries, closure, batchSize );
+        Collection<Collection<Map.Entry<K, V>>> batches = batch( entries, batchSize );
+        runParallelInBatchAndWait( batches, closure, logger );
+    }
+
+    private <T> void runParallelInBatchAndWait( Collection<Collection<T>> batches, Closure closure, Logger logger )
+    {
+        final CountDownLatch latch = new CountDownLatch( batches.size() );
+        batches.forEach( batch -> ruleParallelExecutor.execute( () -> {
+            try
+            {
+                logger.trace( "The paralleled exe on batch {}", batch );
+                batch.forEach( e -> closure.call( e ) );
+            }
+            finally
+            {
+                latch.countDown();
+            }
+        } ) );
+
+        waitForCompletion( latch );
+    }
+
+    private <T> Collection<Collection<T>> batch( Collection<T> collection, int batchSize )
+    {
+        Collection<Collection<T>> batches = new ArrayList<>();
+        Collection<T> batch = new ArrayList<>( batchSize );
+        int count = 0;
+        for ( T t : collection )
+        {
+            ( (ArrayList<T>) batch ).add( t );
+            count++;
+            if ( count >= batchSize )
+            {
+                ( (ArrayList<Collection<T>>) batches ).add( batch );
+                batch = new ArrayList<>( batchSize );
+                count = 0;
+            }
+        }
+        if ( batch != null && !batch.isEmpty() )
+        {
+            ( (ArrayList<Collection<T>>) batches ).add( batch ); // first batch
+        }
+        return batches;
+    }
+
     private <T> void runParallelAndWait( Collection<T> runCollection, Closure closure, Logger logger )
     {
         Set<T> todo = new HashSet<>( runCollection );
@@ -577,6 +650,11 @@ private <T> void runParallelAndWait( Collection<T> runCollection, Closure closur
             }
         } ) );
 
+        waitForCompletion( latch );
+    }
+
+    private void waitForCompletion( CountDownLatch latch )
+    {
         try
         {
             // true if the count reached zero and false if timeout

addons/promote/common/src/test/java/org/commonjava/indy/promote/data/PromotionManagerTest.java

@@ -172,15 +172,15 @@ public void setup()
         WeftExecutorService validateService =
                         new PoolWeftExecutorService( "test-validate-executor", (ThreadPoolExecutor) Executors.newCachedThreadPool(), 2, 10f, false,null, null );
         MavenModelProcessor modelProcessor = new MavenModelProcessor();
+
+        PromoteConfig config = new PromoteConfig();
         validator = new PromotionValidator( validationsManager,
                                             new PromotionValidationTools( contentManager, storeManager,
                                                                           galleyParts.getPomReader(),
                                                                           galleyParts.getMavenMetadataReader(),
                                                                           modelProcessor, galleyParts.getTypeMapper(),
                                                                           galleyParts.getTransferManager(),
-                                                                          contentDigester ), storeManager, downloadManager, validateService, null );
-
-        PromoteConfig config = new PromoteConfig();
+                                                                          contentDigester, null, config ), storeManager, downloadManager, validateService, null );
 
         WeftExecutorService svc =
                 new PoolWeftExecutorService( "test-executor", (ThreadPoolExecutor) Executors.newCachedThreadPool(), 2, 10f, false,null, null );

addons/promote/common/src/test/java/org/commonjava/indy/promote/validate/PromotionValidationToolsTest.java

@@ -0,0 +1,84 @@
+package org.commonjava.indy.promote.validate;
+
+import groovy.lang.Closure;
+import org.commonjava.indy.promote.conf.PromoteConfig;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+import static junit.framework.TestCase.assertTrue;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+public class PromotionValidationToolsTest
+{
+    final String[] array = { "this", "is", "a", "err_weird", "test", "err_for", "paralleled", "err_in", "batch" };
+
+    @Test
+    public void testParalleledInBatch()
+    {
+        PromoteConfig config = new PromoteConfig();
+        Executor executor = Executors.newCachedThreadPool();
+        PromotionValidationTools tools =
+                        new PromotionValidationTools( null, null, null, null, null, null, null, null, executor,
+                                                      config );
+
+        List<String> errors = Collections.synchronizedList( new ArrayList<>() );
+        Closure closure = new Closure<String>( null )
+        {
+            @Override
+            public String call( Object arg )
+            {
+                if ( ( (String) arg ).startsWith( "err_" ) )
+                {
+                    errors.add( (String) arg );
+                }
+                return null;
+            }
+        };
+
+        tools.paralleledInBatch( array, closure );
+        verifyIt( errors );
+    }
+
+    @Test
+    public void testParalleledInBatch_smallSize()
+    {
+        PromoteConfig config = new PromoteConfig();
+        config.setParalleledBatchSize( 2 );
+
+        Executor executor = Executors.newCachedThreadPool();
+        PromotionValidationTools tools =
+                        new PromotionValidationTools( null, null, null, null, null, null, null, null, executor,
+                                                      config );
+
+        List<String> errors = Collections.synchronizedList( new ArrayList<>() );
+        Closure closure = new Closure<String>( null )
+        {
+            @Override
+            public String call( Object arg )
+            {
+                if ( ( (String) arg ).startsWith( "err_" ) )
+                {
+                    errors.add( (String) arg );
+                }
+                return null;
+            }
+        };
+
+        tools.paralleledInBatch( array, closure );
+        verifyIt( errors );
+    }
+
+    private void verifyIt( List<String> errors )
+    {
+        assertThat( errors.size(), equalTo( 3 ) );
+        assertTrue( errors.contains( "err_weird" ) );
+        assertTrue( errors.contains( "err_for" ) );
+        assertTrue( errors.contains( "err_in" ) );
+    }
+}