Node FFI Callback from C++ thread

Question

I ran into a frustrating problem today. I'm working with node-ffi to run C++ code within my electron application. In general I've had good experiences, but I started working with multithreading today and ran into some difficulty. The ffi callback that I pass in is called from the thread just fine. However when I end my loop and try to join the loop thread to the main thread it completely freezes the electron app.

Full disclaimer: I'm pretty new to C++, and would appreciate any feedback on my code to improve it, especially any red flags you think I should be aware of.

Here are two repos that demonstrate the error I ran into:
Electron Project - https://github.com/JakeDluhy/threading-test
C++ DLL - https://github.com/JakeDluhy/ThreadedDll

And here's an overview of what I'm doing:
In my dll, I expose functions to begin/end a session and start/stop streaming. These call the reference of a class instance to actually implement the functionality. Essentially, it's a C wrapper around the more powerful C++ class.

// ThreadedDll.h
#pragma once

#ifdef __cplusplus
extern "C" {
#endif

#ifdef THREADEDDLL_EXPORTS
#define THREADEDDLL_API __declspec(dllexport)
#else
#define THREADEDDLL_API __declspec(dllimport)
#endif
    THREADEDDLL_API void beginSession(void(*frameReadyCB)());
    THREADEDDLL_API void endSession();

    THREADEDDLL_API void startStreaming();
    THREADEDDLL_API void stopStreaming();
#ifdef __cplusplus
}
#endif

// ThreadedDll.cpp
#include "ThreadedDll.h"
#include "Threader.h"

static Threader *threader = NULL;

void beginSession(void(*frameReadyCB)())
{
    threader = new Threader(frameReadyCB);
}

void endSession()
{
    delete threader;
    threader = NULL;
}

void startStreaming()
{
    if (threader) threader->start();
}

void stopStreaming()
{
    if (threader) threader->stop();
}

Here's what the Threader class looks like:

// Threader.h
#pragma once

#include <thread>
#include <atomic>

using std::thread;
using std::atomic;

class Threader
{
public:
    Threader(void(*frameReadyCB)());
    ~Threader();

    void start();
    void stop();
private:
    void renderLoop();

    atomic<bool> isThreading;
    void(*frameReadyCB)();
    thread myThread;
};

// Threader.cpp
#include "Threader.h"

Threader::Threader(void(*frameReadyCB)()) :
    isThreading{ false },
    frameReadyCB{ frameReadyCB }
{
}


Threader::~Threader()
{
    if (myThread.joinable()) myThread.join();
}

void Threader::start()
{
    isThreading = true;

    myThread = thread(&Threader::renderLoop, this);
}

void Threader::stop()
{
    isThreading = false;

    if (myThread.joinable()) myThread.join();
}

void Threader::renderLoop()
{
    while (isThreading) {
        frameReadyCB();
    }
}

And then here's my test javascript that uses it:

// ThreadedDll.js
const ffi = require('ffi');
const path = require('path');

const DllPath = path.resolve(__dirname, '../dll/ThreadedDll.dll');
// Map the library functions in the way that FFI expects
const DllMap = {
    'beginSession':     [ 'void', [ 'pointer' ] ],
    'endSession':       [ 'void', [] ],

    'startStreaming':   [ 'void', [] ],
    'stopStreaming':    [ 'void', [] ],
};
// Create the Library using ffi, the DLL, and the Function Table
const DllLib = ffi.Library(DllPath, DllMap);

class ThreadedDll {
    constructor(args) {
        this.frameReadyCB = ffi.Callback('void', [], () => {
            console.log('Frame Ready');
        });

        DllLib.beginSession(this.frameReadyCB);
    }

    startStreaming() {
        DllLib.startStreaming();
    }

    stopStreaming() {
        DllLib.stopStreaming();
    }

    endSession() {
        DllLib.endSession();
    }
}

module.exports = ThreadedDll;

// app.js
const ThreadedDll = require('./ThreadedDll');

