Merge pull request #42745 from gabilang/fix-worker-autoclose-master

Fix worker send hangs when worker early return within if stmt
ballerina-platform · Jul 9, 2024 · 6c996f9 · 6c996f9
2 parents b8fd113 + f50ebd7
commit 6c996f9
Show file tree

Hide file tree

Showing 6 changed files with 145 additions and 4 deletions.
diff --git a/bvm/ballerina-runtime/src/main/java/module-info.java b/bvm/ballerina-runtime/src/main/java/module-info.java
@@ -65,7 +65,7 @@
             io.ballerina.lang.regexp;
     exports io.ballerina.runtime.internal.values to io.ballerina.testerina.core, io.ballerina.testerina.runtime,
             io.ballerina.lang.xml, org.ballerinalang.debugadapter.runtime, io.ballerina.lang.query,
-            io.ballerina.lang.function, io.ballerina.lang.regexp, io.ballerina.lang.value;
+            io.ballerina.lang.function, io.ballerina.lang.regexp, io.ballerina.lang.value, io.ballerina.lang.internal;
     exports io.ballerina.runtime.internal.configurable to io.ballerina.lang.internal;
     exports io.ballerina.runtime.internal.types to io.ballerina.lang.typedesc, io.ballerina.testerina.runtime,
             org.ballerinalang.debugadapter.runtime, io.ballerina.lang.function, io.ballerina.lang.regexp, io.ballerina.testerina.core;

diff --git a/langlib/lang.__internal/src/main/java/org/ballerinalang/langlib/internal/WorkerChannels.java b/langlib/lang.__internal/src/main/java/org/ballerinalang/langlib/internal/WorkerChannels.java
@@ -21,6 +21,7 @@
 import io.ballerina.runtime.internal.scheduling.Scheduler;
 import io.ballerina.runtime.internal.scheduling.Strand;
 import io.ballerina.runtime.internal.scheduling.WorkerDataChannel;
+import io.ballerina.runtime.internal.values.ChannelDetails;
 
 import java.util.Objects;
 
@@ -40,9 +41,20 @@ public static void autoClose(BString[] channelIds) {
         Strand currentStrand = Scheduler.getStrand();
         Strand channelHoldingStrand = Objects.requireNonNullElse(currentStrand.parent, currentStrand);
         for (BString channelId : channelIds) {
-            String channelName = channelId.getValue() + ":" + (channelHoldingStrand.functionInvocation - 1);
+            String channelName = getMatchingChannelName(channelId.getValue(), currentStrand);
             WorkerDataChannel workerDataChannel = channelHoldingStrand.wdChannels.getWorkerDataChannel(channelName);
             workerDataChannel.autoClose();
         }
     }
+
+    private static String getMatchingChannelName(String channelId, Strand currentStrand) {
+        String channelName = null;
+        for (ChannelDetails channelDetail : currentStrand.channelDetails) {
+            if (channelDetail.name.contains(channelId)) {
+                channelName = channelDetail.name;
+                break;
+            }
+        }
+        return channelName;
+    }
 }
diff --git a/tests/jballerina-unit-test/src/test/java/org/ballerinalang/test/worker/BasicWorkerTest.java b/tests/jballerina-unit-test/src/test/java/org/ballerinalang/test/worker/BasicWorkerTest.java
@@ -131,7 +131,8 @@ public void testWorkerMessagePassing(String funcName) {
     public Object[] workerMessagePassingFunctions() {
         return new Object[]{
                 "testWorkerMessagePassingRepeatedly",
-                "testPanicWithMessagePassing"
+                "testPanicWithMessagePassing",
+                "testEarlyReturnWithMessagePassing"
         };
     }
 

diff --git a/...rina-unit-test/src/test/java/org/ballerinalang/test/worker/WorkerConditionalSendTest.java b/...rina-unit-test/src/test/java/org/ballerinalang/test/worker/WorkerConditionalSendTest.java
@@ -67,7 +67,8 @@ public static String[] functionProvider() {
                 "multipleReceiveConditional",
                 "multipleReceiveWithNonConditionalSend",
                 "testNonTopLevelSend",
-                "testSendWithEarlyReturnError"
+                "testSendWithEarlyReturnError",
+                "testWorkerEarlyReturnWithinIf"
         };
     }
 

diff --git a/tests/jballerina-unit-test/src/test/resources/test-src/workers/basic-worker-actions.bal b/tests/jballerina-unit-test/src/test/resources/test-src/workers/basic-worker-actions.bal
@@ -15,6 +15,7 @@
 // under the License.
 
 import ballerina/jballerina.java;
+import ballerina/lang.'error as errorLib;
 import ballerina/lang.'value as value;
 import ballerina/test;
 
@@ -389,6 +390,32 @@ function foo() {
     }
 }
 
