The underlying layer uses C+Openmp multithreading to process function callback calculations. How to link through Dart ffi

[PythonnotJava]

The underlying layer uses C+Openmp multithreading to process function callback calculations. How to link through Dart ffi

Following is my C-lang code

void testOmpCanRun(int row, int column, double ** data, void (*func)(int, int, double ** data)){
    // The data has been allocated memory before.
    #pragma omp parallel for collapse(2)
    for (int r = 0;r < row;r++) {
        for (int c = 0;c < column;c ++) {
            func(r, c, data);
        }
    }
}

Following is my Dart-lang code

typedef testOmpCanRun__base__ffi = Void Function(
    Int32 row,
    Int32 column,
    Pointer<Pointer<Double>> data,
    Pointer<NativeFunction<Void Function(Int32, Int32, Pointer<Pointer<Double>> data)>> func
);
typedef testOmpCanRun__base = void Function(
    int row,
    int column,
    Pointer<Pointer<Double>> data,
    Pointer<NativeFunction<Void Function(Int32, Int32, Pointer<Pointer<Double>> data)>> func
);
late final testOmpCanRun__base matply__testOmpCanRun = dylib.lookupFunction<testOmpCanRun__base__ffi, testOmpCanRun__base>('testOmpCanRun');

// Question core 
// Using NativeCallable<T>.listener
// MatrixType is a Matrix-like data-structure，main-component is like as follwed -
/*
  - int row;
  - int column;
  - Pointer<Pointer<Double>> data;
* */
typedef CallBackUp = Void Function(Int32, Int32, Pointer<Pointer<Double>>);
MatrixType justTestOmpCanRun({
  required int row,
  required int column,
  required double Function(int, int) func,  
}) {
  assert(row > 0 && column > 0);
  Pointer<Pointer<Double>> data = matply__allocateButNoNumbers(row, column);  // Just allocate memory
  final NativeCallable<CallBackUp> callback = NativeCallable<CallBackUp>.listener((int r, int c, Pointer<Pointer<Double>> data){
    data[r][c] = func(r, c);  // Wrap it as a void function according to the documentation
  });
  matply__testOmpCanRun(row, column, data, callback.nativeFunction);
  callback.close();
  return MatrixType.__fromDataPointer(data, [row, column]);  
}

Test && Output

double f(int r, int c){
  print("Call");
  print(r - c);
  return 3;
}

main() {
  final mt = justTestOmpCanRun(row: 2, column: 8, func: f);
  mt.visible();
  freeMp(visible: true, hex: false); // just free memory.
}

/*output
[
    [0.00000	 0.00000	 0.00000	 0.00000	 0.00000	 0.00000	 0.00000	 0.00000	]
    [0.00000	 0.00000	 0.00000	 0.00000	 0.00000	 0.00000	 0.00000	 0.00000	]
]  
Free Memory in location : 2535398893120
 */

As the results are shown , Matrix datas are not changed, keeping initial all-zeros.

Due to the lack of cases and information, I don't konw how to connect openmp multithreaded callback function to Dart ffi (single thread has been solved, and openmp multithreaded non-function callback test was successful).

[dart-github-bot]

Summary: User struggles to pass a Dart callback function to a C function using Dart FFI when the C function uses OpenMP for multithreading. The callback isn't executed correctly within the multithreaded C code.

[HosseinYousefi]

@dcharkes

[liamappelbe]

@PythonnotJava You're creating the listener callback, then calling matply__testOmpCanRun, then immediately closing the callback. Does matply__testOmpCanRun block until the computation is complete? Even if matply__testOmpCanRun does block, the listener callback doesn't. So it's possible that matply__testOmpCanRun runs those computations, then the results are sent to the listener, then matply__testOmpCanRun returns and you close the listener before it receives the results. Calls sent to closed listeners are simply ignored.

Have you tried putting a print in the listener, to see if it's being invoked? As a cheesy workaround (just to verify that this is actually the issue), try await Future.delayed(const Duration(seconds: 3)); before you close the callback. If that fixes the issue then you can try coming up with a better approach to decide when to close the callback (eg if you know how many times it should be invoked, keep a count, and close it when the count reaches the expected number).

