Changeset 7022

Timestamp:

Feb 6, 2023, 2:45:22 PM (8 weeks ago)

Author:

Nicklas Nordborg

Message:

References #1453: Make the release exporter multi-threaded

Use 1/3 of the available cpus for the export. The ReleaseExporter will start multiple threads. It is important that all *Writer implementations are thread-safe.

Location:

extensions/net.sf.basedb.reggie/trunk/src/net/sf/basedb/reggie

Files:

• converter/DaysSinceRefDateConverter.java (modified) (3 diffs)
• plugins/release/CohortItem.java (modified) (1 diff)
• plugins/release/CohortWriter.java (modified) (1 diff)
• plugins/release/IncaWriter.java (modified) (3 diffs)
• plugins/release/JsonWriter.java (modified) (2 diffs)
• plugins/release/ReleaseExporter.java (modified) (6 diffs)
• plugins/release/ScriptWriter.java (modified) (3 diffs)

extensions/net.sf.basedb.reggie/trunk/src/net/sf/basedb/reggie/converter/DaysSinceRefDateConverter.java

@@ -19 +19 @@
   private static long MILLIS_IN_A_DAY = 24 * MILLIS_IN_AN_HOUR;
   
-  private Long refTime;
+  // Need ThreadLocal so a single instance can handle 
+  // multiple threads at the same time
+  private final ThreadLocal<Long> refTime;
   
   public DaysSinceRefDateConverter()
-  {}
+  {
+    this.refTime = new ThreadLocal<>();
+  }
   
   /*
@@ -31 +35 @@
   public Integer convert(Date value) 
   {
-    return value == null || refTime == null ? null : (int)((value.getTime() - refTime) / MILLIS_IN_A_DAY);
+    Long refT = refTime.get();
+    return value == null || refT == null ? null : (int)((value.getTime() - refT) / MILLIS_IN_A_DAY);
   }
   // ----------------------------------
@@ -42 +47 @@
     // The calculations rely on rounding down to return the correct result but 
     // it is important that no time part is present in the date values
-    this.refTime = refDate == null ? null : refDate.getTime() - MILLIS_IN_AN_HOUR;
+    this.refTime.set(refDate == null ? null : refDate.getTime() - MILLIS_IN_AN_HOUR);
   }
   

extensions/net.sf.basedb.reggie/trunk/src/net/sf/basedb/reggie/plugins/release/CohortItem.java

@@ -124 +124 @@
     @since 4.21
   */
