[GitHub] [ignite-3] PakhomovAlexander commented on a diff in pull request #2219: IGNITE-19708 Check refcounter of unit before undeploy

["PakhomovAlexander (via GitHub)" <gi...@apache.org>]

PakhomovAlexander commented on code in PR #2219:
URL: https://github.com/apache/ignite-3/pull/2219#discussion_r1243482637


##########
modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentManagerImpl.java:
##########
@@ -353,6 +369,7 @@ public void start() {
         deploymentUnitStore.registerClusterStatusListener(clusterStatusWatchListener);
         messaging.subscribe();
         failover.registerTopologyChangeCallback(nodeStatusCallback, clusterEventCallback);
+        undeployer.start(5, TimeUnit.SECONDS);

Review Comment:
   What does 5 mean here? A good candidate to be a constant.



##########
modules/code-deployment/src/test/java/org/apache/ignite/internal/deployunit/DeploymentUnitUndeployerTest.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.deployunit;
+
+import static java.util.function.Predicate.isEqual;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.compute.DeploymentUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class DeploymentUnitUndeployerTest {
+    private static final int DELAY_IN_MILLIS = 500;
+
+    private final Set<DeploymentUnit> removingUnits = new CopyOnWriteArraySet<>();
+
+    @Mock
+    private DeploymentUnitAccessor deploymentUnitAccessor;
+
+
+    private DeploymentUnitUndeployer undeployer;
+
+    @BeforeEach
+    void setUp() {
+        undeployer = new DeploymentUnitUndeployer(
+                "testNode",
+                deploymentUnitAccessor,
+                removingUnits::add
+        );
+        undeployer.start(DELAY_IN_MILLIS, TimeUnit.MILLISECONDS);
+    }
+
+    @Test
+    void undeployNotAcquiredUnits() {
+        DeploymentUnit unit1 = new DeploymentUnit("unit1", "1.0.0");
+        DeploymentUnit unit2 = new DeploymentUnit("unit2", "1.0.0");
+        DeploymentUnit unit3 = new DeploymentUnit("unit3", "1.0.0");
+
+        // all units are released.
+        doReturn(false).when(deploymentUnitAccessor).isAcquired(any());
+
+        undeployer.undeploy(unit1);
+        undeployer.undeploy(unit2);
+        undeployer.undeploy(unit3);
+
+        // check all units are removed.
+        // 101 due to the fact that timeout must be greater than the poll delay (100 milliseconds).
+        await().timeout(101, TimeUnit.MILLISECONDS).until(() -> removingUnits.contains(unit1));

Review Comment:
   Where does the poll delay == 100 come from? I only see DELAY_IN_MILLIS = 500



##########
modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitUndeployer.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.deployunit;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import org.apache.ignite.compute.DeploymentUnit;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.thread.NamedThreadFactory;
+
+/**
+ * Undeploys deployment units that are not acquired by any class loader. This class is thread-safe.
+ */
+class DeploymentUnitUndeployer {
+    private static final IgniteLogger LOG = Loggers.forClass(DeploymentUnitUndeployer.class);
+
+    /** Deployment units to undeploy. */
+    private final Queue<DeploymentUnit> unitsToUndeploy = new ConcurrentLinkedDeque<>();
+
+    /** Deployment unit accessor. */
+    private final DeploymentUnitAccessor deploymentUnitAccessor;
+
+    /** Executor. */
+    private final ScheduledExecutorService executor;
+
+    /** Undeploy function. */
+    private final Consumer<DeploymentUnit> undeploy;
+
+    /**
+     * Creates undeployer.
+     *
+     * @param nodeName node name.
+     * @param deploymentUnitAccessor deployment unit accessor.
+     * @param undeploy undeploy function.
+     */
+    DeploymentUnitUndeployer(
+            String nodeName,
+            DeploymentUnitAccessor deploymentUnitAccessor,
+            Consumer<DeploymentUnit> undeploy) {
+        this.deploymentUnitAccessor = deploymentUnitAccessor;
+        this.executor = Executors.newScheduledThreadPool(
+                1, NamedThreadFactory.create(nodeName, "deployment-unit-undeployer", LOG));
+        this.undeploy = undeploy;
+    }
+
+    /**
+     * Starts the undeployer.
+     *
+     * @param delay delay between undeploy attempts.
+     * @param unit time unit of the delay.
+     */
+    public void start(long delay, TimeUnit unit) {
+        executor.scheduleWithFixedDelay(this::undeployUnits, 0, delay, unit);
+    }
+
+    /**
+     * Stops the undeployer.
+     */
+    public void stop() {
+        executor.shutdown();
+    }
+
+    /**
+     * Undeploys deployment units that are not acquired by any class loader. If a deployment unit is acquired, it is put back to the queue.
+     */
+    private void undeployUnits() {
+        int size = unitsToUndeploy.size();
+        for (int i = 0; i < size; i++) {
+            undeploy(unitsToUndeploy.remove());
+        }
+    }
+
+    /**
+     * Undeploys deployment unit. If the unit is acquired by any class loader, it is put to the queue.
+     *
+     * @param unit deployment unit to undeploy.
+     */
+    public void undeploy(DeploymentUnit unit) {
+        if (!deploymentUnitAccessor.isAcquired(unit)) {
+            undeploy.accept(unit);

Review Comment:
   You do not get the lock for the unit. It might be acquired by the time you call  `undeploy.accept(unit)`



##########
modules/code-deployment/src/test/java/org/apache/ignite/internal/deployunit/DeploymentUnitUndeployerTest.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.deployunit;
+
+import static java.util.function.Predicate.isEqual;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.compute.DeploymentUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class DeploymentUnitUndeployerTest {
+    private static final int DELAY_IN_MILLIS = 500;
+
+    private final Set<DeploymentUnit> removingUnits = new CopyOnWriteArraySet<>();
+
+    @Mock
+    private DeploymentUnitAccessor deploymentUnitAccessor;
+
+
+    private DeploymentUnitUndeployer undeployer;
+
+    @BeforeEach
+    void setUp() {
+        undeployer = new DeploymentUnitUndeployer(
+                "testNode",
+                deploymentUnitAccessor,
+                removingUnits::add
+        );
+        undeployer.start(DELAY_IN_MILLIS, TimeUnit.MILLISECONDS);
+    }
+
+    @Test
+    void undeployNotAcquiredUnits() {
+        DeploymentUnit unit1 = new DeploymentUnit("unit1", "1.0.0");
+        DeploymentUnit unit2 = new DeploymentUnit("unit2", "1.0.0");
+        DeploymentUnit unit3 = new DeploymentUnit("unit3", "1.0.0");
+
+        // all units are released.
+        doReturn(false).when(deploymentUnitAccessor).isAcquired(any());
+
+        undeployer.undeploy(unit1);
+        undeployer.undeploy(unit2);
+        undeployer.undeploy(unit3);
+
+        // check all units are removed.
+        // 101 due to the fact that timeout must be greater than the poll delay (100 milliseconds).
+        await().timeout(101, TimeUnit.MILLISECONDS).until(() -> removingUnits.contains(unit1));
+        await().timeout(101, TimeUnit.MILLISECONDS).until(() -> removingUnits.contains(unit2));
+        await().timeout(101, TimeUnit.MILLISECONDS).until(() -> removingUnits.contains(unit3));
+    }
+
+    @Test
+    void undeployAcquiredUnits() {
+        DeploymentUnit unit1 = new DeploymentUnit("unit1", "1.0.0");
+        DeploymentUnit unit2 = new DeploymentUnit("unit2", "1.0.0");
+        DeploymentUnit unit3 = new DeploymentUnit("unit3", "1.0.0");
+
+        // all units are acquired and will not be released in the future.
+        doReturn(true).when(deploymentUnitAccessor).isAcquired(any());
+
+        undeployer.undeploy(unit1);
+        undeployer.undeploy(unit2);
+        undeployer.undeploy(unit3);
+
+        // check all units are still not removed.
+        await().during(2, TimeUnit.SECONDS).until(
+                () -> removingUnits.contains(unit1) && removingUnits.contains(unit2) && removingUnits.contains(unit3),
+                isEqual(false)
+        );
+    }
+
+    @Test
+    void undeployReleasedUnits() {
+        DeploymentUnit unit1 = new DeploymentUnit("unit1", "1.0.0");
+        DeploymentUnit unit2 = new DeploymentUnit("unit2", "1.0.0");
+        DeploymentUnit unit3 = new DeploymentUnit("unit3", "1.0.0");
+
+        // unit1 and unit3 will be released in the future.
+        doReturn(true, true, false).when(deploymentUnitAccessor).isAcquired(eq(unit1));
+        doReturn(true, true, true).when(deploymentUnitAccessor).isAcquired(eq(unit2));
+        doReturn(true, false, false).when(deploymentUnitAccessor).isAcquired(eq(unit3));
+
+        undeployer.undeploy(unit1);
+        undeployer.undeploy(unit2);
+        undeployer.undeploy(unit3);
+
+        // check unit1 and unit3 were removed.
+        await().timeout(2, TimeUnit.SECONDS).until(() -> removingUnits.contains(unit1));
+        await().timeout(2, TimeUnit.SECONDS).until(() -> removingUnits.contains(unit3));
+
+        // check unit2 is still not removed.
+        await().during(2, TimeUnit.SECONDS).until(() -> removingUnits.contains(unit2), isEqual(false));
+    }
+
+    @Test
+    void notInfinityLoop() {
+        DeploymentUnit unit1 = new DeploymentUnit("unit1", "1.0.0");
+
+        doReturn(true).when(deploymentUnitAccessor).isAcquired(any());
+
+        undeployer.undeploy(unit1);
+
+        // check delay between attempts to undeploy the unit.
+        verify(deploymentUnitAccessor, timeout(DELAY_IN_MILLIS * 2).times(2)).isAcquired(eq(unit1));

Review Comment:
   This is potentially a flaky test. Is there a guarantee that it will be acquired exactly twice? What if it will be acquired only once due to GC pause?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org