setTimeout(() => {
    const threaded = new ThreadedDll();
    console.log('start stream');
    threaded.startStreaming();

    setTimeout(() => {
        console.log('stop stream');
        threaded.stopStreaming();
        console.log('end session');
        threaded.endSession();
    }, 1000);
}, 2000);

And it is in app.js that the main electron process runs. I would expect to see

start stream
Frame Ready (3800)
stop stream
end session

But it shows no end session. However if I remove the line frameReadyCB() within the c++ it works as expected. So somehow the ffi callback reference is screwing up the multithreading environment. Would love to get some thoughts on this. Thanks!

Answer 1

Problem

Your application is deadlocked. In your example, you have two threads:

1. thread-1 - created when you run $ npm start, and
2. thread-2 - created in Threader::start().

In thread-2, you call frameReadyCB(), which is going to block the thread until it has completed. A previous answer shows the callback will get executed on thread-1.

Unfortunately, thread-1 is already busy with the second setTimeout, calling stopStreaming(). Threader::stop attempts to join thread-2, blocking until thread-2 has completed.

You are now deadlocked. thread-2 is waiting for thread-1 to execute the callback, and thread-1 is waiting for thread-2 to complete execution. They are both waiting on each other.

Solution via node-ffi

It seems node-ffi handles the callbacks running on a separate thread when the thread is created via node-ffi using async(). So, you can remove the threading from your C++ library, and instead call DllLib.startStreaming.async(() => {}) from your node library.

Solution via C++

In order to solve this, you need to ensure you never try to join thread-2 while it's waiting for frameReadyCB() to complete. You can do this using a mutex. Also, you need to make sure you don't wait on locking the mutex while thread-2 is waiting for frameReadyCB(). The only way to do this is to create another thread to stop streaming. The example below does this using node-ffi async, although it could be done within the C++ library to hide this from your node library.

// Threader.h
#pragma once

#include <thread>
#include <atomic>

using std::thread;
using std::atomic;
using std::mutex;

class Threader
{
public:
    Threader(void(*frameReadyCB)());
    ~Threader();

    void start();
    void stop();
private:
    void renderLoop();

    atomic<bool> isThreading;
    void(*frameReadyCB)();
    thread myThread;
    mutex mtx;
};

// Threader.cpp
#include "Threader.h"

Threader::Threader(void(*frameReadyCB)()) :
    isThreading{ false },
    frameReadyCB{ frameReadyCB }
{
}


Threader::~Threader()
{
    stop();
}

void Threader::start()
{
    isThreading = true;

    myThread = thread(&Threader::renderLoop, this);
}

void Threader::stop()
{
    isThreading = false;

    mtx.lock();
    if (myThread.joinable()) myThread.join();
    mtx.unlock();
}

void Threader::renderLoop()
{
    while (isThreading) {
        mtx.lock();
        frameReadyCB();
        mtx.unlock();
    }
}

// ThreadedDll.js
const ffi = require('ffi');
const path = require('path');

const DllPath = path.resolve(__dirname, '../dll/ThreadedDll.dll');
// Map the library functions in the way that FFI expects
const DllMap = {
    'beginSession':     [ 'void', [ 'pointer' ] ],
    'endSession':       [ 'void', [] ],

    'startStreaming':   [ 'void', [] ],
    'stopStreaming':    [ 'void', [] ],
};
// Create the Library using ffi, the DLL, and the Function Table
const DllLib = ffi.Library(DllPath, DllMap);

class ThreadedDll {
    constructor(args) {
        this.frameReadyCB = ffi.Callback('void', [], () => {
            console.log('Frame Ready');
        });

        DllLib.beginSession(this.frameReadyCB);
    }

    startStreaming() {
        DllLib.startStreaming();
    }

    stopStreaming() {
        DllLib.stopStreaming.async(() => {});
    }

    endSession() {
        DllLib.endSession.async(() => {});
    }
}

module.exports = ThreadedDll;