How can I convert this c++ snippet to kotlin using coroutines?

Support
1

I have the following:

// The coroutine executing on the current thread (if it is a coroutine thread)
static thread_local TestCoroutine* GThreadCoroutine = NULL;

TestCoroutine::TestCoroutine(TestCoroutineFunc func, void* ctx)
{
    CoroutineRunning = false;
    CoroutineTerminated = false;
    Thread = new std::thread([this, func, ctx]
    {
        // Set the thread coroutine
        GThreadCoroutine = this;
        // Wait for initial Run()
        while(1)
        {
            std::unique_lock<std::mutex> lock(StateMutex);
            if (CoroutineRunning)
                break;
            StateChange.wait(lock);
        }
        // Run user code, which will then call Yield() when it wants to yield control
        func(ctx);
        // Mark as terminated
        {
            std::lock_guard<std::mutex> lock(StateMutex);
            CoroutineTerminated = true;
            CoroutineRunning = false;
            StateChange.notify_all();
        }
    });
}

TestCoroutine::~TestCoroutine()
{
    IM_ASSERT(CoroutineTerminated); // The coroutine needs to run to termination otherwise it may leak all sorts of things and this will deadlock    
    if (Thread)
    {
        Thread->join();
        delete Thread;
        Thread = NULL;
    }
}

// Run the coroutine until the next call to Yield(). Returns TRUE if the coroutine yielded, FALSE if it terminated (or had previously terminated)
bool TestCoroutine::Run()
{
    // Wake up coroutine thread
    {
        std::lock_guard<std::mutex> lock(StateMutex);
        if (CoroutineTerminated)
            return false; // Coroutine has already finished
        CoroutineRunning = true;
        StateChange.notify_all();
    }
    // Wait for coroutine to stop
    while (1)
    {
        std::unique_lock<std::mutex> lock(StateMutex);
        if (!CoroutineRunning)
        {
            if (CoroutineTerminated)
                return false; // Coroutine finished
            break;
        }
        StateChange.wait(lock);
    }
    return true;
}

// Yield the current coroutine (can only be called from a coroutine)
void TestCoroutine::Yield()
{
    IM_ASSERT(GThreadCoroutine); // This can only be called from a coroutine thread
    GThreadCoroutine->YieldInternal();
}

// Yield the current coroutine (internal implementation)
void TestCoroutine::YieldInternal()
{
    // Flag that we are not running any more
    {
        std::lock_guard<std::mutex> lock(StateMutex);
        CoroutineRunning = false;
        StateChange.notify_all();
    }
    // Wait until we get started up again
    while (1)
    {
        std::unique_lock<std::mutex> lock(StateMutex);
        if (CoroutineRunning)
            break;
        StateChange.wait(lock);
    }
}

init:

        TestQueueCoroutine = new TestCoroutine([](void* ctx)
        {
            TestEngine* engine = (TestEngine*)ctx;
            while (!engine->TestQueueCoroutineShouldExit)
            {
                TestEngine_ProcessTestQueue(engine);
                TestCoroutine::Yield();
            }
        }, this);

deinit:

        if (TestQueueCoroutine != NULL)
        {
            // Run until the coroutine exits
            TestQueueCoroutineShouldExit = true;
            while (TestQueueCoroutine->Run());
            delete TestQueueCoroutine;
            TestQueueCoroutine = NULL;
        }

anchors:

    TestCoroutine*         TestQueueCoroutine = NULL;      // Coroutine to run the test queue

    bool                        TestQueueCoroutineShouldExit = false; // Flag to indicate that we are shutting down and the test queue coroutine should stop

Now, I tried to implement it using kotlin coroutines…

So far I did this:

// A coroutine function - ctx is an arbitrary context object
typealias TestCoroutineFunc = suspend (ctx: Any) -> Unit

val testCoroutineFunc: TestCoroutineFunc = { ctx: Any ->
    val engine = ctx as TestEngine
    while (!engine.testQueueCoroutineShouldExit) {
        engine.processTestQueue()
        yield()
    }
}

To run, simply:

testQueueCoroutine = GlobalScope.launch { testCoroutineFunc(this) }

The problems I can see are:

  • in kotlin, the coroutine isnt re-usable, so I need to create one everytime I want to run it

  • apparently I have no way to detect if the coroutines finished because it yielded or terminated, so I have no idea how I can rewrite the destroy function…

Any ideas?

2

Take a look at the Guide, especially the part about Structured Concurrency. The Job returned from launch has properties about if it is canceled or completed.