Generic queued emitter for threading

[ahayzen-kdab]

Aims

In the current API we have an "UpdateRequester", this specifically calls handle_update_request on the Rust side via a queued metamethod call (which means that the UpdateRequester is thread safe and can be move across to other threads).

Instead this could be made generic so that we can remove our special trait.

Solution

Instead have a QueuedEmitter which takes an invokable name (and potentially parameters?), then the developer can call arbitrary invokables from background threads and not forced to use our API.

impl MyObject {
  #[invokable]
  fn start_thread(&mut self) {
    // we need to check that this is a valid invokable name in the C++ side, if it isn't we return Err instead of Ok ?
    // or a nullptr on the UniquePtr<QueuedEmitter> ?
    let mut queued_emitter = self.cpp.queued_emitter("sync_method");

    // in another thread
    if let Ok(queued_emitter) = queued_emitter {
        queued_emitter.emit();
    }
  }

  #[invokable]
  fn sync_method(&mut self) {
    // perform tasks on GUI thread
  }
}

Problems

• How can we ensure that the invokable name is valid in queued_emitter(method_name) ? do we check when creating the object on the C++ side and return a Result ? Do we require you to use a function pointer on the Rust side and somehow check that the function is an invokable ?
• Another option would be, could we have a Closure on the Rust side that is stored and called?

Closure route

With the closure route the closure would have to take the fat pointer (or CppObj + RustObj) as a parameter and not move the self one into it, as otherwise there will be problems with locking and holding onto references.

impl MyObject {
  #[invokable]
  fn start_thread(&mut self) {
    let mut queued_emitter = self.cpp.queued_emitter(|fat_pointer| {
        // perform tasks on the GUI thread
    });

    // in another thread
    queued_emitter.emit();
  }
}

[ahayzen-kdab]

Should this be called QueuedEmitter or QueuedCall or something else ? As it's not emitting signals, but calling a public invokable/method via a queued connection.

[ahayzen-kdab]

Stepping back, should we call an invokable with it or should we call a signal ?

As a signal, then we would need a connection in C++/QML to connect the "ready" from the background thread to the "sync" method on the Rust side.

As an invokable, then we don't need the connection on the C++/QML side, install the background thread calls the "sync" method.

Or could we have both a QueuedEmitter and QueuedCall, how do we make these safe syntax wise to fail at compile time ? and can we pass arguments in any way through them ?

[LeonMatthesKDAB]

I think having a QueuedEmitter that actually emits signals is something we'll definitely want to have.
Simply being able to notify the user of events that happened in a background thread would be very valuable.

[LeonMatthesKDAB]

That said, being able to execute code on the GUI thread is also very valuable, so we'll likely want to have both.
But we probably don't need to have these be a separate type, as all they really are are wrappers over a QPointer.
So we could hopefully just have a single "QueuedEmitter" or maybe just call it GUIThread with two methods:

1. emit or emit_queued
2. run or execute
   Which would make for a relatively nice syntax:

let qt_thread = self.cpp.qt_thread();
qt_thread.run(|obj| obj.set_prop());
qt_thread.emit(...);

[ahayzen-kdab]

Right, the emit name should match whatever we have for emitting signals, so currently emit_queued and the run or execute maybe needs to indicate that it's "later" and queued onto the GUI thread, so like run_queued.

[ahayzen-kdab]

We can potentially pass a function pointer (lambda) over CXX via function pointers https://cxx.rs/binding/fn.html

[ahayzen-kdab]

As discussed trying to make the call from the Rust background thread totally safe is tricky. The following C++ code shows how we could lock objects enough to stop delete obj; from breaking the run or execute call from the QueuedEmitter obj. However this doesn't stop free(obj) being called.

#include <atomic>
#include <functional>
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>

using namespace std;

class MyObject;

class ThreadObj
{
public:
  void loadClosures(MyObject* obj)
  {
    // Take the list of closures
    std::vector<std::function<void()>> vec = {};
    {
      std::lock_guard<std::mutex> guard(m_mutex);
      std::swap(vec, m_closures);
    }

    // Add each closure to the Qt event loop (here we fake it)
    // but this is "safe" as we hold a lock on the mutex in MyObject
    for (auto it = vec.cbegin(); it != vec.cend(); it++) {
      // QMetaObject::invokeMethod(obj, (*it), Qt::QueuedConnection);
      (*it)();
    }
  }

  void queue(std::function<void()> closure)
  {
    // Queue the closure onto the event loop
    std::lock_guard<std::mutex> guard(m_mutex);
    m_closures.push_back(closure);
  }

private:
  std::vector<std::function<void()>> m_closures;
  std::mutex m_mutex;
};

class MyObject
{
public:
  MyObject()
    : m_running(true)
    , m_threadEventLoop(&MyObject::eventLoop, this)
  {}
  ~MyObject()
  {
    // Stop the fake event loop
    m_running = false;
    m_threadEventLoop.join();

    // Stop the obj
    m_mutex.lock();
  }

  int add(int a, int b) { return a + b; }

  void slowMethod()
  {
    std::lock_guard<std::mutex> guard(m_mutex);
    for (int i = 0; i < 10; i++) {
      this_thread::sleep_for(10ms);

      // Create a queued closure that will run on the event loop
      auto thread = threadObj();
      thread->queue([]() { std::cout << "Queued" << std::endl; });

      m_count = add(m_count, 1);
      std::cout << "Count: " << m_count << std::endl;
    }
  }

  void event()
  {
    // Load any closures from child thread objects
    std::lock_guard<std::mutex> guard(m_mutex);
    for (auto it = m_threadObjs.begin(); it != m_threadObjs.end(); it++) {
      (*it)->loadClosures(this);
    }
  }

  void eventLoop()
  {
    while (m_running) {
      event();
      this_thread::sleep_for(10ms);
    }
  }

  std::shared_ptr<ThreadObj> threadObj()
  {
    auto obj = std::make_shared<ThreadObj>();
    m_threadObjs.push_back(obj);
    return obj;
  }

private:
  int m_count = 0;
  std::mutex m_mutex;
  std::atomic_bool m_running;
  std::thread m_threadEventLoop;
  std::vector<std::shared_ptr<ThreadObj>> m_threadObjs;
};

void
deconstruct(MyObject* obj)
{
  this_thread::sleep_for(50ms);
  // Attempt to delete the object while in a method
  std::cout << "About to delete" << std::endl;
  // FIXME: we can't stop free though
//   free(obj);
  delete obj;
  std::cout << "Deleted obj" << std::endl;
}

int
main(void)
{
  //   MyObject* obj = (MyObject*)malloc(sizeof(MyObject*));
  MyObject* obj = new MyObject();

  std::thread threadSlow(&MyObject::slowMethod, obj);
  std::thread threadDeconstruct(deconstruct, obj);

  threadDeconstruct.join();
  threadSlow.join();

  return 0;
}

Then this outputs

Count: 1
Count: 2
Count: 3
Count: 4
About to delete
Count: 5
Count: 6
Count: 7
Count: 8
Count: 9
Count: 10
Queued
Queued
Queued
Queued
Queued
Queued
Queued
Queued
Queued
Queued
Deleted obj

[ahayzen-kdab]

The question is do we want to protect against the C++ potentially being deleted in the middle of a method, or do we consider that the developers problem as they are writing unsafe code in C++. And this would be a problem in a typical Qt code as well.

Or do we want to prove that any code written on the Rust side will never break, whatever you do on the C++ side ? (but I'm not sure all cases can be prevented due to things like free()).

Or do we want a typical QML application or GUI thread widgets app, to have always valid Rust code. But if you start having pointers to the QObject across threads in C++ then you are "on your own".

I guess i'm saying at what level do we want to claim things are "safe".

We should also inspect what CXX does, as i imagine they'll suffer the same issue if someone delete/free's the class that Rust is calling/being called on.

[ahayzen-kdab]

We now have the ability to add a function pointer to the event loop and then work in the C++ context inside that. So lets close this issue for now.