issue #270

Author: wowasa

curation-api/src/main/java/eu/clarin/cmdi/curation/api/conf/ApiConfig.java

@@ -38,7 +38,7 @@ public class ApiConfig {
 
    private String clientPassword;
 
-   private int maxThreads;
+   private int maxInQueue;
 
    @Bean
    public VloConfig vloConfig() {

curation-api/src/main/java/eu/clarin/cmdi/curation/api/subprocessor/collection/CollectionAggregator.java

@@ -1,5 +1,6 @@
 package eu.clarin.cmdi.curation.api.subprocessor.collection;
 
+import eu.clarin.cmdi.curation.api.CurationModule;
 import eu.clarin.cmdi.curation.api.conf.ApiConfig;
 import eu.clarin.cmdi.curation.api.entity.CMDCollection;
 import eu.clarin.cmdi.curation.api.entity.CMDInstance;
@@ -12,6 +13,8 @@
 import eu.clarin.cmdi.curation.api.report.collection.sec.ResProxyReport.InvalidReference;
 import eu.clarin.cmdi.curation.api.report.instance.CMDInstanceReport;
 import eu.clarin.cmdi.curation.api.exception.MalFunctioningProcessorException;
+import eu.clarin.cmdi.curation.api.report.profile.CMDProfileReport;
+import eu.clarin.cmdi.curation.cr.CRService;
 import eu.clarin.linkchecker.persistence.model.AggregatedStatus;
 import eu.clarin.linkchecker.persistence.repository.AggregatedStatusRepository;
 import eu.clarin.linkchecker.persistence.repository.UrlRepository;
@@ -28,11 +31,9 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.attribute.BasicFileAttributes;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Stream;
@@ -52,6 +53,8 @@ public class CollectionAggregator {
 
     private final Map<String, Collection<String>> mdSelfLinks = new HashMap<>();
 
+    private final Map<String, AtomicInteger> profileUsage = new HashMap<>();
+
     private final Lock lock = new ReentrantLock();
 
     public CollectionAggregator(ApiConfig conf, ApplicationContext ctx, AggregatedStatusRepository aRep, UrlRepository uRep) {
@@ -77,7 +80,7 @@ public void process(CMDCollection collection, CollectionReport collectionReport)
 
         final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
         //
-        final Semaphore maxTreads = new Semaphore(conf.getMaxThreads());
+        final Semaphore maxInQueue = new Semaphore(conf.getMaxInQueue());
 
         try {
             Files.walkFileTree(collection.getPath(), new FileVisitor<>() {
@@ -91,6 +94,13 @@ public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) {
                 @Override
                 public FileVisitResult visitFile(Path filePath, BasicFileAttributes attrs) {
 
+                    try {
+                        maxInQueue.acquire();
+                    }
+                    catch (InterruptedException e) {
+                        throw new RuntimeException(e);
+                    }
+
                     collectionReport.fileReport.numOfFiles++;
 
                     if (attrs.size() > collectionReport.fileReport.maxFileSize) {
@@ -111,16 +121,16 @@ public FileVisitResult visitFile(Path filePath, BasicFileAttributes attrs) {
 
                         CMDInstanceReport instanceReport;
                         try {
-                            maxTreads.acquire();
                             instanceReport = instance.generateReport();
                         }
-                        catch (MalFunctioningProcessorException | InterruptedException e) {
+                        catch (MalFunctioningProcessorException e) {
                             executor.shutdownNow();
                             throw new RuntimeException(e);
-                        } finally {
-                            maxTreads.release();
                         }
-
+                        finally {
+                            maxInQueue.release();
+                        }
+                        CollectionAggregator.this.lock.lock();
                         addReport(collectionReport, instanceReport);
 
                     }); // end executor.execute
@@ -168,8 +178,6 @@ public FileVisitResult postVisitDirectory(Path dir, IOException exc) {
      */
     public void addReport(CollectionReport collectionReport, CMDInstanceReport instanceReport) {
 
-        this.lock.lock();
-
         try {
             if (!instanceReport.details.isEmpty()) {//only add a record if there are details to report
 
@@ -254,7 +262,10 @@ public void addReport(CollectionReport collectionReport, CMDInstanceReport insta
                                 .count++
                         );
                 collectionReport.facetReport.aggregatedScore += instanceReport.facetReport.score;
-            } else {
+
+                this.profileUsage.computeIfAbsent(instanceReport.profileHeaderReport.getSchemaLocation(), k -> new AtomicInteger()).incrementAndGet();
+            }
+            else {
                 collectionReport.fileReport.numOfFilesNonProcessable++;
             }
         }
@@ -448,6 +459,11 @@ private void calculateAverages(CollectionReport collectionReport) {
             collectionReport.avgScore = collectionReport.aggregatedScore
                     / (double) collectionReport.fileReport.numOfFilesProcessable;
 
+            // add profileUsage
+            this.profileUsage.forEach((k,v) -> this.ctx.getBean(CurationModule.class)
+                                                    .processCMDProfile(k)
+                                                    .collectionUsage
+                                                    .add(new CMDProfileReport.CollectionUsage(collectionReport.fileReport.provider, v.get())));
         }
     }
 }

curation-api/src/test/resources/application.yml

@@ -50,7 +50,7 @@ curation:
       out: >
          /tmp/public
    max_file_size: 100000
-   maxThreads: 10
+   max_in_queue: 10
 logging:
    level:
       root: ERROR 

curation-app/src/main/resources/application.yml

@@ -102,8 +102,8 @@ curation:
       - temporalCoverage
    # maximum size of the CMD file. Bigger files are treated as non-processable
    max_file_size: 100000
-   # maximum number of parallel threads in collection processing
-   maxThreads: ${MAX_THREADS:10}
+   # maximum number CMDI instances in the processing queue when processing collection
+   max_in_queue: ${MAX_IN_QUEUE:10}
    # URL of the web application
    # default:= String ""
    #base_url: