Fix concurrency issue

Author: João Jandre

plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/BlockCommitListener.java

@@ -27,24 +27,39 @@
 import org.libvirt.event.BlockJobStatus;
 import org.libvirt.event.BlockJobType;
 
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
 public class BlockCommitListener implements BlockJobListener {
     private String result;
     private String vmName;
 
     private Logger logger;
     private String logid;
+    private Semaphore semaphore;
 
     protected BlockCommitListener(String vmName, String logid) {
         this.vmName = vmName;
         this.logid = logid;
         this.logger = LogManager.getLogger(getClass());
+        this.semaphore = new Semaphore(0);
         this.result = String.format("Failed to block commit disk of VM [%s]. Libvirt did not launch an event for it.", vmName);
     }
 
-    protected String getResult() {
+    protected String getResult(int timeout) {
+        this.waitBlockCommit(timeout);
         return result;
     }
 
+    protected void waitBlockCommit(int timeout) {
+        try {
+            logger.debug("Trying to acquire result semaphore. If the correct event was not launched, will wait for [{}] seconds before giving up.", timeout);
+            this.semaphore.tryAcquire(timeout, TimeUnit.SECONDS);
+        } catch (InterruptedException ex) {
+            logger.error("Thread that was tracking the progress for the block commit job of vm {} was interrupted.", vmName, ex);
+        }
+    }
+
     @Override
     public void onEvent(Domain domain, String diskPath, BlockJobType type, BlockJobStatus status) {
         if (!BlockJobType.COMMIT.equals(type) && !BlockJobType.ACTIVE_COMMIT.equals(type)) {
@@ -56,17 +71,20 @@ public void onEvent(Domain domain, String diskPath, BlockJobType type, BlockJobS
         switch (status) {
             case COMPLETED:
                 result = null;
+                semaphore.release();
                 return;
             case READY:
                 try {
                     logger.debug("Pivoting disk [{}] of VM [{}].", diskPath, vmName);
                     domain.blockJobAbort(diskPath, Domain.BlockJobAbortFlags.PIVOT);
                 } catch (LibvirtException ex) {
                     result = String.format("Failed to pivot disk due to [%s].", ex.getMessage());
+                    semaphore.release();
                 }
                 return;
             default:
                 result = String.format("Failed to block commit disk with status [%s].", status);
+                semaphore.release();
         }
     }
 }

plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java

@@ -6652,6 +6652,8 @@ protected void mergeSnapshotIntoBaseFileWithEventsAndConfigurableTimeout(Domain
         }
 
         BlockCommitListener blockCommitListener = getBlockCommitListener(vmName);
+        int remainingTimeout = 0;
+        String mergeResult = "Got an exception during block commit wait. Check earlier logs for more info.";
         try {
             vm.addBlockJobListener(blockCommitListener);
 
@@ -6660,12 +6662,12 @@ protected void mergeSnapshotIntoBaseFileWithEventsAndConfigurableTimeout(Domain
 
             vm.blockCommit(diskLabel, baseFilePath, topFilePath, 0, commitFlags);
 
-            checkBlockCommitProgress(vm, diskLabel, vmName, snapshotName, topFilePath, baseFilePath);
+            remainingTimeout = checkBlockCommitProgress(vm, diskLabel, vmName, snapshotName, topFilePath, baseFilePath);
+            mergeResult = blockCommitListener.getResult(remainingTimeout);
         } finally {
             vm.removeBlockJobListener(blockCommitListener);
         }
 
-        String mergeResult = blockCommitListener.getResult();
         if (mergeResult != null) {
             String commitError = String.format("Failed the block commit of top file [%s] into base file [%s] for snapshot [%s] of VM [%s]. The job will be left running to avoid" +
                     " data corruption, but ACS will return an error and volume [%s] will need to be normalized manually. If the commit involved the active image, the pivot will" +
@@ -6730,7 +6732,7 @@ protected BlockCommitListener getBlockCommitListener(String vmName) {
         return new BlockCommitListener(vmName, ThreadContext.get("logcontextid"));
     }
 
-    protected void checkBlockCommitProgress(Domain vm, String diskLabel, String vmName, String snapshotName, String topFilePath, String baseFilePath) {
+    protected int checkBlockCommitProgress(Domain vm, String diskLabel, String vmName, String snapshotName, String topFilePath, String baseFilePath) {
         int timeout = qcow2DeltaMergeTimeout;
         DomainBlockJobInfo result;
         long lastCommittedBytes = 0;
@@ -6751,12 +6753,12 @@ protected void checkBlockCommitProgress(Domain vm, String diskLabel, String vmNa
                 result = vm.getBlockJobInfo(diskLabel, 0);
             } catch (LibvirtException ex) {
                 logger.warn("Exception while getting block job info {}: [{}].", partialLog, ex.getMessage(), ex);
-                return;
+                return timeout;
             }
 
             if (result == null || result.type == 0 && result.end == 0 && result.cur == 0) {
                 logger.debug("Block commit job {} has already finished.", partialLog);
-                return;
+                return timeout;
             }
 
             long currentCommittedBytes = result.cur;
@@ -6766,7 +6768,9 @@ protected void checkBlockCommitProgress(Domain vm, String diskLabel, String vmNa
             lastCommittedBytes = currentCommittedBytes;
             endBytes = result.end;
         }
-        logger.warn("Block commit {} has timed out after waiting at least {} seconds. The progress of the operation was [{}] of [{}].", partialLog, qcow2DeltaMergeTimeout, lastCommittedBytes, endBytes);
+        logger.warn("Block commit {} has timed out after waiting at least {} seconds. The progress of the operation was [{}] of [{}].", partialLog,
+                qcow2DeltaMergeTimeout, lastCommittedBytes, endBytes);
+        return 0;
     }
 
     /**

plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResourceTest.java

@@ -27,6 +27,7 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.nullable;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.doReturn;
@@ -6697,7 +6698,7 @@ public void mergeSnapshotIntoBaseFileTestActiveAndDeleteFlags() throws Exception
             threadContextMockedStatic.when(() ->
                     ThreadContext.get(Mockito.anyString())).thenReturn("logid");
             Mockito.doReturn(blockCommitListenerMock).when(libvirtComputingResourceSpy).getBlockCommitListener(Mockito.any());
-            Mockito.doReturn(null).when(blockCommitListenerMock).getResult();
+            Mockito.doReturn(null).when(blockCommitListenerMock).getResult(anyInt());
             Mockito.doNothing().when(domainMock).addBlockJobListener(blockCommitListenerMock);
             Mockito.doReturn(null).when(domainMock).getBlockJobInfo(Mockito.anyString(), Mockito.anyInt());
             Mockito.doNothing().when(domainMock).removeBlockJobListener(blockCommitListenerMock);
@@ -6726,7 +6727,7 @@ public void mergeSnapshotIntoBaseFileTestActiveFlag() throws Exception {
             threadContextMockedStatic.when(() ->
                     ThreadContext.get(Mockito.anyString())).thenReturn("logid");
             Mockito.doReturn(blockCommitListenerMock).when(libvirtComputingResourceSpy).getBlockCommitListener(Mockito.any());
-            Mockito.doReturn(null).when(blockCommitListenerMock).getResult();
+            Mockito.doReturn(null).when(blockCommitListenerMock).getResult(anyInt());
             Mockito.doNothing().when(domainMock).addBlockJobListener(blockCommitListenerMock);
             Mockito.doNothing().when(domainMock).removeBlockJobListener(blockCommitListenerMock);
             Mockito.doNothing().when(libvirtComputingResourceSpy).manuallyDeleteUnusedSnapshotFile(Mockito.anyBoolean(), Mockito.anyString());
@@ -6753,7 +6754,7 @@ public void mergeSnapshotIntoBaseFileTestDeleteFlag() throws Exception {
             libvirtUtilitiesHelperMockedStatic.when(() -> LibvirtUtilitiesHelper.isLibvirtSupportingFlagDeleteOnCommandVirshBlockcommit(Mockito.any())).thenReturn(true);
             threadContextMockedStatic.when(() -> ThreadContext.get(Mockito.anyString())).thenReturn("logid");
             Mockito.doReturn(blockCommitListenerMock).when(libvirtComputingResourceSpy).getBlockCommitListener(Mockito.any());
-            Mockito.doReturn(null).when(blockCommitListenerMock).getResult();
+            Mockito.doReturn(null).when(blockCommitListenerMock).getResult(anyInt());
             Mockito.doNothing().when(domainMock).addBlockJobListener(blockCommitListenerMock);
             Mockito.doReturn(null).when(domainMock).getBlockJobInfo(Mockito.anyString(), Mockito.anyInt());
             Mockito.doNothing().when(domainMock).removeBlockJobListener(blockCommitListenerMock);
@@ -6781,7 +6782,7 @@ public void mergeSnapshotIntoBaseFileTestNoFlags() throws Exception {
             libvirtUtilitiesHelperMockedStatic.when(() -> LibvirtUtilitiesHelper.isLibvirtSupportingFlagDeleteOnCommandVirshBlockcommit(Mockito.any())).thenReturn(false);
             threadContextMockedStatic.when(() -> ThreadContext.get(Mockito.anyString())).thenReturn("logid");
             Mockito.doReturn(blockCommitListenerMock).when(libvirtComputingResourceSpy).getBlockCommitListener(Mockito.any());
-            Mockito.doReturn(null).when(blockCommitListenerMock).getResult();
+            Mockito.doReturn(null).when(blockCommitListenerMock).getResult(anyInt());
             Mockito.doNothing().when(domainMock).addBlockJobListener(blockCommitListenerMock);
             Mockito.doReturn(null).when(domainMock).getBlockJobInfo(Mockito.anyString(), Mockito.anyInt());
             Mockito.doNothing().when(domainMock).removeBlockJobListener(blockCommitListenerMock);
@@ -6811,7 +6812,7 @@ public void mergeSnapshotIntoBaseFileTestMergeFailsThrowException() throws Excep
             Mockito.doNothing().when(domainMock).removeBlockJobListener(Mockito.any());
 
             Mockito.doReturn(blockCommitListenerMock).when(libvirtComputingResourceSpy).getBlockCommitListener(Mockito.any());
-            Mockito.doReturn("Failed").when(blockCommitListenerMock).getResult();
+            Mockito.doReturn("Failed").when(blockCommitListenerMock).getResult(anyInt());
 
             String diskLabel = "vda";
             String baseFilePath = "/file";