You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "Flaugh24 (via GitHub)" <gi...@apache.org> on 2023/06/19 17:33:26 UTC
[GitHub] [ignite-3] Flaugh24 opened a new pull request, #2219: IGNITE-19708 Check refcounter of unit before undeploy
Flaugh24 opened a new pull request, #2219:
URL: https://github.com/apache/ignite-3/pull/2219
https://issues.apache.org/jira/browse/IGNITE-19708
--
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
[GitHub] [ignite-3] valepakh commented on a diff in pull request #2219: IGNITE-19708 Check refcounter of unit before undeploy
Posted by "valepakh (via GitHub)" <gi...@apache.org>.
valepakh commented on code in PR #2219:
URL: https://github.com/apache/ignite-3/pull/2219#discussion_r1241835616
##########
modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DefaultNodeCallback.java:
##########
@@ -53,20 +56,23 @@ public class DefaultNodeCallback extends NodeEventCallback {
*
* @param deploymentUnitStore Deployment units store.
* @param messaging Deployment messaging service.
+ * @param undeployer Deployment unit undeployer.
* @param deployer Deployment unit file system service.
* @param cmgManager Cluster management group manager.
* @param nodeName Node consistent ID.
*/
public DefaultNodeCallback(
DeploymentUnitStore deploymentUnitStore,
DeployMessagingService messaging,
+ DeploymentUnitUndeployer undeployer,
Review Comment:
Could you please move the `undeployer` parameter after the `deployer`?
##########
modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitUndeployer.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.Deque;
+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 Deque<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 i = 0;
+ while (i < unitsToUndeploy.size()) {
+ DeploymentUnit unit = unitsToUndeploy.poll();
+ if (!deploymentUnitAccessor.isAcquired(unit)) {
+ undeploy.accept(unit);
+ } else {
+ unitsToUndeploy.offer(unit);
+ }
+ }
+ }
+
+ /**
+ * Adds a deployment unit to the queue of units to undeploy.
+ *
+ * @param unit deployment unit to undeploy.
+ */
+ public void undeploy(DeploymentUnit unit) {
+ unitsToUndeploy.offer(unit);
Review Comment:
Given the chosen time delay of 5 seconds, let's first check that the unit is not used right now and add it to the queue only if it is.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeTestStandalone.java:
##########
@@ -115,20 +118,58 @@ void executesJobWithLatestUnitVersion() throws IOException {
IgniteImpl entryNode = node(0);
DeploymentUnit firstVersion = new DeploymentUnit("latest-unit", Version.parseVersion("1.0.0"));
- deployJar(entryNode, firstVersion.name(), firstVersion.version(), "unit1-1.0-SNAPSHOT.jar");
+ deployJar(entryNode, firstVersion.name(), firstVersion.version(), "ignite-ut-job1-1.0-SNAPSHOT.jar");
CompletableFuture<Integer> result1 = entryNode.compute()
.execute(Set.of(entryNode.node()), jobUnits, "org.my.job.compute.unit.UnitJob");
assertThat(result1, willBe(1));
DeploymentUnit secondVersion = new DeploymentUnit("latest-unit", Version.parseVersion("1.0.1"));
- deployJar(entryNode, secondVersion.name(), secondVersion.version(), "unit2-1.0-SNAPSHOT.jar");
+ deployJar(entryNode, secondVersion.name(), secondVersion.version(), "ignite-ut-job2-1.0-SNAPSHOT.jar");
CompletableFuture<String> result2 = entryNode.compute()
.execute(Set.of(entryNode.node()), jobUnits, "org.my.job.compute.unit.UnitJob");
assertThat(result2, willBe("Hello World!"));
}
+ @Test
+ void undeployAcquiredUnit() {
+ IgniteImpl entryNode = node(0);
+ CompletableFuture<Void> job = entryNode.compute().execute(Set.of(entryNode.node()), units, "org.example.SleepJob", 3L);
+
+ assertThat(entryNode.deployment().undeployAsync(unit.name(), unit.version()), willCompleteSuccessfully());
+
+ assertThat(entryNode.deployment().clusterStatusAsync(unit.name(), unit.version()), willBe(OBSOLETE));
+ assertThat(entryNode.deployment().nodeStatusAsync(unit.name(), unit.version()), willBe(OBSOLETE));
+
+ await().failFast("The unit must not be removed until the job is completed", () -> {
+ assertThat(entryNode.deployment().clusterStatusAsync(unit.name(), unit.version()), willBe(OBSOLETE));
+ assertThat(entryNode.deployment().nodeStatusAsync(unit.name(), unit.version()), willBe(OBSOLETE));
+ }).until(() -> job, willCompleteSuccessfully());
+
+ await().until(
+ () -> entryNode.deployment().clusterStatusAsync(unit.name(), unit.version()),
+ willBe(nullValue())
+ );
+ }
+
+ @Test
+ void executeJonWithObsoleteUnit() {
Review Comment:
```suggestion
void executeJobWithObsoleteUnit() {
```
--
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
[GitHub] [ignite-3] PakhomovAlexander merged pull request #2219: IGNITE-19708 Check refcounter of unit before undeploy
Posted by "PakhomovAlexander (via GitHub)" <gi...@apache.org>.
PakhomovAlexander merged PR #2219:
URL: https://github.com/apache/ignite-3/pull/2219
--
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
[GitHub] [ignite-3] Flaugh24 commented on a diff in pull request #2219: IGNITE-19708 Check refcounter of unit before undeploy
Posted by "Flaugh24 (via GitHub)" <gi...@apache.org>.
Flaugh24 commented on code in PR #2219:
URL: https://github.com/apache/ignite-3/pull/2219#discussion_r1243765330
##########
modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitAccessorImpl.java:
##########
@@ -27,6 +30,8 @@
public class DeploymentUnitAccessorImpl implements DeploymentUnitAccessor {
private final RefCountedObjectPool<DeploymentUnit, DisposableDeploymentUnit> pool = new RefCountedObjectPool<>();
+ private final RefCountedObjectPool<DeploymentUnit, Lock> locks = new RefCountedObjectPool<>();
Review Comment:
Yep, we have to remove unused locks
--
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
[GitHub] [ignite-3] PakhomovAlexander commented on a diff in pull request #2219: IGNITE-19708 Check refcounter of unit before undeploy
Posted by "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
[GitHub] [ignite-3] PakhomovAlexander commented on a diff in pull request #2219: IGNITE-19708 Check refcounter of unit before undeploy
Posted by "PakhomovAlexander (via GitHub)" <gi...@apache.org>.
PakhomovAlexander commented on code in PR #2219:
URL: https://github.com/apache/ignite-3/pull/2219#discussion_r1243684104
##########
modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitAccessorImpl.java:
##########
@@ -36,7 +37,7 @@ public DeploymentUnitAccessorImpl(FileDeployerService deployer) {
* {@inheritDoc}
*/
@Override
- public DisposableDeploymentUnit acquire(DeploymentUnit unit) {
+ public synchronized DisposableDeploymentUnit acquire(DeploymentUnit unit) {
Review Comment:
You synchronize by accessor instance which is not the best choice here. The completely independent process that is needed `unit1` will be blocked by another process that runs `computeIfNotAcquired` for `unit2`.
--
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
[GitHub] [ignite-3] Flaugh24 commented on a diff in pull request #2219: IGNITE-19708 Check refcounter of unit before undeploy
Posted by "Flaugh24 (via GitHub)" <gi...@apache.org>.
Flaugh24 commented on code in PR #2219:
URL: https://github.com/apache/ignite-3/pull/2219#discussion_r1243690872
##########
modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitAccessorImpl.java:
##########
@@ -36,7 +37,7 @@ public DeploymentUnitAccessorImpl(FileDeployerService deployer) {
* {@inheritDoc}
*/
@Override
- public DisposableDeploymentUnit acquire(DeploymentUnit unit) {
+ public synchronized DisposableDeploymentUnit acquire(DeploymentUnit unit) {
Review Comment:
I'm not sure that is going to be a bottle neck
--
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
[GitHub] [ignite-3] PakhomovAlexander commented on a diff in pull request #2219: IGNITE-19708 Check refcounter of unit before undeploy
Posted by "PakhomovAlexander (via GitHub)" <gi...@apache.org>.
PakhomovAlexander commented on code in PR #2219:
URL: https://github.com/apache/ignite-3/pull/2219#discussion_r1243761584
##########
modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitAccessorImpl.java:
##########
@@ -27,6 +30,8 @@
public class DeploymentUnitAccessorImpl implements DeploymentUnitAccessor {
private final RefCountedObjectPool<DeploymentUnit, DisposableDeploymentUnit> pool = new RefCountedObjectPool<>();
+ private final RefCountedObjectPool<DeploymentUnit, Lock> locks = new RefCountedObjectPool<>();
Review Comment:
Why do you use `RefCountedObjectPool` for locks? Do you need to count them?
--
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
[GitHub] [ignite-3] Pochatkin commented on a diff in pull request #2219: IGNITE-19708 Check refcounter of unit before undeploy
Posted by "Pochatkin (via GitHub)" <gi...@apache.org>.
Pochatkin commented on code in PR #2219:
URL: https://github.com/apache/ignite-3/pull/2219#discussion_r1243585152
##########
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;
+
+
Review Comment:
Remove empty line
##########
modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitProcessor.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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;
+
+/**
+ * Executes action on the deployment unit if it is not acquired. Otherwise, puts it to the queue.
+ */
+class DeploymentUnitProcessor {
+ private static final IgniteLogger LOG = Loggers.forClass(DeploymentUnitProcessor.class);
+
+ /** Deployment units to undeploy. */
+ private final Queue<DeploymentUnit> unitsToProcess = new ConcurrentLinkedDeque<>();
+
+ /** Deployment unit accessor. */
+ private final DeploymentUnitAccessor deploymentUnitAccessor;
+
+ /** Executor. */
+ private final ScheduledExecutorService executor;
+
+ /** Action. */
+ private final Consumer<DeploymentUnit> action;
+
+ /**
+ * Creates processor.
+ *
+ * @param nodeName node name.
+ * @param deploymentUnitAccessor deployment unit accessor.
+ * @param action action.
+ */
+ DeploymentUnitProcessor(
Review Comment:
Processor is so abstract. Lets rename it to DeploymentUnitAcquiredWaiter/Checker ?
##########
modules/code-deployment/src/test/java/org/apache/ignite/internal/deployunit/DeploymentUnitProcessorTest.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.emptyIterable;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.after;
+import static org.mockito.Mockito.atLeastOnce;
+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.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Spy;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class DeploymentUnitProcessorTest {
+ private static final int DELAY_IN_MILLIS = 500;
+
+ private final Set<DeploymentUnit> removingUnits = new CopyOnWriteArraySet<>();
+
+ @Mock
+ private FileDeployerService deployerService;
+
+ @Spy
+ @InjectMocks
+ private DeploymentUnitAccessorImpl deploymentUnitAccessor;
+
+
Review Comment:
Remove empty line
##########
modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitProcessor.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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;
+
+/**
+ * Executes action on the deployment unit if it is not acquired. Otherwise, puts it to the queue.
+ */
+class DeploymentUnitProcessor {
+ private static final IgniteLogger LOG = Loggers.forClass(DeploymentUnitProcessor.class);
+
+ /** Deployment units to undeploy. */
+ private final Queue<DeploymentUnit> unitsToProcess = new ConcurrentLinkedDeque<>();
+
+ /** Deployment unit accessor. */
+ private final DeploymentUnitAccessor deploymentUnitAccessor;
+
+ /** Executor. */
+ private final ScheduledExecutorService executor;
+
+ /** Action. */
+ private final Consumer<DeploymentUnit> action;
+
+ /**
+ * Creates processor.
+ *
+ * @param nodeName node name.
+ * @param deploymentUnitAccessor deployment unit accessor.
+ * @param action action.
+ */
+ DeploymentUnitProcessor(
+ String nodeName,
+ DeploymentUnitAccessor deploymentUnitAccessor,
+ Consumer<DeploymentUnit> action) {
+ this.deploymentUnitAccessor = deploymentUnitAccessor;
+ this.executor = Executors.newScheduledThreadPool(
+ 1, NamedThreadFactory.create(nodeName, "deployment-unit-undeployer", LOG));
+ this.action = action;
+ }
+
+ /**
+ * Starts the processor.
+ *
+ * @param delay delay between undeploy attempts.
+ * @param unit time unit of the delay.
+ */
+ public void start(long delay, TimeUnit unit) {
+ executor.scheduleWithFixedDelay(this::processUnits, 0, delay, unit);
+ }
+
+ /**
+ * Stops the processor.
+ */
+ public void stop() {
+ executor.shutdown();
+ }
+
+ /**
+ * Processes all deployment units in the queue.
+ */
+ private void processUnits() {
+ int size = unitsToProcess.size();
+ for (int i = 0; i < size; i++) {
+ process(unitsToProcess.remove());
+ }
+ }
+
+ /**
+ * Executes action on the deployment unit if it is not acquired. Otherwise, puts it to the queue.
+ *
+ * @param unit deployment unit to undeploy.
+ */
+ public void process(DeploymentUnit unit) {
Review Comment:
Name process also so abstract
--
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