Merge branch 'bug4339' into bug4693

2026-09-20 04:35:32 +00:00 · 2018-07-31 17:27:31 +02:00
parent f542068d66 15386f1c03
commit 01483a973c
42 changed files with 354 additions and 123 deletions
@@ -24,21 +24,24 @@ public class BravoFixRetrievalProcessor extends AbstractRetrievalProcessor<HasTr
    @Override
    protected Iterable<HasBravoFixContext> retrieveData(HasTrackedLegOfCompetitorContext element) {
        Collection<HasBravoFixContext> bravoFixesWithContext = new ArrayList<>();
-        BravoFixTrack<Competitor> bravoFixTrack = element.getTrackedLegContext().getTrackedRaceContext().getTrackedRace().getSensorTrack(element.getCompetitor(), BravoFixTrack.TRACK_NAME);
-        if (bravoFixTrack != null) {
-            bravoFixTrack.lockForRead();
-            try {
-                TrackedLegOfCompetitor trackedLegOfCompetitor = element.getTrackedLegOfCompetitor();
-                if (trackedLegOfCompetitor.getStartTime() != null && trackedLegOfCompetitor.getFinishTime() != null) {
+        TrackedLegOfCompetitor trackedLegOfCompetitor = element.getTrackedLegOfCompetitor();
+        if (trackedLegOfCompetitor.getStartTime() != null && trackedLegOfCompetitor.getFinishTime() != null) {
+            BravoFixTrack<Competitor> bravoFixTrack = element.getTrackedLegContext().getTrackedRaceContext().getTrackedRace().getSensorTrack(element.getCompetitor(), BravoFixTrack.TRACK_NAME);
+            if (bravoFixTrack != null) {
+                bravoFixTrack.lockForRead();
+                try {
                    for (BravoFix bravoFix : bravoFixTrack.getFixes(trackedLegOfCompetitor.getStartTime(), true, trackedLegOfCompetitor.getFinishTime(), true)) {
+                        if (isAborted()) {
+                            break;
+                        }
                        BravoFixWithContext gpsFixWithContext = new BravoFixWithContext(
                                new TrackedLegOfCompetitorWithSpecificTimePointWithContext(
                                        element.getTrackedLegContext(), element.getTrackedLegOfCompetitor(), bravoFix.getTimePoint()), bravoFix);
                        bravoFixesWithContext.add(gpsFixWithContext);
                    }
+                } finally {
+                    bravoFixTrack.unlockAfterRead();
                }
-            } finally {
-                bravoFixTrack.unlockAfterRead();
            }
        }
        return bravoFixesWithContext;
@@ -1,7 +1,7 @@
 package com.sap.sailing.datamining.impl.components;

-import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.concurrent.ExecutorService;

 import com.sap.sailing.datamining.data.HasBravoFixTrackContext;
@@ -28,13 +28,9 @@ public class BravoFixTrackRetrievalProcessor extends AbstractRetrievalProcessor<

    @Override
    protected Iterable<HasBravoFixTrackContext> retrieveData(HasRaceOfCompetitorContext element) {
-        Collection<HasBravoFixTrackContext> bravoTracksWithContext = new ArrayList<>();
        final TrackedRace trackedRace = element.getTrackedRaceContext().getTrackedRace();
        final BravoFixTrack<Competitor> bravoFixTrack = trackedRace.getSensorTrack(element.getCompetitor(), BravoFixTrack.TRACK_NAME);
-        if (bravoFixTrack != null) {
-            bravoTracksWithContext.add(new BravoFixTrackWithContext(element, bravoFixTrack));
-        }
-        return bravoTracksWithContext;
+        return bravoFixTrack == null ? Collections.emptySet() : Collections.singleton(new BravoFixTrackWithContext(element, bravoFixTrack));
    }

 }
@@ -24,7 +24,13 @@ public class CompetitorOfRaceInLeaderboardRetrievalProcessor extends
    protected Iterable<HasRaceResultOfCompetitorContext> retrieveData(HasLeaderboardContext element) {
        Collection<HasRaceResultOfCompetitorContext> raceResultsOfCompetitor = new ArrayList<>();
        for (RaceColumn raceColumn : element.getLeaderboard().getRaceColumns()) {
+            if (isAborted()) {
+                break;
+            }
            for (Competitor competitor : element.getLeaderboard().getCompetitors()) {
+                if (isAborted()) {
+                    break;
+                }
                HasRaceResultOfCompetitorContext raceResultOfCompetitorContext = new RaceResultOfCompetitorWithContext(element, raceColumn, competitor,
                		element.getLeaderboardGroupContext().getPolarDataService());
                raceResultsOfCompetitor.add(raceResultOfCompetitorContext);
@@ -49,14 +49,26 @@ public class CompleteManeuverCurveWithEstimationDataRetrievalProcessor extends
        ManeuverDetectorWithEstimationDataSupport maneuverDetector = new ManeuverDetectorWithEstimationDataSupportDecoratorImpl(
                new ManeuverDetectorImpl(trackedRace, competitor),
                element.getTrackedRaceContext().getLeaderboardContext().getLeaderboardGroupContext().getPolarDataService());
+        
        Iterable<Maneuver> maneuvers = trackedRace.getManeuvers(competitor, false);
+        if (isAborted()) {
+            return result;
+        }
+        
        Iterable<CompleteManeuverCurve> maneuverCurves = maneuverDetector.getCompleteManeuverCurves(maneuvers);
+        if (isAborted()) {
+            return result;
+        }
+        
        Iterable<CompleteManeuverCurveWithEstimationData> maneuversWithEstimationData = maneuverDetector
                .getCompleteManeuverCurvesWithEstimationData(maneuverCurves);
-
        CompleteManeuverCurveWithEstimationData previousManeuver = null;
        CompleteManeuverCurveWithEstimationData currentManeuver = null;
        for (CompleteManeuverCurveWithEstimationData nextManeuver : maneuversWithEstimationData) {
+            if (isAborted()) {
+                break;
+            }
+            
            if (currentManeuver != null) {
                CompleteManeuverCurveWithEstimationDataWithContext maneuverWithContext = new CompleteManeuverCurveWithEstimationDataWithContext(
                        element, currentManeuver, settings, previousManeuver, nextManeuver);
@@ -50,6 +50,9 @@ public class FoilingSegmentRetrievalProcessor extends AbstractRetrievalProcessor
                bravoFixTrack.lockForRead();
                try {
                    for (final BravoFix bravoFix : bravoFixTrack.getFixes(startOfRace, /* fromInclusive */ true, end, /* toInclusive */ false)) {
+                        if (isAborted()) {
+                            break;
+                        }
                        final boolean currentFixIsFoiling = 
                                (bravoFix.isFoiling(settings.getMinimumRideHeight()) &&
                                        (settings.getMinimumSpeedForFoiling() == null || settings.getMinimumSpeedForFoiling().compareTo(
@@ -30,6 +30,9 @@ public class GPSFixRetrievalProcessor extends AbstractRetrievalProcessor<HasTrac
            TrackedLegOfCompetitor trackedLegOfCompetitor = element.getTrackedLegOfCompetitor();
            if (trackedLegOfCompetitor.getStartTime() != null && trackedLegOfCompetitor.getFinishTime() != null) {
                for (GPSFixMoving gpsFix : competitorTrack.getFixes(trackedLegOfCompetitor.getStartTime(), true, trackedLegOfCompetitor.getFinishTime(), true)) {
+                    if (isAborted()) {
+                        break;
+                    }
                    HasGPSFixContext gpsFixWithContext = new GPSFixWithContext(new TrackedLegOfCompetitorWithSpecificTimePointWithContext(
                            element.getTrackedLegContext(), element.getTrackedLegOfCompetitor(), gpsFix.getTimePoint()), gpsFix);
                    gpsFixesWithContext.add(gpsFixWithContext);
@@ -1,11 +1,13 @@
 package com.sap.sailing.datamining.impl.components;

 import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.concurrent.ExecutorService;
-import java.util.stream.Collectors;

 import com.sap.sailing.datamining.data.HasLeaderboardGroupContext;
 import com.sap.sailing.datamining.impl.data.LeaderboardGroupWithContext;
+import com.sap.sailing.domain.leaderboard.LeaderboardGroup;
 import com.sap.sailing.domain.polars.PolarDataService;
 import com.sap.sailing.server.RacingEventService;
 import com.sap.sse.datamining.components.Processor;
@@ -20,12 +22,15 @@ public class LeaderboardGroupRetrievalProcessor extends AbstractRetrievalProcess

    @Override
    protected Iterable<HasLeaderboardGroupContext> retrieveData(RacingEventService element) {
-        final PolarDataService polarDataService = element.getPolarDataService();
-        return element.getLeaderboardGroups()
-                .values()
-                .stream()
-                .map(lg -> new LeaderboardGroupWithContext(lg, polarDataService))
-                .collect(Collectors.toSet());
+        Set<HasLeaderboardGroupContext> data = new HashSet<>();
+        PolarDataService polarDataService = element.getPolarDataService();
+        for (LeaderboardGroup leaderboardGroup : element.getLeaderboardGroups().values()) {
+            if (isAborted()) {
+                break;
+            }
+            data.add(new LeaderboardGroupWithContext(leaderboardGroup, polarDataService));
+        }
+        return data;
    }

 }
@@ -22,6 +22,9 @@ public class LeaderboardRetrievalProcessor extends AbstractRetrievalProcessor<Ha
    protected Iterable<HasLeaderboardContext> retrieveData(HasLeaderboardGroupContext element) {
        Collection<HasLeaderboardContext> leaderboardsWithContext = new ArrayList<>();
        for (Leaderboard leaderboard : element.getLeaderboardGroup().getLeaderboards()) {
+            if (isAborted()) {
+                break;
+            }
            leaderboardsWithContext.add(new LeaderboardWithContext(leaderboard, element));
        }
        return leaderboardsWithContext;
@@ -48,6 +48,10 @@ public class ManeuverRetrievalProcessor
            Maneuver previousManeuver = null;
            Maneuver currentManeuver = null;
            for (Maneuver nextManeuver : maneuvers) {
+                if (isAborted()) {
+                    break;
+                }
+                
                if (currentManeuver != null) {
                    ManeuverWithContext maneuverWithContext = new ManeuverWithContext(new TrackedLegOfCompetitorWithSpecificTimePointWithContext(
                            element.getTrackedLegContext(), element.getTrackedLegOfCompetitor(), currentManeuver.getTimePoint()), currentManeuver,
@@ -87,6 +87,10 @@ public class ManeuverSpeedDetailsRetrievalProcessor

        double directionChangeForAnalysisSignum = Math.signum(directionChangeInDegreesForAnalysis);
        for (SpeedWithBearingStep bearingStep : maneuverBearingSteps) {
+            if (isAborted()) {
+                break;
+            }
+            
            currentDirectionChangeSumInDegrees += bearingStep.getCourseChangeInDegrees();
            if (previousRoundedTWA != -1
                    && Math.signum(currentDirectionChangeSumInDegrees) != directionChangeForAnalysisSignum
@@ -123,6 +127,9 @@ public class ManeuverSpeedDetailsRetrievalProcessor
                        for (int step = 1, fillingTWA = twaIterationFunction
                                .apply(previousRoundedTWA); fillingTWA != roundedTWA; fillingTWA = twaIterationFunction
                                        .apply(fillingTWA), ++step) {
+                            if (isAborted()) {
+                                break;
+                            }
                            if (speedPerTWA[fillingTWA] == 0) {
                                speedPerTWA[fillingTWA] = previousSpeed
                                        + diffWithPreviousSpeed * step / diffWithPreviousTWA;
@@ -29,6 +29,9 @@ public class MarkPassingRetrievalProcessor extends AbstractRetrievalProcessor<Ha
            try {
                Iterable<Maneuver> maneuvers = element.getTrackedLegOfCompetitor().getManeuvers(finishTime, false);
                for (Maneuver maneuver : maneuvers) {
+                    if (isAborted()) {
+                        break;
+                    }
                    if (maneuver.isMarkPassing()) {
                        maneuversWithContext.add(new MarkPassingWithContext(new TrackedLegOfCompetitorWithSpecificTimePointWithContext(
                                element.getTrackedLegContext(), element.getTrackedLegOfCompetitor(), maneuver.getTimePoint()), maneuver));
@@ -22,6 +22,9 @@ public class RaceOfCompetitorRetrievalProcessor extends AbstractRetrievalProcess
    protected Iterable<HasRaceOfCompetitorContext> retrieveData(HasTrackedRaceContext element) {
        Collection<HasRaceOfCompetitorContext> raceOfCompetitorsWithContext = new ArrayList<>();
        for (Competitor competitor : element.getTrackedRace().getRace().getCompetitors()) {
+            if (isAborted()) {
+                break;
+            }
            HasRaceOfCompetitorContext raceOfCompetitorWithContext = new RaceOfCompetitorWithContext(element, competitor);
            raceOfCompetitorsWithContext.add(raceOfCompetitorWithContext);
        }
@@ -22,6 +22,9 @@ public class TrackedLegOfCompetitorRetrievalProcessor extends AbstractRetrievalP
    protected Iterable<HasTrackedLegOfCompetitorContext> retrieveData(HasTrackedLegContext element) {
        Collection<HasTrackedLegOfCompetitorContext> trackedLegOfCompetitorsWithContext = new ArrayList<>();
        for (Competitor competitor : element.getTrackedRaceContext().getTrackedRace().getRace().getCompetitors()) {
+            if (isAborted()) {
+                break;
+            }
            HasTrackedLegOfCompetitorContext trackedLegOfCompetitorWithContext = new TrackedLegOfCompetitorWithContext(element, element.getTrackedLeg().getTrackedLeg(competitor));
            trackedLegOfCompetitorsWithContext.add(trackedLegOfCompetitorWithContext);
        }
@@ -23,6 +23,9 @@ public class TrackedLegRetrievalProcessor extends AbstractRetrievalProcessor<Has
        Collection<HasTrackedLegContext> trackedLegsWithContext = new ArrayList<>();
        int legNumber = 1;
        for (TrackedLeg trackedLeg : element.getTrackedRace().getTrackedLegs()) {
+            if (isAborted()) {
+                break;
+            }
            HasTrackedLegContext trackedLegWithContext = new TrackedLegWithContext(element, trackedLeg, legNumber);
            trackedLegsWithContext.add(trackedLegWithContext);
            legNumber++;
@@ -25,7 +25,13 @@ public class TrackedRaceRetrievalProcessor extends AbstractRetrievalProcessor<Ha
    protected Iterable<HasTrackedRaceContext> retrieveData(HasLeaderboardContext element) {
        Collection<HasTrackedRaceContext> trackedRacesWithContext = new ArrayList<>();
        for (RaceColumn raceColumn : element.getLeaderboard().getRaceColumns()) {
+            if (isAborted()) {
+                break;
+            }
            for (Fleet fleet : raceColumn.getFleets()) {
+                if (isAborted()) {
+                    break;
+                }
                TrackedRace trackedRace = raceColumn.getTrackedRace(fleet);
                if (trackedRace != null) {
                    Regatta regatta = trackedRace.getTrackedRegatta().getRegatta();
@@ -33,6 +33,9 @@ public class WindFixRetrievalProcessor extends AbstractRetrievalProcessor<HasWin
        windTrack.lockForRead();
        try {
            for (final Wind wind : windTrack.getFixes()) {
+                if (isAborted()) {
+                    break;
+                }
                windFixesWithContext.add(new WindFixWithContext(element.getTrackedRaceContext(), wind, element.getWindSourceType()));
            }
        } finally {
@@ -31,6 +31,9 @@ public class WindTrackRetrievalProcessor extends AbstractRetrievalProcessor<HasT
        Collection<HasWindTrackContext> windTracksWithContext = new ArrayList<>();
        final TrackedRace trackedRace = element.getTrackedRace();
        for (final WindSource windSource : trackedRace.getWindSources()) {
+            if (isAborted()) {
+                break;
+            }
            if (!trackedRace.getWindSourcesToExclude().contains(windSource)) {
                final WindTrack windTrack = trackedRace.getOrCreateWindTrack(windSource);
                windTracksWithContext.add(new WindTrackWithContext(element, windTrack, windSource));
@@ -64,6 +64,9 @@ public abstract class AbstractParallelAverageAggregationProcessor<T extends Comp
        Map<GroupKey, T> minAggregation = minAggregationProcessor.getResult();
        Map<GroupKey, T> maxAggregation = maxAggregationProcessor.getResult();
        for (Entry<GroupKey, T> sumAggregationEntry : sumAggregation.entrySet()) {
+            if (isAborted()) {
+                break;
+            }
            GroupKey key = sumAggregationEntry.getKey();
            result.put(key, new AverageWithStatsImpl<>(
                    /* average */ divide(sumAggregationEntry.getValue(), elementAmountPerKey.get(key).longValue()),
@@ -19,8 +19,6 @@ public abstract class AbstractParallelSumAggregationProcessor<T> extends
        super(executor, resultReceivers, "Sum");
        results = new HashMap<>();
    }
-    
-    protected abstract T add(T t1, T t2);

    @Override
    protected void handleElement(GroupedDataEntry<T> element) {
@@ -31,6 +29,8 @@ public abstract class AbstractParallelSumAggregationProcessor<T> extends
            results.put(key, add(results.get(key), element.getDataEntry()));
        }
    }
+    
+    protected abstract T add(T t1, T t2);

    @Override
    protected Map<GroupKey, T> getResult() {
@@ -47,6 +47,9 @@ public class ParallelBearingAverageDegreesAggregationProcessor
    protected Map<GroupKey, Double> aggregateResult() {
        Map<GroupKey, Double> result = new HashMap<>();
        for (Entry<GroupKey, BearingCluster> clusterEntry : results.entrySet()) {
+            if (isAborted()) {
+                break;
+            }
            GroupKey key = clusterEntry.getKey();
            result.put(key, clusterEntry.getValue().getAverage().getDegrees());
        }
@@ -50,6 +50,9 @@ public class ParallelDistanceMedianAggregationProcessor
    protected Map<GroupKey, Distance> aggregateResult() {
        Map<GroupKey, Distance> result = new HashMap<>();
        for (Entry<GroupKey, List<Distance>> groupedValuesEntry : groupedValues.entrySet()) {
+            if (isAborted()) {
+                break;
+            }
            result.put(groupedValuesEntry.getKey(), getMedianOf(groupedValuesEntry.getValue()));
        }
        return result;
@@ -5,9 +5,11 @@ import java.util.Collection;
 import java.util.Comparator;
 import java.util.Date;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Set;
 import java.util.SortedMap;
 import java.util.TreeMap;

@@ -132,6 +134,8 @@ public class EditMarkPositionPanel extends AbstractRaceChart<AbstractSettings> i
    private final RaceIdentifierToLeaderboardRaceColumnAndFleetMapper raceIdentifierToLeaderboardRaceColumnAndFleetMapper;
    protected boolean nonTrackingWarningWasDisplayed;

+    private final Set<MarkDTO> marksCurrentlyRequestedViaRemoteCall = new HashSet<>();
+
    public EditMarkPositionPanel(Component<?> parent, ComponentContext<?> context, final RaceMap raceMap,
            final SingleRaceLeaderboardPanel leaderboardPanel,
            RegattaAndRaceIdentifier selectedRaceIdentifier, String leaderboardName, final StringMessages stringMessages,
@@ -735,10 +739,7 @@ public class EditMarkPositionPanel extends AbstractRaceChart<AbstractSettings> i
            }
            raceMap.unregisterAllCourseMarkInfoWindowClickHandlers();
        } else {
-            if (currentFixPositionChooser != null) {
-                currentFixPositionChooser.cancel();
-                currentFixPositionChooser = null;
-            }
+            cancelFixPositionChooserAndNotification();
            marksPanel.deselectMark();
            selectedMark = null;
            if (sideBySideComponentViewer != null) {
@@ -754,6 +755,17 @@ public class EditMarkPositionPanel extends AbstractRaceChart<AbstractSettings> i
        super.setVisible(visible);
    }

+    private void cancelFixPositionChooserAndNotification() {
+        if (currentFixPositionChooser != null) {
+            currentFixPositionChooser.cancel();
+            currentFixPositionChooser = null;
+            if (notificationTimer.isRunning()) {
+                notificationTimer.run();
+                notificationTimer.cancel();
+            }
+        }
+    }
+
    private void checkIfTracking(Runnable continuation) {
        if (nonTrackingWarningWasDisplayed) {
            continuation.run();
@@ -859,11 +871,10 @@ public class EditMarkPositionPanel extends AbstractRaceChart<AbstractSettings> i
    }
    
    private void selectMark(MarkDTO mark) {
-        selectedMark = mark;
-        if (currentFixPositionChooser != null) {
-            currentFixPositionChooser.cancel();
-            currentFixPositionChooser = null;
+        if (selectedMark != mark) {
+            cancelFixPositionChooserAndNotification();
        }
+        selectedMark = mark;
        if (selectedMark != null) {
            if (marksFromToTimes.get(selectedMark) != null) {
                // For some reason the time slider does not change with this method only if you comment out line 430 and 432 in TimePanel it works
@@ -905,24 +916,43 @@ public class EditMarkPositionPanel extends AbstractRaceChart<AbstractSettings> i
    @Override
    public void onSelectionChange(SelectionChangeEvent event) {
        final MarkDTO mark = marksPanel.getSelectedMark();
+        retrieveAndSelectMarkIfNecessary(mark, null);
+    }
+
+    protected void retrieveAndSelectMarkIfNecessary(final MarkDTO mark, final Runnable callback) {
        if (mark != null && (marks.get(mark) == null || marks.get(mark).isEmpty())) {
-            if (mark != null) {
-                markPositionService.getMarkTrack(raceIdentifierToLeaderboardRaceColumnAndFleetMapper.getLeaderboardNameAndRaceColumnNameAndFleetName(selectedRaceIdentifier), 
+            
+            if (marksCurrentlyRequestedViaRemoteCall.add(mark)) {
+                markPositionService.getMarkTrack(
+                        raceIdentifierToLeaderboardRaceColumnAndFleetMapper
+                                .getLeaderboardNameAndRaceColumnNameAndFleetName(selectedRaceIdentifier),
                        mark.getIdAsString(), new AsyncCallback<MarkTrackDTO>() {
                            @Override
                            public void onFailure(Throwable caught) {
-                                errorReporter.reportError(stringMessages.errorCommunicatingWithServer()+": "+caught.getMessage());
+                                marksCurrentlyRequestedViaRemoteCall.remove(mark);
+                                errorReporter.reportError(
+                                        stringMessages.errorCommunicatingWithServer() + ": " + caught.getMessage());
                            }
-    
+
                            @Override
                            public void onSuccess(MarkTrackDTO result) {
+                                marksCurrentlyRequestedViaRemoteCall.remove(mark);
                                createMarkTrackUi(mark, result.getFixes());
                                selectMark(mark);
+                                if (callback != null) {
+                                    callback.run();
+                                }
                            }
                        });
            }
+            else {
+                // remote call of the same mark already in progress -> ignore this request
+            }
        } else {
            selectMark(mark);
+            if (callback != null) {
+                callback.run();
+            }
        }
    }
    
@@ -85,22 +85,27 @@ public class MarksPanel extends AbstractCompositeComponent<AbstractSettings> {
            @Override
            public void update(int index, final MarkDTO mark, String value) {
                final Date timePoint = parent.timer.getTime();
-                select(mark);
-                if (parent.hasFixAtTimePoint(mark, timePoint)) {
-                    parent.showNotification(stringMessages.pleaseSelectOtherTimepoint(), NotificationType.ERROR);
-                } else {
-                    parent.createFixPositionChooserToAddFixToMark(mark, new Callback<Position, Exception>() {
-                        @Override
-                        public void onFailure(Exception reason) {
-                            parent.resetCurrentFixPositionChooser();
+                parent.retrieveAndSelectMarkIfNecessary(mark, new Runnable() {
+
+                    @Override
+                    public void run() {
+                        if (parent.hasFixAtTimePoint(mark, timePoint)) {
+                            parent.showNotification(stringMessages.pleaseSelectOtherTimepoint(), NotificationType.ERROR);
+                        } else {
+                            parent.createFixPositionChooserToAddFixToMark(mark, new Callback<Position, Exception>() {
+                                @Override
+                                public void onFailure(final Exception reason) {
+                                    parent.resetCurrentFixPositionChooser();
+                                }
+                                @Override
+                                public void onSuccess(Position result) {
+                                    parent.addMarkFix(mark, timePoint, result);
+                                    parent.resetCurrentFixPositionChooser();
+                                }
+                            });
                        }
-                        @Override
-                        public void onSuccess(Position result) {
-                            parent.addMarkFix(mark, timePoint, result);
-                            parent.resetCurrentFixPositionChooser();
-                        }
-                    });
-                }
+                    }
+                });
            }
        });
        markTable.addColumn(addFixColumn);
@@ -1,9 +1,11 @@
 package com.sap.sailing.polars.datamining.components;

 import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.concurrent.ExecutorService;
-import java.util.stream.Collectors;

+import com.sap.sailing.domain.base.BoatClass;
 import com.sap.sailing.domain.polars.PolarDataService;
 import com.sap.sailing.polars.datamining.data.HasBackendPolarBoatClassContext;
 import com.sap.sailing.polars.datamining.data.impl.BoatClassWithBackendPolarContext;
@@ -20,11 +22,15 @@ public class BackendPolarsBoatClassRetrievalProcessor extends AbstractRetrievalP

    @Override
    protected Iterable<HasBackendPolarBoatClassContext> retrieveData(RacingEventService element) {
+        Set<HasBackendPolarBoatClassContext> data = new HashSet<>();
        PolarDataService polarDataService = element.getPolarDataService();
-        return polarDataService.getAllBoatClassesWithPolarSheetsAvailable()
-                .stream()
-                .map(bc -> new BoatClassWithBackendPolarContext(bc, polarDataService))
-                .collect(Collectors.toSet());
+        for (BoatClass boatClass : polarDataService.getAllBoatClassesWithPolarSheetsAvailable()) {
+            if (isAborted()) {
+                break;
+            }
+            data.add(new BoatClassWithBackendPolarContext(boatClass, polarDataService));
+        }
+        return data;
    }

 }
@@ -25,6 +25,9 @@ public class PolarCompetitorRetrievalProcessor extends AbstractRetrievalProcesso
        TrackedRace trackedRace = element.getTrackedRace();
        Set<HasCompetitorPolarContext> competitorWithContext = new HashSet<>();
        for (Competitor competitor : trackedRace.getRace().getCompetitors()) {
+            if (isAborted()) {
+                break;
+            }
            competitorWithContext.add(new CompetitorWithPolarContext(competitor, trackedRace, element.getLeg(), element));
        }
        return competitorWithContext;
@@ -25,6 +25,9 @@ public class PolarFleetRetrievalProcessor extends AbstractRetrievalProcessor<Has
        Set<HasFleetPolarContext> fleetWithContext = new HashSet<>();
        RaceColumn raceColumn = element.getRaceColumn();
        for (Fleet fleet : raceColumn.getFleets()) {
+            if (isAborted()) {
+                break;
+            }
            fleetWithContext.add(new FleetWithPolarContext(fleet, raceColumn, element));
        }
        return fleetWithContext;
@@ -59,6 +59,9 @@ public class PolarGPSFixRetrievalProcessor extends AbstractRetrievalProcessor<Ha
            try {
                Iterable<GPSFixMoving> fixes = track.getFixes(startTime, true, finishTime, false);
                for (GPSFixMoving fix : fixes) {
+                    if (isAborted()) {
+                        break;
+                    }
                    WindWithConfidence<Pair<Position, TimePoint>> wind = trackedRace.getWindWithConfidence(
                            fix.getPosition(),
                            fix.getTimePoint(),
@@ -1,9 +1,11 @@
 package com.sap.sailing.polars.datamining.components;

 import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.concurrent.ExecutorService;
-import java.util.stream.Collectors;

+import com.sap.sailing.domain.leaderboard.LeaderboardGroup;
 import com.sap.sailing.polars.datamining.data.HasLeaderboardGroupPolarContext;
 import com.sap.sailing.polars.datamining.data.impl.LeaderboardGroupWithPolarContext;
 import com.sap.sailing.server.RacingEventService;
@@ -19,11 +21,14 @@ public class PolarLeaderboardGroupRetrievalProcessor extends AbstractRetrievalPr

    @Override
    protected Iterable<HasLeaderboardGroupPolarContext> retrieveData(RacingEventService element) {
-        return element.getLeaderboardGroups()
-                .values()
-                .stream()
-                .map(lg -> new LeaderboardGroupWithPolarContext(lg))
-                .collect(Collectors.toSet());
+        Set<HasLeaderboardGroupPolarContext> data = new HashSet<>();
+        for (LeaderboardGroup leaderboardGroup : element.getLeaderboardGroups().values()) {
+            if (isAborted()) {
+                break;
+            }
+            data.add(new LeaderboardGroupWithPolarContext(leaderboardGroup));
+        }
+        return data;
    }

 }
@@ -23,6 +23,9 @@ public class PolarLeaderboardRetrievalProcessor extends AbstractRetrievalProcess
    protected Iterable<HasLeaderboardPolarContext> retrieveData(HasLeaderboardGroupPolarContext element) {
        Set<HasLeaderboardPolarContext> leaderboardsWithContext = new HashSet<>();
        for (Leaderboard leaderboard : element.getLeaderboardGroup().getLeaderboards()) {
+            if (isAborted()) {
+                break;
+            }
            leaderboardsWithContext.add(new LeaderboardWithPolarContext(leaderboard, element));
        }
        return leaderboardsWithContext;
@@ -35,6 +35,9 @@ public class PolarLegRetrievalProcessor extends AbstractRetrievalProcessor<HasFl
        if (raceDefinition != null) {
            Course course = raceDefinition.getCourse();
            for (Leg leg : course.getLegs()) {
+                if (isAborted()) {
+                    break;
+                }
                legWithContext.add(new LegWithPolarContext(leg, trackedRace, element));
            }
        } 
@@ -25,6 +25,9 @@ public class PolarRaceColumnRetrievalProcessor extends AbstractRetrievalProcesso
        Set<HasRaceColumnPolarContext> raceColumnWithContext = new HashSet<>();
        Leaderboard leaderboard = element.getLeaderboard();
        for (RaceColumn raceColumn : leaderboard.getRaceColumns()) {
+            if (isAborted()) {
+                break;
+            }
            raceColumnWithContext.add(new RaceColumnWithPolarContext(raceColumn, element));
        }
        return raceColumnWithContext;
@@ -100,10 +100,17 @@ public class PolarBackendDataAggregationProcessor extends AbstractParallelGroupe
        } catch (NotEnoughDataHasBeenAddedException e) {
            hasDownwindAngleData = false;
        }
+        
        for (int angleInDeg = 0; angleInDeg < 360; angleInDeg++) {
+            if (isAborted()) {
+                break;
+            }
            int convertedAngle = angleInDeg  > 180 ? angleInDeg  - 360 : angleInDeg ;
            try {
                for (int x = 0; x < 30; x++) {
+                    if (isAborted()) {
+                        break;
+                    }
                    SpeedWithConfidence<Void> speed = polarDataService.getSpeed(boatClass, new KnotSpeedImpl(x), new DegreeBearingImpl(convertedAngle));
                    if (speed.getConfidence() > 0.1) {
                        hasDataForAngle[angleInDeg] = true;
@@ -127,6 +134,9 @@ public class PolarBackendDataAggregationProcessor extends AbstractParallelGroupe
    
    private void setArrayValuesForFunction(PolynomialFunction function, double[] yOverWindSpeed) {
        for (int x = 0; x < 30; x++) {
+            if (isAborted()) {
+                break;
+            }
            yOverWindSpeed[x] = function.value(x);
        }
    }
@@ -71,24 +71,31 @@ public class TestAbortingHeavyLoadQuery {
    // Test Configuration ----------------------------------------------------------------------------------------
    /** 
     * Number of threads in the executor. Determines the number of elements (and thus heavy load instructions)
-     * retrieved for each group. This means that each group should have a runtime of {@value #HeavyLoadInstructionDuration}ms,
+     * retrieved for each group. This means that each group should have a runtime of {@value #HeavyLoadInstructionTotalDuration}ms,
     * since the single instructions are executed concurrently.
     */
    private static final int ExecutorPoolSize = Math.max(3, Runtime.getRuntime().availableProcessors());
+    
    /** The number of groups contained in the initial data source. */
    private static final int DataSourceSize = 2000;
    private static final String GroupKeyPrefix = "G";
-    /** The time a heavy load instruction blocks the executing thread using {@link Thread#sleep(long)}. */
-    private static final long HeavyLoadInstructionDuration = 500;
+    
+    /** The time a step of a heavy load instruction blocks the executing thread (using {@link Thread#sleep(long)}). */
+    private static final long HeavyLoadInstructionStepDuration = 50;
+    /** The number of steps a heavy load instruction performs */
+    private static final int HeavyLoadInstructionNumberOfSteps = 10;
+    /** The total time a heavy load instruction blocks the executing thread (using {@link Thread#sleep(long)}). */
+    private static final long HeavyLoadInstructionTotalDuration = HeavyLoadInstructionStepDuration * HeavyLoadInstructionNumberOfSteps;
+    
    /** The number of milliseconds to wait before {@link Query#abort()} is called. */
-    private static final long AbortQueryDelay = (long) (HeavyLoadInstructionDuration * 5.45);
+    private static final long AbortQueryDelay = (long) (HeavyLoadInstructionTotalDuration * 5.45);
    /** The time given to the executor to complete all unfinished instructions */
-    private static final long TerminationTimeout = (long) (HeavyLoadInstructionDuration * 1.5);
+    private static final long TerminationTimeout = HeavyLoadInstructionStepDuration * 2;
    //------------------------------------------------------------------------------------------------------------
    
    // Execution Recording Configuration -------------------------------------------------------------------------
    /** Enables concurrent logging of the query execution and prints the current record before assertions. */
-    private static final boolean RecordExecution = false;
+    private static final boolean RecordExecution = true;
    private static final SimpleDateFormat DateFormatter = new SimpleDateFormat("HH:mm:ss.SSS");
    private ConcurrentLinkedQueue<String> executionRecord;
    //------------------------------------------------------------------------------------------------------------
@@ -174,6 +181,10 @@ public class TestAbortingHeavyLoadQuery {
        for (StatefulProcessorInstruction<?> instruction : runningInstructions) {
            assertTrue("computeResult() of a running unfinished instruction wasn't called", instruction.computeResultWasCalled());
            assertTrue("computeResult() of a running unfinished instruction didn't finish", instruction.computeResultWasFinished());
+            if (instruction instanceof StatefulBlockingInstruction) {
+                StatefulBlockingInstruction<?> blockingInstruction = (StatefulBlockingInstruction<?>) instruction;
+                assertTrue("computeResult() of a running heavy load instruction wasn't aborted", blockingInstruction.computeResultWasAborted());
+            }
        }
        for (StatefulProcessorInstruction<?> instruction : notStartedInstructions) {
            assertTrue("run() of an unstarted unfinished instruction wasn't called", instruction.runWasCalled());
@@ -216,7 +227,7 @@ public class TestAbortingHeavyLoadQuery {
     *   </li>
     *   <li>
     *     A heavy load instruction for each element is scheduled, which blocks the running thread for
-     *     {@value #HeavyLoadInstructionDuration}ms.
+     *     {@value #HeavyLoadInstructionTotalDuration}ms.
     *   </li>
     *   <li>
     *     Each element is grouped by its name and its value is used as value for the {@link GroupedDataEntry}.
@@ -306,8 +317,8 @@ public class TestAbortingHeavyLoadQuery {
                    @Override
                    protected ProcessorInstruction<Element> createInstruction(Element element) {
                        StatefulProcessorInstruction<Element> instruction = new HeavyLoadInstruction(this,
-                                ProcessorInstructionPriority.Extraction, HeavyLoadInstructionDuration, element,
-                                TestAbortingHeavyLoadQuery.this::logExecution);
+                                ProcessorInstructionPriority.Extraction, HeavyLoadInstructionStepDuration,
+                                HeavyLoadInstructionNumberOfSteps, element, TestAbortingHeavyLoadQuery.this::logExecution);
                        unfinishedInstructions.add(instruction);
                        return instruction;
                    }
@@ -397,9 +408,9 @@ public class TestAbortingHeavyLoadQuery {

        private final Consumer<String> recorder;

-        public HeavyLoadInstruction(ProcessorInstructionHandler<Element> handler,
-                ProcessorInstructionPriority priority, long blockDuration, Element result, Consumer<String> recorder) {
-            super(handler, priority, blockDuration, result);
+        public HeavyLoadInstruction(ProcessorInstructionHandler<Element> handler, ProcessorInstructionPriority priority,
+                long stepDuration, int numberOfSteps, Element result, Consumer<String> recorder) {
+            super(handler, priority, stepDuration, numberOfSteps, result);
            this.recorder = recorder;
        }
        
@@ -414,6 +425,11 @@ public class TestAbortingHeavyLoadQuery {
            recorder.accept("Starting work for heavy load instruction for " + result);
        }
        
+        @Override
+        protected void actionBeforeAbort() {
+            recorder.accept("Aborting heavy load instruction for " + result);
+        }
+        
        @Override
        protected void actionAfterBlock() {
            recorder.accept("Finished heavy load instruction for " + result);
@@ -1,7 +1,9 @@
 package com.sap.sse.datamining.impl.components;

 import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;

 import java.util.ArrayList;
 import java.util.Collections;
@@ -13,6 +15,7 @@ import java.util.concurrent.TimeUnit;
 import org.junit.Before;
 import org.junit.Test;

+import com.sap.sse.common.Util.Pair;
 import com.sap.sse.datamining.components.AdditionalResultDataBuilder;
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.components.ProcessorInstruction;
@@ -22,7 +25,7 @@ import com.sap.sse.datamining.test.util.components.StatefulBlockingInstruction;
 public class TestAbstractParallelProcessorElementProcessing {
    
    private ThreadPoolExecutor executor;
-    private Processor<Integer, Object> processor;
+    private Processor<Pair<Long, Integer>, Object> processor;
    private List<StatefulBlockingInstruction<?>> createdInstructions;
    
    @Before
@@ -30,10 +33,13 @@ public class TestAbstractParallelProcessorElementProcessing {
        int corePoolSize = Runtime.getRuntime().availableProcessors();
        executor = new ThreadPoolExecutor(corePoolSize, corePoolSize, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
        createdInstructions = new ArrayList<>();
-        processor = new AbstractParallelProcessor<Integer, Object>(Integer.class, Object.class, executor, Collections.emptySet()) {
+        
+        @SuppressWarnings("unchecked")
+        Class<Pair<Long, Integer>> inputType = (Class<Pair<Long, Integer>>)(Class<?>) Pair.class;
+        processor = new AbstractParallelProcessor<Pair<Long, Integer>, Object>(inputType, Object.class, executor, Collections.emptySet()) {
            @Override
-            protected ProcessorInstruction<Object> createInstruction(Integer sleepTime) {
-                StatefulBlockingInstruction<Object> instruction = new StatefulBlockingInstruction<>(this, sleepTime);
+            protected ProcessorInstruction<Object> createInstruction(Pair<Long, Integer> sleepTime) {
+                StatefulBlockingInstruction<Object> instruction = new StatefulBlockingInstruction<>(this, sleepTime.getA(), sleepTime.getB());
                createdInstructions.add(instruction);
                return instruction;
            }
@@ -44,79 +50,101 @@ public class TestAbstractParallelProcessorElementProcessing {

    @Test
    public void testSimpleProcessing() throws InterruptedException {
+        long stepDuration = 10;
+        int numberOfSteps = 1;
+        Pair<Long, Integer> input = new Pair<Long, Integer>(stepDuration, numberOfSteps);
        int elementCount = executor.getMaximumPoolSize() * 2;
        for (int i = 0; i < elementCount; i++) {
-            processor.processElement(10);
+            processor.processElement(input);
        }
        
        assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
+        double executionTime = Math.ceil((double) elementCount / executor.getMaximumPoolSize()) * stepDuration * numberOfSteps;
        executor.shutdown();
-        assertThat("Executor couldn't terminate", executor.awaitTermination(1, TimeUnit.SECONDS), is(true));
+        assertTrue("Executor couldn't terminate", executor.awaitTermination((long) (executionTime * 1.2), TimeUnit.SECONDS));
        for (StatefulBlockingInstruction<?> instruction : createdInstructions) {
-            assertThat("run wasn't called", instruction.runWasCalled(), is(true));
-            assertThat("computeResult wasn't called", instruction.computeResultWasCalled(), is(true));
-            assertThat("computeResult didn't finish", instruction.computeResultWasFinished(), is(true));
+            assertTrue("run wasn't called", instruction.runWasCalled());
+            assertTrue("computeResult wasn't called", instruction.computeResultWasCalled());
+            assertTrue("computeResult didn't finish", instruction.computeResultWasFinished());
+            assertFalse("computeResult was aborted", instruction.computeResultWasAborted());
        }
    }
    
    @Test
    public void testProcessingAfterFinish() throws InterruptedException {
-        int instructionDuration = 50;
+        long stepDuration = 10;
+        int numberOfSteps = 5;
+        Pair<Long, Integer> input = new Pair<Long, Integer>(stepDuration, numberOfSteps);
        int elementCount = executor.getMaximumPoolSize() + 1; // Last instruction will be queued
        for (int i = 0; i < elementCount; i++) {
-            processor.processElement(instructionDuration);
+            processor.processElement(input);
        }
        assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
        
-        Thread.sleep(instructionDuration / 3); // Giving some time to ensure execution of unqueued instructions
+        Thread.sleep(stepDuration); // Giving some time to ensure execution of unqueued instructions
        Thread finishingThread = ConcurrencyTestsUtil.tryToFinishTheProcessorInAnotherThread(processor);
        do {
            Thread.sleep(1);
        } while (!finishingThread.isAlive());
-        assertThat("Processor is already finished", processor.isFinished(), is(false));
+        assertFalse("Processor is already finished", processor.isFinished());
        // Processor not yet finished. New elements will be accepted
-        processor.processElement(instructionDuration);
+        processor.processElement(input);
        elementCount++;
        assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
        
-        finishingThread.join(1000);
-        assertThat("Processor isn't finished", processor.isFinished(), is(true));
-        processor.processElement(instructionDuration);
+        finishingThread.join();
+        assertTrue("Processor isn't finished", processor.isFinished());
+        processor.processElement(input);
        assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
        
        for (StatefulBlockingInstruction<?> instruction : createdInstructions) {
-            assertThat("run wasn't called", instruction.runWasCalled(), is(true));
-            assertThat("computeResult wasn't called", instruction.computeResultWasCalled(), is(true));
-            assertThat("computeResult didn't finish", instruction.computeResultWasFinished(), is(true));
+            assertTrue("run wasn't called", instruction.runWasCalled());
+            assertTrue("computeResult wasn't called", instruction.computeResultWasCalled());
+            assertTrue("computeResult didn't finish", instruction.computeResultWasFinished());
+            assertFalse("computeResult was aborted", instruction.computeResultWasAborted());
        }
    }
    
    @Test
    public void testProcessingAfterAbort() throws InterruptedException {
-        int instructionDuration = 50;
-        int elementCount = executor.getMaximumPoolSize() + 1; // Last instruction will be queued
+        long stepDuration = 10;
+        int numberOfSteps = 5;
+        Pair<Long, Integer> input = new Pair<Long, Integer>(stepDuration, numberOfSteps);
+        int elementCount = executor.getMaximumPoolSize()   // Set of finished instructions
+                           + executor.getMaximumPoolSize() // Set of started, but not yet finished instructions
+                           + 1; // Scheduled instruction
        for (int i = 0; i < elementCount; i++) {
-            processor.processElement(instructionDuration);
+            processor.processElement(input);
        }
        assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
-        
-        Thread.sleep(instructionDuration / 3); // Giving some time to ensure execution of unqueued instructions
+
+        long instructionDuration = stepDuration * numberOfSteps;
+        Thread.sleep(instructionDuration + stepDuration); // Time to finish first set and start second set
        processor.abort();
-        processor.processElement(0);
+        processor.processElement(new Pair<>(0L, 0));
        assertThat("Unexpected amount of created instructions", createdInstructions.size(), is(elementCount));
        
        executor.shutdown();
-        assertThat("Executor couldn't terminate", executor.awaitTermination(1, TimeUnit.SECONDS), is(true));
+        assertTrue("Executor couldn't terminate", executor.awaitTermination(2 * stepDuration, TimeUnit.MILLISECONDS));
        for (int i = 0; i < createdInstructions.size(); i++) {
            StatefulBlockingInstruction<?> instruction = createdInstructions.get(i);
-            assertThat("run wasn't called", instruction.runWasCalled(), is(true));
-            // Last instructions was executed after the processor has been aborted. computeResult() should not be called
-            if (i == createdInstructions.size() - 1) {
-                assertThat("computeResult of last instruction was called", instruction.computeResultWasCalled(), is(false));
-                assertThat("computeResult of last instruction finished", instruction.computeResultWasFinished(), is(false));
+            // run should be called for all instructions
+            assertTrue("run wasn't called", instruction.runWasCalled());
+            if (i < executor.getMaximumPoolSize()) {
+                // First set of instructions should be processed normally
+                assertTrue("computeResult wasn't called", instruction.computeResultWasCalled());
+                assertTrue("computeResult didn't finish", instruction.computeResultWasFinished());
+                assertFalse("computeResult was aborted", instruction.computeResultWasAborted());
+            } else if (i < executor.getMaximumPoolSize() * 2) {
+                // Second set of instructions was running when the processor was aborted. computeResult should be aborted
+                assertTrue("computeResult wasn't called", instruction.computeResultWasCalled());
+                assertTrue("computeResult didn't finish", instruction.computeResultWasFinished());
+                assertTrue("computeResult wasn't aborted", instruction.computeResultWasAborted());
            } else {
-                assertThat("computeResult wasn't called", instruction.computeResultWasCalled(), is(true));
-                assertThat("computeResult didn't finish", instruction.computeResultWasFinished(), is(true));
+                // Last instructions was still scheduled when the processor was aborted. computeResult should not be called
+                assertFalse("computeResult of last instruction was called", instruction.computeResultWasCalled());
+                assertFalse("computeResult of last instruction finished", instruction.computeResultWasFinished());
+                assertFalse("computeResult was aborted", instruction.computeResultWasAborted());
            }
        }
    }
@@ -5,38 +5,54 @@ import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;

 public class StatefulBlockingInstruction<ResultType> extends StatefulProcessorInstruction<ResultType> {
    
-    protected final long blockDuration;
+    protected final long stepDuration;
+    protected final int numberOfSteps;
    protected final ResultType result;
+    
+    private boolean computeResultWasAborted;

-    public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, long blockDuration) {
-        this(handler, 0, blockDuration, null);
+    public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, long stepDuration, int numberOfSteps) {
+        this(handler, 0, stepDuration, numberOfSteps, null);
    }
    
-    public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, ProcessorInstructionPriority priority, long blockDuration, ResultType result) {
-        this(handler, priority.asIntValue(), blockDuration, result);
+    public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, ProcessorInstructionPriority priority, long stepDuration, int numberOfSteps, ResultType result) {
+        this(handler, priority.asIntValue(), stepDuration, numberOfSteps, result);
    }
    
-    public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, int priority, long blockDuration, ResultType result) {
+    public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, int priority, long stepDuration, int numberOfSteps, ResultType result) {
        super(handler, priority);
-        this.blockDuration = blockDuration;
+        this.stepDuration = stepDuration;
+        this.numberOfSteps = numberOfSteps;
        this.result = result;
    }

    @Override
    protected ResultType internalComputeResult() throws Exception {
-        actionBeforeBlock();
-        if (blockDuration > 0) {
-            Thread.sleep(blockDuration);
+        if (getTotalBlockDuration() > 0) {
+            actionBeforeBlock();
+            for (int i = 0; i < numberOfSteps; i++) {
+                if (isAborted()) {
+                    actionBeforeAbort();
+                    computeResultWasAborted = true;
+                    break;
+                }
+                Thread.sleep(stepDuration);
+            }
+            actionAfterBlock();
        }
-        actionAfterBlock();
        return result;
    }
-    
+
    protected void actionBeforeBlock() { }
+    protected void actionBeforeAbort() { }
    protected void actionAfterBlock() { }
    
-    public long getBlockDuration() {
-        return blockDuration;
+    public long getTotalBlockDuration() {
+        return stepDuration * numberOfSteps;
+    }
+    
+    public boolean computeResultWasAborted() {
+        return computeResultWasAborted;
    }
    
 }
@@ -69,6 +69,9 @@ public abstract class AbstractParallelMultiDimensionalNestingGroupingProcessor<D
        } else {
            List<GroupKey> keys = new ArrayList<>();
            for (ParameterizedFunction<?> parameterizedDimension : parameterizedDimensions) {
+                if (isAborted()) {
+                    break;
+                }
                keys.add(createGroupKeyFor(input, parameterizedDimension.getFunction(), parameterizedDimension.getParameterProvider()));
            }
            return new CompoundGroupKey(keys);
@@ -52,7 +52,7 @@ public abstract class AbstractProcessorInstruction<ResultType> implements Proces
    @Override
    public void run() {
        try {
-            if (!handler.isAborted()) {
+            if (!isAborted()) {
                ResultType result = computeResult();
                handler.instructionSucceeded(result);
            }
@@ -62,6 +62,10 @@ public abstract class AbstractProcessorInstruction<ResultType> implements Proces
            handler.afterInstructionFinished(this);
        }
    }
+    
+    protected boolean isAborted() {
+        return handler.isAborted();
+    }

    protected abstract ResultType computeResult() throws Exception;
    
@@ -38,6 +38,9 @@ public abstract class AbstractRetrievalProcessor<InputType, ResultType> extends
            @Override
            public ResultType computeResult() {
                for (ResultType retrievedElement : retrieveData(element)) {
+                    if (isAborted()) {
+                        break;
+                    }
                    retrievedDataAmount.incrementAndGet();
                    forwardResultToReceivers(retrievedElement);
                }
@@ -82,6 +82,9 @@ public class ParallelGroupedNumberDataAverageAggregationProcessor
    protected Map<GroupKey, AverageWithStats<Number>> aggregateResult() {
        Map<GroupKey, AverageWithStats<Number>> result = new HashMap<>();
        for (Entry<GroupKey, DoubleHolder> sumAggregationEntry : sumPerKey.entrySet()) {
+            if (isAborted()) {
+                break;
+            }
            GroupKey key = sumAggregationEntry.getKey();
            result.put(key, new AverageWithStatsImpl<Number>(sumAggregationEntry.getValue().value / elementAmountPerKey.get(key).get(),
                    minPerKey.get(key), maxPerKey.get(key),
@@ -47,6 +47,9 @@ public class ParallelGroupedNumberDataMedianAggregationProcessor
    protected Map<GroupKey, Number> aggregateResult() {
        Map<GroupKey, Number> result = new HashMap<>();
        for (Entry<GroupKey, List<Number>> groupedValuesEntry : groupedValues.entrySet()) {
+            if (isAborted()) {
+                break;
+            }
            result.put(groupedValuesEntry.getKey(), getMedianOf(groupedValuesEntry.getValue()));
        }
        return result;
@@ -90,6 +90,9 @@ public class ParallelGroupedNumberPairAverageAggregationProcessor
    protected Map<GroupKey, PairWithStats<Number>> aggregateResult() {
        Map<GroupKey, PairWithStats<Number>> result = new HashMap<>();
        for (Entry<GroupKey, Pair<Number, Number>> sumAggregationEntry : sumPerKey.entrySet()) {
+            if (isAborted()) {
+                break;
+            }
            GroupKey key = sumAggregationEntry.getKey();
            result.put(key, new PairWithStatsImpl<Number>(new Pair<>(sumAggregationEntry.getValue().getA() != null ? sumAggregationEntry.getValue().getA().doubleValue() / elementAmountPerKey.get(key).get() : null, sumAggregationEntry.getValue().getB() != null ? sumAggregationEntry.getValue().getB().doubleValue() / elementAmountPerKey.get(key).get() : null) ,
                    minPerKey.get(key), maxPerKey.get(key),
@@ -81,6 +81,9 @@ public class ParallelGroupedNumberPairCollectingProcessor
    protected Map<GroupKey, PairWithStats<Number>> aggregateResult() {
        Map<GroupKey, PairWithStats<Number>> result = new HashMap<>();
        for (Entry<GroupKey, HashSet<Pair<Number, Number>>> sumAggregationEntry : individualPairs.entrySet()) {
+            if (isAborted()) {
+                break;
+            }
            GroupKey key = sumAggregationEntry.getKey();
            result.put(key, new PairWithStatsImpl<Number>(null,
                    /* min */ null,