Or maybe you don't need to close the callback at all? If you're only creating a fixed number of callbacks over the lifetime of your app, you can just set them to not keep their isolate alive, and leave them un-closed.

[PythonnotJava]

Thank you for your reply, it really works.
I have made the following changes.

Future<MatrixType> justTestOmpCanRun({
  required int row,
  required int column,
  required double Function(int, int) func,
}) async {
  assert(row > 0 && column > 0);
  Pointer<Pointer<Double>> data = matply__allocateButNoNumbers(row, column);  // Just allocate memory

  final NativeCallable<CallBackUp> callback = NativeCallable<CallBackUp>.listener((int r, int c, Pointer<Pointer<Double>> data) {
    data[r][c] = func(r, c);  // Wrap it as a void function according to the documentation
  });
  matply__testOmpCanRun(row, column, data, callback.nativeFunction);
  // Add a delay to allow OpenMP computation to complete before closing the callback
  await Future.delayed(const Duration(microseconds:10), () {
    callback.close();
  });
  return MatrixType.__fromDataPointer(data, [row, column]);
}

[liamappelbe]

Glad you got it working, but I'd recommend finding another way to decide when to close the callback, rather than just waiting for a fixed amount of time. If you wait too long, it wastes time, but if you wait too short, you risk losing some of the results.

I'm guessing that the callback is called once per matrix cell? In that case you should expect it to run row * column times. So you could wait until its called that many times instead:

Future<MatrixType> justTestOmpCanRun({
  required int row,
  required int column,
  required double Function(int, int) func,
}) async {
  assert(row > 0 && column > 0);
  Pointer<Pointer<Double>> data = matply__allocateButNoNumbers(row, column);  // Just allocate memory

  var calls = 0;
  final running = Completer();
  final NativeCallable<CallBackUp> callback = NativeCallable<CallBackUp>.listener((int r, int c, Pointer<Pointer<Double>> data) {
    data[r][c] = func(r, c);  // Wrap it as a void function according to the documentation
    ++calls;
    if (calls == (row * column)) {
      running.complete();
    }
  });
  matply__testOmpCanRun(row, column, data, callback.nativeFunction);
  // Add a delay to allow OpenMP computation to complete before closing the callback
  await running.future;
  callback.close();
  return MatrixType.__fromDataPointer(data, [row, column]);
}

[PythonnotJava]

Thank you for your solution. I tested your code this time and the previous code and found that as the order of magnitude increases (because the purpose of using omp is to process a large amount of data), the time taken for C+Openmp callback function calculation is much longer than using Dart asynchronous (the code is as follows). Perhaps it is because Dart's AOT optimization performance is better. And perhaps my choice of C+Openmp to handle callback multithreading is not a good solution.

Future<MatrixType> confront({
  required int row,
  required int column,
  required double Function(int, int) func
}) async {
  assert (row > 0 && column > 0);
  List<Future<void>> futures = [];
  final data = matply__allocateButNoNumbers(row, column);
  for (int r = 0; r < row; r++) {
    futures.add(Future(() {
      for (int c = 0; c < column; c ++) {
        data[r][c] = func(r, c);
      }
    }));
  }
  await Future.wait(futures);
  return MatrixType.__fromDataPointer(data, [row, column]);
}

[liamappelbe]

Either way you write it, it's going to be single threaded. Whether you use a NativeCallable.listener or a Future, the Dart code inside will run in the same isolate that creates the callback. So neither solution is multithreaded.

Your second solution is faster because it has less boilerplate running in between callbacks (a single awaited future, vs a call to a native library that spawns threads and sends messages back to the original isolate), and also because you're sharding it less (in the first solution you had one callback per matrix cell, and in the second it's a callback per row).

If your goal is to run a small function per matrix cell, then it would be even faster to just write the whole thing using an ordinary for loop with no async code. If you really want to run per-cell functions on multiple threads, Dart interop probably isn't the right solution.