Bugfix/#69 fix action pipeline deadlocks and other issues that cause random action cancelations.

module/dal/control/src/main/java/org/openbase/bco/dal/control/action/ActionImpl.java

@@ -68,6 +68,11 @@ public class ActionImpl implements SchedulableAction {
 
     protected final AbstractUnitController<?, ?> unit;
     private final SyncObject executionStateChangeSync = new SyncObject("ExecutionStateChangeSync");
+
+    // IMPORTANT LOCKING NOTE: There is no need to acquire the actionDescriptionBuilderLock in every case we need the action task exclusively.
+    // However, in case you acquire the action task lock and need to access the actionDescriptionBuilder within its synchronize scope,
+    // the actionDescriptionBuilderLock has to be acquired first before locking the action task.
+    // Otherwise, there is a high risk of deadlocks.
     private final SyncObject actionTaskLock = new SyncObject("ActionTaskLock");
     private final BundledReentrantReadWriteLock actionDescriptionBuilderLock;
     private ActionDescription.Builder actionDescriptionBuilder;
@@ -742,10 +747,15 @@ private void cancelActionTask() {
     }
 
     private void updateActionStateIfNotTerminating(final ActionState.State state) throws InterruptedException {
-        synchronized (actionTaskLock) {
-            if (isTerminating()) {
-                throw new InterruptedException();
+        actionDescriptionBuilderLock.lockReadInterruptibly();
+        try {
+            synchronized (actionTaskLock) {
+                if (isTerminating()) {
+                    throw new InterruptedException();
+                }
             }
+        } finally {
+            actionDescriptionBuilderLock.unlockRead();
         }
         updateActionState(state);
     }

module/dal/control/src/main/java/org/openbase/bco/dal/control/layer/unit/AbstractUnitController.java

@@ -1720,7 +1720,9 @@ public void cancelAllActions() throws InterruptedException, CouldNotPerformExcep
             }
 
             // final reschedule for cleanup
-            reschedule();
+            if(!isShutdownInProgress()) {
+                reschedule();
+            }
         } finally {
             builderSetup.unlockWrite(NotificationStrategy.AFTER_LAST_RELEASE);
         }

module/dal/control/src/main/java/org/openbase/bco/dal/control/layer/unit/scene/SceneControllerImpl.java

@@ -33,6 +33,7 @@
 import org.openbase.bco.dal.lib.layer.unit.UnitRemote;
 import org.openbase.bco.dal.lib.layer.unit.scene.SceneController;
 import org.openbase.bco.dal.lib.state.States.Activation;
+import org.openbase.bco.dal.remote.action.Actions;
 import org.openbase.bco.dal.remote.action.RemoteAction;
 import org.openbase.bco.dal.remote.action.RemoteActionPool;
 import org.openbase.bco.dal.remote.layer.unit.ButtonRemote;
@@ -256,6 +257,8 @@ private void stop() {
 
     public class ActivationStateOperationServiceImpl implements ActivationStateOperationService {
 
+        private RequiredActionObserver actionObserver = null;
+
         /**
          * Sets the activation state of the scene
          *
@@ -268,6 +271,11 @@ public synchronized Future<ActionDescription> setActivationState(final Activatio
             final ActionDescription.Builder responsibleActionBuilder = activationState.getResponsibleAction().toBuilder();
 
 
+            // shutdown all existing action observer to not let old observation interfere with new activations.
+            if(actionObserver != null) {
+                actionObserver.shutdown();
+            }
+
             // mark scene action as not replaceable since scene takes care of managing this actions.
             responsibleActionBuilder.setReplaceable(false);
 
@@ -305,10 +313,10 @@ public synchronized Future<ActionDescription> setActivationState(final Activatio
                                 requiredActionImpactList.addAll(requiredAction.getActionImpact(true));
 
                             } catch (org.openbase.jul.exception.TimeoutException ex) {
-                                // if the timeout is exhausted than just continue since we want to keep on trying as long as the scene is active.
+                                // if the timeout is exhausted then just continue since we want to keep on trying as long as the scene is active.
                                 continue;
                             } catch (CancellationException | CouldNotPerformException ex) {
-                                ExceptionPrinter.printHistory("Optional " + requiredAction + " of " + this + " could not executed!", ex, logger, LogLevel.WARN);
+                                ExceptionPrinter.printHistory("Required " + requiredAction + " of " + getLabel("?") + " could not be executed!", ex, logger, LogLevel.DEBUG);
                                 return FutureProcessor.canceledFuture(ActionDescription.class, new RejectedException("Required action " + requiredAction + " could not be executed", ex));
                             }
                         }
@@ -321,14 +329,13 @@ public synchronized Future<ActionDescription> setActivationState(final Activatio
                                 // if the timeout is exhausted than just continue since we want to keep on trying as long as the scene is active.
                                 continue;
                             } catch (CancellationException | CouldNotPerformException ex) {
-                                ExceptionPrinter.printHistory("Optional " + optionalAction + " of " + this + " could not executed!", ex, logger, LogLevel.WARN);
+                                ExceptionPrinter.printHistory("Optional " + optionalAction + " of " + getLabel("?") + " could not be executed!", ex, logger, LogLevel.TRACE);
                             }
                         }
 
                         // register an observer which will deactivate the scene if one required action is now longer running
-                        // observer will cleanup itself after the action is no longer valid so no need to care fore it.
                         try {
-                            new RequiredActionObserver(requiredActionImpactList, getActionById(responsibleActionBuilder.getActionId(), "SceneController"));
+                            actionObserver = new RequiredActionObserver(requiredActionImpactList, getActionById(responsibleActionBuilder.getActionId(), "SceneController"));
                         } catch (NotAvailableException ex) {
                             new FatalImplementationErrorException("Action is not available even when just created!", this, ex);
                         }
@@ -370,7 +377,7 @@ public ServiceType getServiceType() {
         }
     }
 
-    public static class RequiredActionObserver implements Observer<ServiceStateProvider<Message>, Message>, Shutdownable {
+    class RequiredActionObserver implements Observer<ServiceStateProvider<Message>, Message>, Shutdownable {
 
         private final Logger LOGGER = LoggerFactory.getLogger(RequiredActionObserver.class);
 
@@ -379,6 +386,8 @@ public static class RequiredActionObserver implements Observer<ServiceStateProvi
         private final HashMap<UnitRemote<?>, RequiredServiceDescription> unitAndRequiredServiceStateMap;
         private final Action responsibleAction;
 
+        private boolean destroy = false;
+
         private RequiredActionObserver(final List<ActionReference> requiredActionImpact, final Action responsibleAction) {
             this.responsibleAction = responsibleAction;
             this.unitAndRequiredServiceStateMap = new HashMap<>();
@@ -427,6 +436,7 @@ private RequiredActionObserver(final List<ActionReference> requiredActionImpact,
 
         private void verifyAllStates() {
             try {
+                logger.trace(() -> "verify "+unitAndRequiredServiceStateMap.entrySet().size()+ " states of "+ getLabel("?"));
                 for (Entry<UnitRemote<? extends Message>, RequiredServiceDescription> unitActionReferenceEntry : unitAndRequiredServiceStateMap.entrySet()) {
                     try {
                         // skip unit in case its offline, since then the verification is automatically
@@ -446,12 +456,25 @@ private void verifyAllStates() {
 
         private void verifyState(final ServiceProvider<? extends Message> unit, final Message serviceState) throws VerificationFailedException {
 
+            // skip verification on destroyed required action observer!
+            if(destroy) {
+                return;
+            }
+
             if (!responsibleAction.isValid()) {
-                throw new VerificationFailedException("Action not valid anymore!");
+                throw new VerificationFailedException("The activation of " + getLabel("?") + " is not valid anymore.");
+            }
+
+            // skip in case no service state was delivered
+            if(serviceState.toString().isBlank()) {
+                return;
             }
 
             if (!Services.equalServiceStates(unitAndRequiredServiceStateMap.get(unit).getServiceState(), serviceState)) {
-                throw new VerificationFailedException("State of " + unit + "not meet!");
+                logger.trace(() -> unitAndRequiredServiceStateMap.get(unit).getServiceState() + " is not equals " + serviceState.toString().substring(0, 20) + " and will cancel: " + SceneControllerImpl.this.getLabel("?"));
+                if(Actions.validateInitialAction(serviceState)) {
+                    throw new VerificationFailedException("State of " + unit + "not meet!");
+                }
             }
         }
 
@@ -477,6 +500,7 @@ public void update(ServiceStateProvider<Message> serviceStateProvider, Message s
 
         @Override
         public void shutdown() {
+            destroy = true;
             // deregister observation
             for (Entry<UnitRemote<?>, RequiredServiceDescription> unitActionReferenceEntry : unitAndRequiredServiceStateMap.entrySet()) {
                 unitActionReferenceEntry.getKey().removeServiceStateObserver(ServiceTempus.CURRENT, unitActionReferenceEntry.getValue().getServiceType(), this);

module/dal/lib/src/main/java/org/openbase/bco/dal/lib/layer/service/Services.java

@@ -1259,7 +1259,7 @@ public static boolean equalServiceStates(final Message serviceState1, final Mess
         }
 
         // fail if only one is empty
-        if (serviceState1.equals(serviceState1.getDefaultInstanceForType()) || serviceState2.equals(serviceState2.getDefaultInstanceForType())) {
+        if (serviceState1.equals(serviceState1.getDefaultInstanceForType()) ^ serviceState2.equals(serviceState2.getDefaultInstanceForType())) {
             return false;
         }
 
@@ -1373,5 +1373,3 @@ public static String getServiceStateClassName(final Message serviceState) {
         return serviceState.getClass().getName();
     }
 }
-
-

module/dal/remote/src/main/java/org/openbase/bco/dal/remote/action/Actions.java

@@ -22,6 +22,9 @@
  * #L%
  */
 
+import com.google.protobuf.Message;
+import org.openbase.bco.dal.lib.action.ActionDescriptionProcessor;
+import org.openbase.bco.dal.lib.layer.service.ServiceStateProcessor;
 import org.openbase.jul.exception.CouldNotPerformException;
 import org.openbase.type.domotic.action.ActionDescriptionType.ActionDescription;
 import org.openbase.type.domotic.state.ActionStateType.ActionState;
@@ -55,4 +58,27 @@ public static RemoteAction waitForActionState(final Future<ActionDescription> ac
         remoteAction.waitForActionState(actionState, timeout, timeUnit);
         return remoteAction;
     }
+
+    public static boolean validateInitialAction(Message serviceState) {
+        try {
+            return validateInitialAction(ServiceStateProcessor.getResponsibleAction(serviceState));
+        } catch (CouldNotPerformException e) {
+            // skip validation in error case.
+        }
+        return false;
+    }
+
+    public static boolean validateInitialAction(ActionDescription actionDescription) {
+        try {
+            final RemoteAction initialActionOfIncomingServiceState =
+                    new RemoteAction(ActionDescriptionProcessor.getInitialActionReference(actionDescription));
+            initialActionOfIncomingServiceState.waitForRegistration();
+            return initialActionOfIncomingServiceState.isValid();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        } catch (CouldNotPerformException e) {
+            // skip if validation failed.
+        }
+        return false;
+    }
 }