(ignite-3) branch main updated: IGNITE-21484 fix possible race in JobExecution.changePriority method (#3186)

[ap...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

apkhmv pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git


The following commit(s) were added to refs/heads/main by this push:
     new 090e8ee3db IGNITE-21484 fix possible race in JobExecution.changePriority method (#3186)
090e8ee3db is described below

commit 090e8ee3dbce8182cfe9e26ed61166ba3a090ecd
Author: Dmitry Baranov <ne...@gmail.com>
AuthorDate: Thu Feb 15 15:18:24 2024 +0300

    IGNITE-21484 fix possible race in JobExecution.changePriority method (#3186)
---
 .../internal/compute/queue/QueueExecutionImpl.java | 47 +++++++++++++++-------
 1 file changed, 33 insertions(+), 14 deletions(-)

diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/QueueExecutionImpl.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/QueueExecutionImpl.java
index bb154876c8..3b29d421d8 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/QueueExecutionImpl.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/QueueExecutionImpl.java
@@ -23,6 +23,8 @@ import java.util.UUID;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import org.apache.ignite.compute.ComputeException;
 import org.apache.ignite.compute.JobStatus;
 import org.apache.ignite.internal.compute.state.ComputeStateMachine;
@@ -46,6 +48,8 @@ class QueueExecutionImpl<R> implements QueueExecution<R> {
 
     private final CompletableFuture<R> result = new CompletableFuture<>();
 
+    private final Lock executionLock = new ReentrantLock();
+
     @Nullable
     private volatile QueueEntry<R> queueEntry;
     private volatile int priority;
@@ -91,11 +95,7 @@ class QueueExecutionImpl<R> implements QueueExecution<R> {
 
             QueueEntry<R> queueEntry = this.queueEntry;
             if (queueEntry != null) {
-                if (executor.remove(queueEntry)) {
-                    result.cancel(true);
-                } else {
-                    queueEntry.interrupt();
-                }
+                cancel(queueEntry);
                 return true;
             }
         } catch (IllegalJobStateTransition e) {
@@ -104,21 +104,38 @@ class QueueExecutionImpl<R> implements QueueExecution<R> {
         return false;
     }
 
+    private void cancel(QueueEntry<R> queueEntry) {
+        executionLock.lock();
+        try {
+            if (executor.remove(queueEntry)) {
+                result.cancel(true);
+            } else {
+                queueEntry.interrupt();
+            }
+        } finally {
+            executionLock.unlock();
+        }
+    }
+
     @Override
     public boolean changePriority(int newPriority) {
         if (newPriority == priority) {
             return false;
         }
+        executionLock.lock();
+        try {
+            QueueEntry<R> queueEntry = this.queueEntry;
 
-        QueueEntry<R> queueEntry = this.queueEntry;
-
-        if (executor.removeFromQueue(queueEntry)) {
-            this.priority = newPriority;
-            this.queueEntry = null;
-            run();
-            return true;
+            if (executor.removeFromQueue(queueEntry)) {
+                this.priority = newPriority;
+                this.queueEntry = null;
+                run();
+                return true;
+            }
+            LOG.info("Cannot change job priority, job already processing. [job id = {}]", job);
+        } finally {
+            executionLock.unlock();
         }
-        LOG.info("Cannot change job priority, job already processing. [job id = {}]", job);
         return false;
     }
 
@@ -142,11 +159,14 @@ class QueueExecutionImpl<R> implements QueueExecution<R> {
         // either after the construction or after the failure.
         this.queueEntry = queueEntry;
 
+        executionLock.lock();
         try {
             executor.execute(queueEntry);
         } catch (QueueOverflowException e) {
             result.completeExceptionally(new ComputeException(QUEUE_OVERFLOW_ERR, e));
             return;
+        } finally {
+            executionLock.unlock();
         }
 
         queueEntry.toFuture().whenComplete((r, throwable) -> {
@@ -172,5 +192,4 @@ class QueueExecutionImpl<R> implements QueueExecution<R> {
             }
         });
     }
-
 }