-  public CohortItem(ReleaseExporter exporter, DbControl dc, SnapshotManager manager, QueryManager queryManager, int biosourceId, ReleaseWriterOptions options)
+  public CohortItem(ReleaseExporter exporter, DbControl dc, SnapshotManager manager, QueryManager queryManager, int biosourceId)
   {
     this.exporter = exporter;

extensions/net.sf.basedb.reggie/trunk/src/net/sf/basedb/reggie/plugins/release/CohortWriter.java

@@ -19 +19 @@
   Subclasses must implement the {@link #writeCohortData(CohortItem)}
   method and make sure the header line is written first.
+  
+  Implementing classes need to be thread-safe since they are
+  used with multiple exporter threads at the same time.
+  See {@link ReleaseExporter#exportCohortData()}.
 
   @author nicklas

extensions/net.sf.basedb.reggie/trunk/src/net/sf/basedb/reggie/plugins/release/IncaWriter.java

@@ -41 +41 @@
 
   private final Annotationtype[] incaTypes;
-  private DaysSinceRefDateConverter daysSinceRefDate;
-  private ValueConverter<?, ?>[] incaConverters;
+  private final DaysSinceRefDateConverter daysSinceRefDate;
+  private final ValueConverter<?, ?>[] incaConverters;
   
+  @SuppressWarnings("rawtypes")
   public IncaWriter(DbControl dc, ReleaseWriterOptions options)
   {
@@ -66 +67 @@
       index++;
     }
+    
+    incaConverters = new ValueConverter[incaTypes.length];
+    daysSinceRefDate = new DaysSinceRefDateConverter();
   }
   
@@ -109 +113 @@
 
   @Override
-  @SuppressWarnings("rawtypes")
   public List<CohortTypeDef> getTypeDefsInJSON()
   {
     DbControl dc = getDbControl();
     CohortTypeDefFactory incaFactory = new CohortTypeDefFactory(dc, Item.SAMPLE, "Case");
-    incaConverters = new ValueConverter[incaTypes.length];
-    daysSinceRefDate = new DaysSinceRefDateConverter();
     Unit days = UnitUtil.getUnit(dc, Quantity.TIME, "d");
     

extensions/net.sf.basedb.reggie/trunk/src/net/sf/basedb/reggie/plugins/release/JsonWriter.java

@@ -126 +126 @@
   public void writeJsonData(CohortItem item) 
   {
+    // This method may be called by multiple threads
+    // The CohortWriter implementations are assumed to be thread-safe
     JSONArray json = new FilteredJSONArray(new NotNullFilter<>(false));
     for (CohortWriter cw : itemWriters)
@@ -132 +134 @@
     }
 
-    String itemName = item.getName();
-    ExportOutputStream out = null;
-    try
-    {
-      String file = "/json/"+itemName+".json";
-      jsonFiles.add(file);
-      out = location.getOutputStream(file, false);
-      out.setCharacterSet("UTF-8");
-      out.setMimeType("application/json");
-
-      Writer writer = new OutputStreamWriter(out, Charset.forName("UTF-8"));
-      json.writeJSONString(writer);
-      writer.flush();
-      out.flush();
-      writer.close();
-    }
-    catch (IOException ex)
-    {
-      throw new RuntimeException(ex);
-    }
-    finally
-    {
-      FileUtil.close(out);
+    // This part need to be synchronized since the OutputLocation
+    // implementation may not be thread-safe
+    synchronized (this)
+    {
+      String itemName = item.getName();
+      ExportOutputStream out = null;
+      try
+      {
+        String file = "/json/"+itemName+".json";
+        jsonFiles.add(file);
+        out = location.getOutputStream(file, false);
+        out.setCharacterSet("UTF-8");
+        out.setMimeType("application/json");
+  
+        Writer writer = new OutputStreamWriter(out, Charset.forName("UTF-8"));
+        json.writeJSONString(writer);
+        writer.flush();
+        out.flush();
+        writer.close();
+      }
+      catch (IOException ex)
+      {
+        throw new RuntimeException(ex);
+      }
+      finally
+      {
+        FileUtil.close(out);
+      }
     }
   }

extensions/net.sf.basedb.reggie/trunk/src/net/sf/basedb/reggie/plugins/release/ReleaseExporter.java

@@ -5 +5 @@
 import java.util.List;
 import java.util.Map;
+import java.util.Queue;
 import java.util.Set;
 import java.util.TreeMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
 
 import net.sf.basedb.core.Application;
@@ -17 +23 @@
 import net.sf.basedb.core.ItemQuery;
 import net.sf.basedb.core.ProgressReporter;
+import net.sf.basedb.core.SessionControl;
 import net.sf.basedb.core.SimpleProgressReporter;
 import net.sf.basedb.core.plugin.ExportOutputStream;
 import net.sf.basedb.core.query.Hql;
 import net.sf.basedb.core.query.Orders;
-import net.sf.basedb.core.signal.ThreadSignalHandler;
+import net.sf.basedb.core.signal.SignalException;
 import net.sf.basedb.core.snapshot.SnapshotManager;
 import net.sf.basedb.reggie.Reggie;
@@ -46 +53 @@
   private OutputLocation outputLocation;
 
-  private ReleaseWriterOptions options;
+  private final ReleaseWriterOptions options;
   
   private ItemList list;
@@ -136 +143 @@
   private void exportCohortData(DbControl dc, QueryManager queryManager, List<BioSource> biosources, ProgressReporter progress)
   {
-    progress.display(15, "Exporting cohort data for " + biosources.size() + " biosources...");
-
+    // Use 1/3 of the available processors
+    int numThreads = Math.max(1, Runtime.getRuntime().availableProcessors()/3);
     // Counters used for progress reporting
     int totalCount = biosources.size();
+    progress.display(15, "Exporting cohort data for " + totalCount + " biosources ("+numThreads+" threads)...");
+    
+    ScriptWriter scriptWriter = null;
+    if (options.exportFileSyncScripts())
+    {
+      scriptWriter = new ScriptWriter(dc, outputLocation, options);
+    }
+    
+    JsonWriter jsonWriter = new JsonWriter(dc, outputLocation, options);
+    jsonWriter.registerItemWriter(new PatientWriter(dc, options));
+    jsonWriter.registerItemWriter(new RetractWriter(dc, options));
+    jsonWriter.registerItemWriter(new NotAskedWriter(dc, options));
+    jsonWriter.registerItemWriter(new NoWriter(dc, options));
+    jsonWriter.registerItemWriter(new BloodWriter(dc, options));
+    jsonWriter.registerItemWriter(new BloodDnaWriter(dc, options));
+    jsonWriter.registerItemWriter(new CaseWriter(dc, options));
+    jsonWriter.registerItemWriter(new IncaWriter(dc, options));
+    jsonWriter.registerItemWriter(new SpecimenWriter(dc, options, outputLocation, scriptWriter));
+    jsonWriter.registerItemWriter(new NoSpecimenWriter(dc, options));
+    jsonWriter.registerItemWriter(new LysateWriter(dc, options));
+    jsonWriter.registerItemWriter(new RnaWriter(dc, options));
+    jsonWriter.registerItemWriter(new DnaWriter(dc, options));
+    jsonWriter.registerItemWriter(new FlowThroughWriter(dc, options));
+    jsonWriter.registerItemWriter(new LibraryWriter(dc, options));
+    jsonWriter.registerItemWriter(new MergedWriter(dc, options, scriptWriter));
+    jsonWriter.registerItemWriter(new AlignedWriter(dc, options, scriptWriter));
+    jsonWriter.registerItemWriter(new StringTieWriter(dc, options, scriptWriter));
+    jsonWriter.registerItemWriter(new CufflinksWriter(dc, options, scriptWriter));
+    jsonWriter.registerItemWriter(new VariantCallingWriter(dc, options, scriptWriter));
+    jsonWriter.registerItemWriter(new MethylationWriter(dc, options, scriptWriter));
+    jsonWriter.registerItemWriter(new OncoarrayWriter(dc, options, scriptWriter));
+    
+    jsonWriter.writeIndex(list, biosources);
+    jsonWriter.writeTypeDefs();
+    
+    Queue<BioSource> bioSourceQueue = new ConcurrentLinkedQueue<>(biosources);
+    
+    CallableExporter[] exporters = new CallableExporter[numThreads];
+    ExecutorService executor = Executors.newFixedThreadPool(numThreads, new ExporterThreadFactory());
+    
     int count = 0;
-    
-    DbControl readerDc = null;
-    SnapshotManager readerManager = null;
     try
     {
-      ScriptWriter scriptWriter = null;
-      if (options.exportFileSyncScripts())
-      {
-        scriptWriter = new ScriptWriter(dc, outputLocation, options);
+      // Start the exporter threads
+      for (int threadNo = 0; threadNo < numThreads; threadNo++)
+      {
+        CallableExporter exp = new CallableExporter(this, dc, jsonWriter, bioSourceQueue);
+        exporters[threadNo] = exp;
+        executor.execute(exp);
       }
       
-      JsonWriter jsonWriter = new JsonWriter(dc, outputLocation, options);
-      jsonWriter.registerItemWriter(new PatientWriter(dc, options));
-      jsonWriter.registerItemWriter(new RetractWriter(dc, options));
-      jsonWriter.registerItemWriter(new NotAskedWriter(dc, options));
-      jsonWriter.registerItemWriter(new NoWriter(dc, options));
-      jsonWriter.registerItemWriter(new BloodWriter(dc, options));
-      jsonWriter.registerItemWriter(new BloodDnaWriter(dc, options));
-      jsonWriter.registerItemWriter(new CaseWriter(dc, options));
-      jsonWriter.registerItemWriter(new IncaWriter(dc, options));
-      jsonWriter.registerItemWriter(new SpecimenWriter(dc, options, outputLocation, scriptWriter));
-      jsonWriter.registerItemWriter(new NoSpecimenWriter(dc, options));
-      jsonWriter.registerItemWriter(new LysateWriter(dc, options));
-      jsonWriter.registerItemWriter(new RnaWriter(dc, options));
-      jsonWriter.registerItemWriter(new DnaWriter(dc, options));
-      jsonWriter.registerItemWriter(new FlowThroughWriter(dc, options));
-      jsonWriter.registerItemWriter(new LibraryWriter(dc, options));
-      jsonWriter.registerItemWriter(new MergedWriter(dc, options, scriptWriter));
-      jsonWriter.registerItemWriter(new AlignedWriter(dc, options, scriptWriter));
-      jsonWriter.registerItemWriter(new StringTieWriter(dc, options, scriptWriter));
-      jsonWriter.registerItemWriter(new CufflinksWriter(dc, options, scriptWriter));
-      jsonWriter.registerItemWriter(new VariantCallingWriter(dc, options, scriptWriter));
-      jsonWriter.registerItemWriter(new MethylationWriter(dc, options, scriptWriter));
-      jsonWriter.registerItemWriter(new OncoarrayWriter(dc, options, scriptWriter));
-      
-      jsonWriter.writeIndex(list, biosources);
-      jsonWriter.writeTypeDefs();
-            
-      readerDc = dc.getSessionControl().newDbControl(dc.getName());
-      readerManager = new SnapshotManager();
-      for (BioSource bs : biosources)
-      {
-        count++;
-        if (count % 10 == 0)
+      // Wait in this loop until all biosources have been processed
+      while (count < totalCount)
+      {
+        try
         {
-          ThreadSignalHandler.checkInterrupted();
-          if (count % 100 == 0)
-          {
-            // Rollback and create new DbControl and snapshot 
-            // manager to allow GC to reclaim memory
-            readerDc.close();
-            readerDc = dc.getSessionControl().newDbControl(dc.getName());
-            readerManager = new SnapshotManager();
-          }
-          
-          progress.display(15 + ((80*count)/totalCount), 
-            "Exporting cohort data (" + count + " of " + totalCount + ")...");
+          Thread.sleep(2000);
         }
-        
-        CohortItem item = new CohortItem(this, readerDc, readerManager, queryManager, bs.getId(), options);
-        jsonWriter.writeJsonData(item);
-      }
-      
-      jsonWriter.writeFiles();
-      jsonWriter.writeBatchIndexLookupFiles(batchIndexProxies);
-      if (scriptWriter != null) scriptWriter.writeScripts();
-      
-      progress.display(98, "Exported cohort data (" + count + " items)");
+        catch (InterruptedException ex)
+        {
+          throw new SignalException("Aborted by user");
+        }
+        // Important loop for progress reporting AND for detecting if one of
+        // the exporter threads have failed (getCount() throws an exception)
+        count = 0;
+        for (CallableExporter exp : exporters)
+        {
+          count += exp.getCount();
+        }
+        progress.display(15 + ((80*count)/totalCount), 
+          "Exporting cohort data ("+numThreads+" threads; " + count + " of " + totalCount + ")...");
+      }
     }
     finally
     {
-      if (readerDc != null) readerDc.close();
-    }
+      try
+      {
+        // This will send an interrupt to all running threads
+        // There should be none if everything went ok, but if one
+        // thread fails, getCount() will throw an exception and we
+        // will ask the other threads to abort. We give them 10 seconds...
+        executor.shutdownNow();
+        executor.awaitTermination(10, TimeUnit.SECONDS);
+      }
+      catch (InterruptedException ex)
+      {}
+    }
+    
+    jsonWriter.writeFiles();
+    jsonWriter.writeBatchIndexLookupFiles(batchIndexProxies);
+    if (scriptWriter != null) scriptWriter.writeScripts();
+    
+    progress.display(98, "Exported cohort data (" + count + " items)");
   }
   
@@ -245 +271 @@
     @param batchKey Typically a date identifying the "batch"
   */
-  public String getBatchIndexProxy(String batchName, String batchKey)
+  public synchronized String getBatchIndexProxy(String batchName, String batchKey)
   {
     Map<String, String> batch = batchIndexProxies.get(batchName);
@@ -265 +291 @@
   }
   
-  
+  /**
+    Implementation for multi-threading support. Any number of instances can be created with
+    the same parameters and started at the same time. Each instance will take biosource items
+    one by one from the queue and will continue processing as long as there are items left.
+    
+    The main code should call getCount() at regular intervals to get information about the
+    progress. This will also check if there has been an error during the processing and
+    re-throw the same error.
+    
+    @since 4.44
+  */
+  public static class CallableExporter
+    implements Runnable
+  {
+    private final ReleaseExporter exporter;
+    private final DbControl writerDc;
+    private final JsonWriter jsonWriter;
+    private final Queue<BioSource> biosources;
+    
+    private volatile int count;
+    private volatile RuntimeException error;
+    
+    CallableExporter(ReleaseExporter exporter, DbControl writerDc, JsonWriter jsonWriter, Queue<BioSource> biosources)
+    {
+      this.exporter = exporter;
+      this.writerDc = writerDc;
+      this.jsonWriter = jsonWriter;
+      this.biosources = biosources;
+    }
+
+    /**
+      Get the number of biosources that have been processed so far by this
+      instance. If there has been an error during the processing, this
+      method will re-throw that error.
+    */
+    public int getCount()
+    {
+      if (error != null) throw error;
+      return count;
+    }
+    
+    @Override
+    public void run() 
+    {
+      DbControl readerDc = null;
+      SnapshotManager manager = null;
+      QueryManager queryManager = null;
+      
+      try
+      {
+        SessionControl sc = writerDc.getSessionControl();
+        readerDc = sc.newDbControl(writerDc.getName());
+        manager = new SnapshotManager();
+        queryManager = new QueryManager(readerDc);
+        
+        do
+        {
+          BioSource bs = biosources.poll();
+          // Exit if there are no more biosources or
+          // if this thread has been interrupted (eg. due to
+          // aborting or an error in another thread)
+          if (bs == null || Thread.interrupted()) break;
+          
+          if (count % 100 == 0)
+          {
+            // Rollback and create new DbControl and snapshot 
+            // manager to allow GC to reclaim memory
+            readerDc.close();
+            readerDc = sc.newDbControl(writerDc.getName());
+            manager = new SnapshotManager();
+            queryManager = new QueryManager(readerDc);
+          }
+          
+          CohortItem item = new CohortItem(exporter, readerDc, manager, queryManager, bs.getId());
+          jsonWriter.writeJsonData(item);
+          count++;
+        } while (true);
+      }
+      catch (RuntimeException ex)
+      {
+        error = ex;
+      }
+      finally
+      {
+        if (readerDc != null) readerDc.close();
+      }
+    }
+  }
+  
+  /**
+    Thread factory for exporter threads.
+  */
+  public static class ExporterThreadFactory
+    implements ThreadFactory
+  {
+  
+    private final ThreadGroup grp;
+    private int nThreads;
+    
+    public ExporterThreadFactory() 
+    {
+      this.grp = new ThreadGroup("ReleaseExporterThread");
+    }
+    
+    @Override
+    public Thread newThread(Runnable r) 
+    {
+      nThreads++;
+      return new Thread(grp, r, grp.getName()+"."+nThreads);
+    }
+  }
+
 }

extensions/net.sf.basedb.reggie/trunk/src/net/sf/basedb/reggie/plugins/release/ScriptWriter.java

@@ -68 +68 @@
     Add a directory that should be created as part of this release.
   */
-  public void addMkDir(String path)
+  public synchronized void addMkDir(String path)
   {
     if (path != null) 
@@ -89 +89 @@
     @param jsonFiles
   */
-  public void addFiles(String releasePath, String sourcePath, String fileList, JSONArray jsonFiles)
+  public synchronized void addFiles(String releasePath, String sourcePath, String fileList, JSONArray jsonFiles)
   {
     addMkDir(releasePath);
@@ -140 +140 @@
     @since 4.21
   */
-  public void addLinkToExistingFile(String releasePath, String filename, String releasedVersion)
+  public synchronized void addLinkToExistingFile(String releasePath, String filename, String releasedVersion)
   {
     addMkDir(releasePath);