Implemented InMemoryModelStore for replicas

java/com.sap.sailing.domain/src/com/sap/sailing/domain/maneuverdetection/impl/WindEstimationData.java

@@ -1,9 +0,0 @@
-package com.sap.sailing.domain.maneuverdetection.impl;
-
-public class WindEstimationData {
-
-    public WindEstimationData() {
-        // TODO Auto-generated constructor stub
-    }
-
-}

java/com.sap.sailing.windestimation.test/src/com/sap/sailing/windestimation/integration/IncrementalMstHmmWindEstimationForTrackedRaceTest.java

@@ -81,7 +81,7 @@ public class IncrementalMstHmmWindEstimationForTrackedRaceTest extends OnlineTra
             throws MalformedURLException, URISyntaxException, ModelPersistenceException {
         dateFormat = new SimpleDateFormat("MM/dd/yyyy-HH:mm:ss");
         dateFormat.setTimeZone(TimeZone.getTimeZone("GMT+2")); // will result in CEST
-        windEstimationFactoryService = new WindEstimationFactoryServiceImpl();
+        windEstimationFactoryService = new WindEstimationFactoryServiceImpl(false);
         modelStore = new ClassPathReadOnlyModelStore("trained_wind_estimation_models", getClass().getClassLoader(),
                 modelFilesNames);
         windEstimationFactoryService.importAllModelsFromModelStore(modelStore);

java/com.sap.sailing.windestimation/src/com/sap/sailing/windestimation/integration/Activator.java

@@ -30,11 +30,14 @@ public class Activator implements BundleActivator {
     private static final String WIND_ESTIMATION_MODEL_DATA_SOURCE_FOLDER_PROPERTY_NAME = "windestimation.source.folder";
 
     private final Set<ServiceRegistration<?>> registrations = new HashSet<>();
-    private final WindEstimationFactoryServiceImpl service = new WindEstimationFactoryServiceImpl();
+    private WindEstimationFactoryServiceImpl service;
 
     @Override
     public void start(BundleContext context) throws Exception {
         logger.info("Registering WindEstimationFactoryService");
+        //FIXME determine whether this is a replica or not
+        boolean replica = false;
+        service = new WindEstimationFactoryServiceImpl(replica);
         final String windEstimationModelDataSourceURL = System
                 .getProperty(WIND_ESTIMATION_MODEL_DATA_SOURCE_URL_PROPERTY_NAME);
         final String windEstimationModelDataSourceFolder = System

java/com.sap.sailing.windestimation/src/com/sap/sailing/windestimation/integration/WindEstimationFactoryServiceImpl.java

@@ -17,6 +17,7 @@ import com.sap.sailing.windestimation.model.classifier.maneuver.ManeuverClassifi
 import com.sap.sailing.windestimation.model.classifier.maneuver.ManeuverFeatures;
 import com.sap.sailing.windestimation.model.exception.ModelPersistenceException;
 import com.sap.sailing.windestimation.model.regressor.twdtransition.GaussianBasedTwdTransitionDistributionCache;
+import com.sap.sailing.windestimation.model.store.InMemoryModelStore;
 import com.sap.sailing.windestimation.model.store.ModelStore;
 import com.sap.sailing.windestimation.model.store.MongoDbModelStore;
 import com.sap.sailing.windestimation.model.store.PersistenceContextType;
@@ -32,21 +33,27 @@ public class WindEstimationFactoryServiceImpl
     private static final boolean ENABLE_POLARS_INFORMATION = true;
     private static final long PRESERVE_LOADED_MODELS_MILLIS = Long.MAX_VALUE;
 
-    public static final ModelStore MODEL_STORE = new MongoDbModelStore(MongoDBService.INSTANCE.getDB());
     private static final ManeuverFeatures MAX_MANEUVER_FEATURES = new ManeuverFeatures(ENABLE_POLARS_INFORMATION,
             ENABLE_SCALED_SPEED, ENABLE_MARKS_INFORMATION);
     private static final PersistenceContextType[] relevantContextTypes = new PersistenceContextType[] {
             PersistenceContextType.MANEUVER_CLASSIFIER, PersistenceContextType.DURATION_BASED_TWD_DELTA_STD_REGRESSOR,
             PersistenceContextType.DISTANCE_BASED_TWD_DELTA_STD_REGRESSOR };
 
-    protected final ManeuverClassifiersCache maneuverClassifiersCache = new ManeuverClassifiersCache(MODEL_STORE,
-            PRELOAD_ALL_MODELS, PRESERVE_LOADED_MODELS_MILLIS, MAX_MANEUVER_FEATURES);
-    protected final GaussianBasedTwdTransitionDistributionCache gaussianBasedTwdTransitionDistributionCache = new GaussianBasedTwdTransitionDistributionCache(
-            MODEL_STORE, PRELOAD_ALL_MODELS, PRESERVE_LOADED_MODELS_MILLIS);
+    public final ModelStore MODEL_STORE;
+    protected final ManeuverClassifiersCache maneuverClassifiersCache;
+    protected final GaussianBasedTwdTransitionDistributionCache gaussianBasedTwdTransitionDistributionCache;
     private final List<WindEstimationModelsChangedListener> modelsChangedListeners = new ArrayList<>();
     private boolean modelsReady = false;
     private boolean shutdown = false;
 
+    public WindEstimationFactoryServiceImpl(boolean replica) {
+        MODEL_STORE = replica ? new InMemoryModelStore() : new MongoDbModelStore(MongoDBService.INSTANCE.getDB());
+        maneuverClassifiersCache = new ManeuverClassifiersCache(MODEL_STORE, PRELOAD_ALL_MODELS,
+                PRESERVE_LOADED_MODELS_MILLIS, MAX_MANEUVER_FEATURES);
+        gaussianBasedTwdTransitionDistributionCache = new GaussianBasedTwdTransitionDistributionCache(MODEL_STORE,
+                PRELOAD_ALL_MODELS, PRESERVE_LOADED_MODELS_MILLIS);
+    }
+
     @Override
     public IncrementalWindEstimationTrack createIncrementalWindEstimationTrack(TrackedRace trackedRace) {
         IncrementalWindEstimationTrack windEstimation = new IncrementalMstHmmWindEstimationForTrackedRace(trackedRace,
@@ -68,7 +75,6 @@ public class WindEstimationFactoryServiceImpl
         for (PersistenceContextType contextType : relevantContextTypes) {
             Map<String, byte[]> exportedModels = (Map<String, byte[]>) is.readObject();
             MODEL_STORE.importPersistedModels(exportedModels, contextType);
-
         }
         clearState();
     }
@@ -108,6 +114,10 @@ public class WindEstimationFactoryServiceImpl
         }
     }
 
+    /**
+     * Deletes all persisted models in {@link #MODEL_STORE} and clears/reloads all caches with machine learning models.
+     */
+    @Override
     public void clearReplicaState() throws MalformedURLException, IOException, InterruptedException {
         for (PersistenceContextType contextType : relevantContextTypes) {
             MODEL_STORE.deleteAll(contextType);
@@ -115,6 +125,10 @@ public class WindEstimationFactoryServiceImpl
         clearState();
     }
 
+    /**
+     * Clears/Reloads all caches with machine learning models. Notifies ready state change listeners about the new ready
+     * state of this wind estimation instance.
+     */
     @Override
     public void clearState() {
         maneuverClassifiersCache.clearCache();
@@ -130,6 +144,9 @@ public class WindEstimationFactoryServiceImpl
         notifyModelsChangedListeners(modelsChangedListeners, modelsReady);
     }
 
+    /**
+     * Imports all the models which are available in the provided model store.
+     */
     public void importAllModelsFromModelStore(ModelStore modelStore) throws ModelPersistenceException {
         for (PersistenceContextType contextType : relevantContextTypes) {
             Map<String, byte[]> exportedModels = modelStore.exportAllPersistedModels(contextType);
@@ -139,6 +156,10 @@ public class WindEstimationFactoryServiceImpl
         clearState();
     }
 
+    /**
+     * Shuts down the wind estimation by calling all ready state change listeners with {@code modelReadyState=false} and
+     * removing all the attached listeners afterwards. This instance will never be ready again for wind estimation.
+     */
     public void shutdown() {
         List<WindEstimationModelsChangedListener> modelsChangedListeners;
         synchronized (this) {

java/com.sap.sailing.windestimation/src/com/sap/sailing/windestimation/model/store/InMemoryModelStore.java

@@ -0,0 +1,120 @@
+package com.sap.sailing.windestimation.model.store;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.sap.sailing.windestimation.model.ContextSpecificModelMetadata;
+import com.sap.sailing.windestimation.model.TrainableModel;
+import com.sap.sailing.windestimation.model.exception.ModelLoadingException;
+import com.sap.sailing.windestimation.model.exception.ModelNotFoundException;
+import com.sap.sailing.windestimation.model.exception.ModelPersistenceException;
+
+public class InMemoryModelStore extends AbstractModelStore {
+
+    private final Map<String, byte[]> serializedModels = new ConcurrentHashMap<>();
+
+    @Override
+    public <InstanceType, T extends ContextSpecificModelMetadata<InstanceType>, ModelType extends TrainableModel<InstanceType, T>> ModelType loadPersistedState(
+            ModelType newModel) throws ModelPersistenceException {
+        PersistenceSupport persistenceSupport = checkAndGetPersistenceSupport(newModel);
+        byte[] serializedModel = serializedModels.get(getFilename(newModel));
+        if (serializedModel != null) {
+            try (InputStream input = new ByteArrayInputStream(serializedModel)) {
+                @SuppressWarnings("unchecked")
+                ModelType loadedModel = (ModelType) persistenceSupport.loadFromStream(input);
+                if (!newModel.getContextSpecificModelMetadata().equals(loadedModel.getContextSpecificModelMetadata())) {
+                    throw new ModelPersistenceException(
+                            "The configuration of the loaded model is: " + loadedModel.getContextSpecificModelMetadata()
+                                    + ". \nExpected: " + newModel.getContextSpecificModelMetadata());
+                }
+                return loadedModel;
+            } catch (IOException e) {
+                throw new ModelPersistenceException(e);
+            }
+        }
+        throw new ModelNotFoundException(newModel.getContextSpecificModelMetadata());
+    }
+
+    @Override
+    public <T extends PersistableModel<?, ?>> void persistState(T trainedModel) throws ModelPersistenceException {
+        PersistenceSupport persistenceSupport = checkAndGetPersistenceSupport(trainedModel);
+        try (ByteArrayOutputStream output = new ByteArrayOutputStream()) {
+            persistenceSupport.saveToStream(trainedModel, output);
+            byte[] serializedModel = output.toByteArray();
+            serializedModels.put(getFilename(trainedModel), serializedModel);
+        } catch (IOException e) {
+            throw new ModelPersistenceException(e);
+        }
+    }
+
+    public <T extends PersistableModel<?, ?>> void delete(T newModel) throws ModelPersistenceException {
+        checkAndGetPersistenceSupport(newModel);
+        serializedModels.remove(getFilename(newModel));
+    }
+
+    @Override
+    public void deleteAll(PersistenceContextType contextType) {
+        for (Iterator<String> iterator = serializedModels.keySet().iterator(); iterator.hasNext();) {
+            String filename = iterator.next();
+            if (filename.endsWith(FILE_EXT) && filename.startsWith(getContextPrefix(contextType))) {
+                iterator.remove();
+            }
+        }
+    }
+
+    @Override
+    public Map<String, byte[]> exportAllPersistedModels(PersistenceContextType contextType) {
+        Map<String, byte[]> exportedModels = new HashMap<>();
+        for (Entry<String, byte[]> entry : serializedModels.entrySet()) {
+            String fileName = entry.getKey();
+            if (isFileBelongingToContextType(fileName, contextType)) {
+                exportedModels.put(fileName, entry.getValue());
+            }
+        }
+        return exportedModels;
+    }
+
+    @Override
+    public void importPersistedModels(Map<String, byte[]> exportedPersistedModels, PersistenceContextType contextType) {
+        for (Entry<String, byte[]> entry : exportedPersistedModels.entrySet()) {
+            String fileName = entry.getKey();
+            byte[] exportedModel = entry.getValue();
+            serializedModels.put(fileName, exportedModel);
+        }
+    }
+
+    @Override
+    public List<PersistableModel<?, ?>> loadAllPersistedModels(PersistenceContextType contextType) {
+        List<PersistableModel<?, ?>> loadedModels = new ArrayList<>();
+        for (Entry<String, byte[]> entry : serializedModels.entrySet()) {
+            String fileName = entry.getKey();
+            if (isFileBelongingToContextType(fileName, contextType)) {
+                PersistenceSupport persistenceSupport = getPersistenceSupportFromFilename(fileName);
+                if (persistenceSupport == null) {
+                    throw new ModelLoadingException(
+                            "Persistence support could not be determined due to invalid filename pattern: \"" + fileName
+                                    + "\"");
+                }
+                PersistableModel<?, ?> loadedModel;
+                try (InputStream input = new ByteArrayInputStream(entry.getValue())) {
+                    loadedModel = persistenceSupport.loadFromStream(input);
+                } catch (IOException e) {
+                    throw new ModelLoadingException(
+                            "Could not read model \"" + fileName + "\" from its serialized in-memory state", e);
+                }
+                loadedModels.add(loadedModel);
+            }
+        }
+        return loadedModels;
+    }
+
+}