refactored data caching

Author: TomBlock

cems/src/main/python/pmonitor.py

@@ -590,7 +590,7 @@ def _inquire_status(self):
         inquiry = {}
         # retrieve status with polling script
         if external_ids:
-            print('retrieving status with ' + self._polling + ' ' + ' '.join(external_ids))
+            # print('retrieving status with ' + self._polling + ' ' + ' '.join(external_ids))
             process = subprocess.Popen(self._polling + ' ' + ' '.join(external_ids),
                                        shell=True, bufsize=1, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
             for l in process.stdout:

post-processing-tool/src/main/java/com/bc/fiduceo/post/plugin/era5/MatchupFields.java

@@ -107,18 +107,18 @@ void compute(Configuration config, NetcdfFile reader, NetcdfFileWriter writer) t
 
                     final InterpolationContext interpolationContext = Era5PostProcessing.getInterpolationContext(lonLayer, latLayer);
                     final Rectangle layerRegion = interpolationContext.getEra5Region();
-                    final int[] offset = new int[]{0, layerRegion.y, layerRegion.x};
-                    final int[] shape = new int[]{1, layerRegion.height, layerRegion.width};
+                    final int[] offset = new int[]{layerRegion.y, layerRegion.x};
+                    final int[] shape = new int[]{layerRegion.height, layerRegion.width};
 
                     // iterate over time stamps
                     for (int t = 0; t < numTimeSteps; t++) {
                         timeIndex.set(m, t);
                         final int timeStamp = targetTimeArray.getInt(timeIndex);
-                        final Variable variable = variableCache.get(variableKey, timeStamp);
+                        VariableCache.CacheEntry cacheEntry = variableCache.get(variableKey, timeStamp);
 
                         // read and get rid of fake z-dimension
-                        Array subset = variable.read(offset, shape).reduce();
-                        subset = NetCDFUtils.scaleIfNecessary(variable, subset);
+                        Array subset = cacheEntry.array.section(offset, shape);
+                        subset = NetCDFUtils.scaleIfNecessary(cacheEntry.variable, subset);
                         final Index subsetIndex = subset.getIndex();
                         final BilinearInterpolator bilinearInterpolator = interpolationContext.get(0, 0);
 

post-processing-tool/src/main/java/com/bc/fiduceo/post/plugin/era5/SatelliteFields.java

@@ -24,20 +24,20 @@ class SatelliteFields extends FieldsProcessor {
     private List<Dimension> dimension3d;
     private Map<String, TemplateVariable> variables;
 
-    static Array readSubset(int numLayers, Rectangle era5RasterPosition, Variable variable) throws IOException, InvalidRangeException {
+    static Array readSubset(int numLayers, Rectangle era5RasterPosition, VariableCache.CacheEntry cacheEntry) throws IOException, InvalidRangeException {
         Array subset;
 
         final int maxRequestedX = era5RasterPosition.x + era5RasterPosition.width - 1;
         if (era5RasterPosition.x < 0 || maxRequestedX >= RASTER_WIDTH) {
-            subset = readVariableDataOverlapped(numLayers, era5RasterPosition, variable);
+            subset = readVariableDataOverlapped(numLayers, era5RasterPosition, cacheEntry.array);
         } else {
-            subset = readVariableData(numLayers, era5RasterPosition, variable);
+            subset = readVariableData(numLayers, era5RasterPosition, cacheEntry.array);
         }
 
-        return NetCDFUtils.scaleIfNecessary(variable, subset);
+        return NetCDFUtils.scaleIfNecessary(cacheEntry.variable, subset);
     }
 
-    private static Array readVariableDataOverlapped(int numLayers, Rectangle era5RasterPosition, Variable variable) throws IOException, InvalidRangeException {
+    private static Array readVariableDataOverlapped(int numLayers, Rectangle era5RasterPosition, Array array) throws IOException, InvalidRangeException {
         Array subset;
         int xMin = 0;
         int xMax;
@@ -53,20 +53,20 @@ private static Array readVariableDataOverlapped(int numLayers, Rectangle era5Ras
             leftWidth = era5RasterPosition.width - rightWidth;
         }
         final Rectangle leftEraPos = new Rectangle(xMin, era5RasterPosition.y, leftWidth, era5RasterPosition.height);
-        final Array leftSubset = readVariableData(numLayers, leftEraPos, variable);
+        final Array leftSubset = readVariableData(numLayers, leftEraPos, array);
 
         final Rectangle rightEraPos = new Rectangle(xMax, era5RasterPosition.y, rightWidth, era5RasterPosition.height);
-        final Array rightSubset = readVariableData(numLayers, rightEraPos, variable);
+        final Array rightSubset = readVariableData(numLayers, rightEraPos, array);
 
-        subset = mergeData(leftSubset, rightSubset, numLayers, era5RasterPosition, variable);
+        subset = mergeData(leftSubset, rightSubset, numLayers, era5RasterPosition, array);
         return subset;
     }
 
-    static Array mergeData(Array leftSubset, Array rightSubset, int numLayers, Rectangle era5RasterPosition, Variable variable) {
-        final int rank = variable.getRank();
+    static Array mergeData(Array leftSubset, Array rightSubset, int numLayers, Rectangle era5RasterPosition, Array array) {
+        final int rank = array.getRank();
         final Array mergedArray;
         if (rank == 4) {
-            mergedArray = Array.factory(variable.getDataType(), new int[]{numLayers, era5RasterPosition.height, era5RasterPosition.width});
+            mergedArray = Array.factory(array.getDataType(), new int[]{numLayers, era5RasterPosition.height, era5RasterPosition.width});
             if (era5RasterPosition.x < 0) {
                 final int xMax = era5RasterPosition.width + era5RasterPosition.x;
                 mergeArrays_3D(leftSubset, rightSubset, era5RasterPosition, mergedArray, xMax);
@@ -75,7 +75,7 @@ static Array mergeData(Array leftSubset, Array rightSubset, int numLayers, Recta
                 mergeArrays_3D(rightSubset, leftSubset, era5RasterPosition, mergedArray, xMax);
             }
         } else {
-            mergedArray = Array.factory(variable.getDataType(), new int[]{era5RasterPosition.height, era5RasterPosition.width});
+            mergedArray = Array.factory(array.getDataType(), new int[]{era5RasterPosition.height, era5RasterPosition.width});
             if (era5RasterPosition.x < 0) {
                 final int xMax = era5RasterPosition.width + era5RasterPosition.x;
                 mergeArrays(leftSubset, rightSubset, era5RasterPosition, mergedArray, xMax);
@@ -141,24 +141,23 @@ private static void mergeArrays_3D(Array leftSubset, Array rightSubset, Rectangl
         }
     }
 
-    private static Array readVariableData(int numLayers, Rectangle era5RasterPosition, Variable variable) throws IOException, InvalidRangeException {
-        final int rank = variable.getRank();
+    private static Array readVariableData(int numLayers, Rectangle era5RasterPosition, Array array) throws IOException, InvalidRangeException {
+        final int rank = array.getRank();
         Array subset;
-        if (rank == 3) {
+        if (rank == 2) {
+            final int[] origin = new int[]{era5RasterPosition.y, era5RasterPosition.x};
+            final int[] shape = new int[]{era5RasterPosition.height, era5RasterPosition.width};
+            final int[] stride = new int[]{1, 1};
+            subset = array.sectionNoReduce(origin, shape, stride);
+        } else if (rank == 3) {
             final int[] origin = new int[]{0, era5RasterPosition.y, era5RasterPosition.x};
-            final int[] shape = new int[]{1, era5RasterPosition.height, era5RasterPosition.width};
-            subset = variable.read(origin, shape);
-        } else if (rank == 4) {
-            final int[] origin = new int[]{0, 0, era5RasterPosition.y, era5RasterPosition.x};
-            final int[] shape = new int[]{1, numLayers, era5RasterPosition.height, era5RasterPosition.width};
-            subset = variable.read(origin, shape);
+            final int[] shape = new int[]{numLayers, era5RasterPosition.height, era5RasterPosition.width};
+            final int[] stride = new int[]{1, 1, 1};
+            subset = array.sectionNoReduce(origin, shape, stride);
         } else {
-            throw new IOException("variable rank invalid: " + variable.getShortName());
+            throw new IOException("variable rank invalid");
         }
 
-        // remove the time dimension of length 1 tb 2021-01-14
-        subset = subset.reduce(0);
-
         return subset;
     }
 
@@ -182,7 +181,7 @@ void compute(Configuration config, NetcdfFile reader, NetcdfFileWriter writer) t
         final SatelliteFieldsConfiguration satFieldsConfig = config.getSatelliteFields();
         final int numLayers = satFieldsConfig.get_z_dim();
         final Era5Archive era5Archive = new Era5Archive(config.getNWPAuxDir());
-        final VariableCache variableCache = new VariableCache(era5Archive, 52); // 4 * 13 variables tb 2020-11-25
+        final VariableCache variableCache = new VariableCache(era5Archive, 156); // 4 * 13 variables tb 2020-11-25
 
         try {
             // open input time variable
@@ -240,8 +239,8 @@ void compute(Configuration config, NetcdfFile reader, NetcdfFileWriter writer) t
                 //     - store to target raster
                 final Set<String> variableKeys = variables.keySet();
                 for (final String variableKey : variableKeys) {
-                    final Variable variable = variableCache.get(variableKey, era5Time);
-                    final Array subset = readSubset(numLayers, layerRegion, variable);
+                    VariableCache.CacheEntry cacheEntry = variableCache.get(variableKey, era5Time);
+                    final Array subset = readSubset(numLayers, layerRegion, cacheEntry);
                     final Index subsetIndex = subset.getIndex();
 
                     final Array targetArray = targetArrays.get(variableKey);

post-processing-tool/src/main/java/com/bc/fiduceo/post/plugin/era5/VariableCache.java

@@ -1,6 +1,7 @@
 package com.bc.fiduceo.post.plugin.era5;
 
 import org.esa.snap.core.util.io.FileUtils;
+import ucar.ma2.Array;
 import ucar.nc2.NetcdfFile;
 import ucar.nc2.Variable;
 
@@ -14,7 +15,7 @@
 class VariableCache {
 
     private final Era5Archive archive;
-    private final HashMap<String, CacheEntry> cache;
+    private final HashMap<String, CacheContainer> cache;
     private final int cacheSize;
 
 
@@ -24,13 +25,18 @@ class VariableCache {
         cache = new HashMap<>();
     }
 
-    Variable get(String variableKey, int era5TimeStamp) throws IOException {
+    CacheEntry get(String variableKey, int era5TimeStamp) throws IOException {
+        final CacheContainer cacheContainer = getCacheContainer(variableKey, era5TimeStamp);
+        return cacheContainer.cacheEntry;
+    }
+
+    private CacheContainer getCacheContainer(String variableKey, int era5TimeStamp) throws IOException {
         final String filePath = archive.get(variableKey, era5TimeStamp);
         final String variableName = getVariableName(variableKey);
         final String key = FileUtils.getFilenameWithoutExtension(new File(filePath));
 
-        CacheEntry cacheEntry = cache.get(key);
-        if (cacheEntry == null) {
+        CacheContainer cacheContainer = cache.get(key);
+        if (cacheContainer == null) {
             final NetcdfFile netcdfFile = NetcdfFile.open(filePath);
             final Variable variable = netcdfFile.findVariable(variableName);
             if (variable == null) {
@@ -39,21 +45,21 @@ Variable get(String variableKey, int era5TimeStamp) throws IOException {
             if (cache.size() == cacheSize) {
                 removeOldest();
             }
-            cacheEntry = new CacheEntry(variable, netcdfFile, System.currentTimeMillis());
-            cache.put(key, cacheEntry);
+            final Array array = variable.read().reduce();
+
+            cacheContainer = new CacheContainer(variable, netcdfFile, array, System.currentTimeMillis());
+            cache.put(key, cacheContainer);
         }
 
-        cacheEntry.lastAccess = System.currentTimeMillis();
-        return cacheEntry.variable;
+        cacheContainer.lastAccess = System.currentTimeMillis();
+        return cacheContainer;
     }
 
-
-
     void close() throws IOException {
-        final Collection<CacheEntry> cacheEntries = cache.values();
-        for (CacheEntry cacheEntry : cacheEntries) {
-            cacheEntry.netcdfFile.close();
-            cacheEntry.netcdfFile = null;
+        final Collection<CacheContainer> cacheEntries = cache.values();
+        for (CacheContainer cacheContainer : cacheEntries) {
+            cacheContainer.netcdfFile.close();
+            cacheContainer.netcdfFile = null;
         }
 
         cache.clear();
@@ -67,31 +73,43 @@ private String getVariableName(String variableKey) {
     private void removeOldest() throws IOException {
         long minTime = Long.MAX_VALUE;
         String toRemove = null;
-        CacheEntry entryToRemove = null;
-        final Set<Map.Entry<String, CacheEntry>> cacheEntries = cache.entrySet();
-        for (Map.Entry<String, CacheEntry> cacheMapEntry : cacheEntries) {
-            final CacheEntry cacheEntry = cacheMapEntry.getValue();
-            if (cacheEntry.lastAccess < minTime) {
-                minTime = cacheEntry.lastAccess;
+        CacheContainer entryToRemove = null;
+        final Set<Map.Entry<String, CacheContainer>> cacheEntries = cache.entrySet();
+        for (Map.Entry<String, CacheContainer> cacheMapEntry : cacheEntries) {
+            final CacheContainer cacheContainer = cacheMapEntry.getValue();
+            if (cacheContainer.lastAccess < minTime) {
+                minTime = cacheContainer.lastAccess;
                 toRemove = cacheMapEntry.getKey();
-                entryToRemove = cacheEntry;
+                entryToRemove = cacheContainer;
             }
         }
 
         if (entryToRemove != null) {
+            entryToRemove.cacheEntry = null;
             entryToRemove.netcdfFile.close();
 
             cache.remove(toRemove);
         }
     }
 
-    private class CacheEntry {
-        Variable variable;
+    static class CacheEntry {
+
+        final Variable variable;
+        final Array array;
+
+        CacheEntry(Variable variable, Array array) {
+            this.variable = variable;
+            this.array = array;
+        }
+    }
+
+    private static class CacheContainer {
+        CacheEntry cacheEntry;
         NetcdfFile netcdfFile;
         long lastAccess;
 
-        CacheEntry(Variable variable, NetcdfFile netcdfFile, long lastAccess) {
-            this.variable = variable;
+        CacheContainer(Variable variable, NetcdfFile netcdfFile, Array array, long lastAccess) {
+            this.cacheEntry = new CacheEntry(variable, array);
             this.netcdfFile = netcdfFile;
             this.lastAccess = lastAccess;
         }