Strands get blocked under high load during the concurrent access of the Pipe package

Description: When using the xlibb/pipe module to produce and consume messages under high load, Ballerina strands get blocked, causing the code to hang and preventing it from completing execution.

Steps to reproduce:

Run the following Ballerina service

import ballerina/http;
import ballerina/io;

import xlibb/pipe;

isolated class Hello {
    private final pipe:Pipe pipe = new (100000000);

    isolated function process() returns pipe:Error? {
        int i = 0;
        while i < 999999 {
            _ = check self.pipe.produce(string `Value ${i}`, -1);
            if i == 999998 {
                io:println("All values are published");
            }
            i += 1;
        }
    }

    isolated function consume() {
        int i = 0;
        while i < 999999 {
            string|error value = self.pipe.consume(-1);
            if i == 999998 {
                io:println("Last Element: ", value);
            }
            i += 1;
        }
    }
}

service / on new http:Listener(8080) {
    isolated resource function get test() returns http:Ok & readonly {
        Hello hello = new;
        _ = start hello.consume();
        _ = start hello.process();
        return http:OK;
    }
}

Invoke the resource the resource method to execute consume() and produce() APIs

curl "http://localhost:8080/test"

ballerina-platform / ballerina-library

Strands get blocked under high load during the concurrent access of the Pipe package #6812