+function testEarlyReturnWithMessagePassing() {
+    foreach int i in 0 ... 100 {
+        func1();
+        func2();
+    }
+}
+
+function func2() {
+    func1();
+}
+
+function func1() {
+    foo();
+    worker w1 {
+        if (1 > 0) {
+            return;
+        } else {
+            10 -> w2;
+        }
+    }
+
+    worker w2 {
+        int|errorLib:NoMessage y = <- w1;
+    }
+}
+
 function testWorkerMessagePassingRepeatedly() {
 
     future<()> futureResult1 = @strand {thread: "any"} start testSyncSendRecursively(0, 10000);

diff --git a/tests/jballerina-unit-test/src/test/resources/test-src/workers/workers_conditional_send.bal b/tests/jballerina-unit-test/src/test/resources/test-src/workers/workers_conditional_send.bal
@@ -475,3 +475,103 @@ function testSendWithEarlyReturnError() {
     test:assertEquals(mapResult["a"], false, "Invalid boolean result");
     test:assertEquals(mapResult["b"], 2, "Invalid int result");
 }
+
+public function testWorkerEarlyReturnWithinIf() {
+    int _ = foo(1);
+    worker w1 {
+        boolean b = true;
+        if b {
+            return;
+        } else {
+            30 -> w4;
+            if (b) {
+                return;
+            } else {
+                40 -> w4;
+                if (b) {
+                    return;
+                } else {
+                    50 -> w5;
+                }
+            }
+        }
+    }
+
+    worker w2 {
+        boolean b = true;
+        if (b) {
+            return;
+        } else {
+            60 -> w4;
+        }
+    }
+
+    worker w3 {
+        boolean b = true;
+        if (b) {
+            if (b) {
+                return;
+            } else {
+                70 -> w4;
+            }
+        } else if (b) {
+            80 -> w5;
+        } else {
+            90 -> w4;
+            if (b) {
+                return;
+            } else {
+                100 -> w5;
+            }
+        }
+    }
+
+    worker w4 {
+        map<int|errorLib:NoMessage> m = <- {w1, w2, w3};
+        test:assertTrue(m["w1"] is error);
+        error err = <error> m["w1"];
+        test:assertEquals(err.message(), "NoMessage", "Invalid error message");
+        test:assertEquals(err.detail().toString(), "{\"message\":\"no message received from worker 'w1' to worker 'w4'\"}", "Invalid error detail");
+        test:assertTrue(m["w2"] is error);
+        err = <error> m["w2"];
+        test:assertEquals(err.message(), "NoMessage", "Invalid error message");
+        test:assertEquals(err.detail().toString(), "{\"message\":\"no message received from worker 'w2' to worker 'w4'\"}", "Invalid error detail");
+
+        int|errorLib:NoMessage res1 = <- w1;
+        test:assertTrue(res1 is error);
+        error err1 = <error> res1;
+        test:assertEquals(err1.message(), "NoMessage", "Invalid error message");
+        test:assertEquals(err1.detail().toString(), "{\"message\":\"no message received from worker 'w1' to worker 'w4'\"}", "Invalid error detail");
+
+        int|errorLib:NoMessage res2 = <- w3;
+        test:assertTrue(res2 is error);
+        error err2 = <error> res2;
+        test:assertEquals(err2.message(), "NoMessage", "Invalid error message");
+        test:assertEquals(err2.detail().toString(), "{\"message\":\"no message received from worker 'w3' to worker 'w4'\"}", "Invalid error detail");
+    }
+
+    worker w5 {
+        map<int|errorLib:NoMessage> m = <- {w1, w3};
+        test:assertTrue(m["w1"] is error);
+        error err = <error> m["w1"];
+        test:assertEquals(err.message(), "NoMessage", "Invalid error message");
+        test:assertEquals(err.detail().toString(), "{\"message\":\"no message received from worker 'w1' to worker 'w5'\"}", "Invalid error detail");
+        test:assertTrue(m["w3"] is error);
+        err = <error> m["w3"];
+        test:assertEquals(err.message(), "NoMessage", "Invalid error message");
+        test:assertEquals(err.detail().toString(), "{\"message\":\"no message received from worker 'w3' to worker 'w5'\"}", "Invalid error detail");
+
+        int|errorLib:NoMessage res1 = <- w3;
+        test:assertTrue(res1 is error);
+        error err1 = <error> res1;
+        test:assertEquals(err1.message(), "NoMessage", "Invalid error message");
+        test:assertEquals(err1.detail().toString(), "{\"message\":\"no message received from worker 'w3' to worker 'w5'\"}", "Invalid error detail");
+    }
+
+    wait w4;
+    wait w5;
+}
+
+function foo(int i) returns int {
+    return i;
+}