POSIX 线程与 Dart Isolate：在 FFI 中创建原生线程的上下文管理

大家好，今天我们要深入探讨一个复杂但非常有用的主题：在 Dart FFI 中创建原生线程，并管理其上下文。这个主题涉及到并发编程的底层细节，以及 Dart 如何与原生代码交互。理解它对于构建高性能、需要利用多核处理器能力的 Dart 应用至关重要。

为什么要在 Dart 中使用原生线程？

Dart 的 Isolate 是一个强大的并发模型，但它也有局限性。Isolate 之间的通信需要消息传递，这会带来一定的开销。在某些情况下，使用原生线程可能更合适：

避免 Isolate 的消息传递开销： 当需要频繁地在线程之间共享数据时，原生线程通过共享内存可以更高效。
与已有的原生代码集成： 有些库可能已经使用了原生线程，为了更好地集成，需要在 Dart 中创建并管理这些线程。
利用某些原生线程 API： 某些操作系统或平台提供了只能通过原生线程访问的 API。
CPU 密集型任务： 对于 CPU 密集型任务，原生线程可能能够更好地利用多核处理器，避免 Isolate 带来的上下文切换开销。

POSIX 线程 (pthreads) 简介

POSIX 线程 (pthreads) 是一个定义在 POSIX 标准（例如 IEEE Std 1003.1）中的线程 API。它提供了一组函数，用于创建、管理和同步线程。pthreads 在 Unix-like 系统（如 Linux、macOS）和一些其他平台上广泛使用。

Dart FFI 简介

Dart FFI (Foreign Function Interface) 允许 Dart 代码调用原生代码（如 C、C++）。这使得我们可以利用已有的原生库，或者编写性能关键的代码部分，并将其集成到 Dart 应用中。

创建原生线程的步骤

在 Dart FFI 中创建原生线程通常涉及以下步骤：

定义原生函数接口： 使用 dart:ffi 定义一个 Dart 函数签名，对应于原生函数。
加载原生库： 使用 DynamicLibrary.open() 加载包含原生函数的动态链接库。
获取函数指针： 使用 lookupFunction() 获取指向原生函数的指针。
调用原生函数创建线程： 调用原生函数，通常是 pthread_create()，来创建一个新的线程。
管理线程上下文： 在原生线程中管理 Dart VM 的上下文，确保 Dart 对象和函数可以安全地访问。
线程同步： 使用互斥锁、条件变量等机制，确保线程之间的同步和互斥。
清理资源： 在线程退出时，释放分配的资源。

代码示例：使用 pthreads 创建线程

下面是一个简单的例子，演示如何在 Dart 中使用 pthreads 创建一个线程，并在该线程中执行一些 Dart 代码。

1. 定义原生函数接口 (Dart):

import 'dart:ffi';
import 'dart:isolate';

// 定义原生函数的类型
typedef NativeThreadFunc = Int32 Function(Pointer<Void>);
typedef DartThreadFunc = Int32 Function(Handle);

// 定义 pthread_create 函数的类型
typedef PthreadCreateNative = Int32 Function(
    Pointer<IntPtr>, Pointer<Void>, NativeThreadFunc, Pointer<Void>);
typedef PthreadCreateDart = int Function(
    Pointer<Pointer<IntPtr>>, Pointer<Void>, Pointer<NativeFunction<NativeThreadFunc>>, Pointer<Void>);

class NativeLibrary {
  static DynamicLibrary lib = DynamicLibrary.open("libnative.so"); // 替换为你的库名称

  static final PthreadCreateDart pthread_create = lib
      .lookupFunction<PthreadCreateNative, PthreadCreateDart>('pthread_create');
}

2. 实现原生线程函数 (C/C++):

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h> // For intptr_t
#include <iostream>

#include "include/dart_api.h"
#include "include/dart_native_api.h"

// 全局 Dart VM 访问函数 (需要在使用前初始化)
Dart_Handle (*Dart_NewPersistentHandle_)(Dart_Handle obj);
void (*Dart_HandleFromPersistent_)(Dart_PersistentHandle handle, Dart_Handle* result);
void (*Dart_DeletePersistentHandle_)(Dart_PersistentHandle handle);
Dart_Handle (*Dart_Invoke_)(Dart_Handle target, Dart_Handle name, int argc, Dart_Handle* argv);
Dart_Handle (*Dart_Null_)(void);
Dart_Handle (*Dart_RootLibrary_)(void);
Dart_Handle (*Dart_GetClosure_)(Dart_Handle target, Dart_Handle name);
bool (*Dart_IsError_)(Dart_Handle handle);
const char* (*Dart_GetError_)(Dart_Handle handle);

// 初始化 Dart API 函数指针
bool initDartApi(void* data) {
  if (Dart_InitializeApiDL(data) != 0) {
    return false;
  }

  Dart_NewPersistentHandle_ = reinterpret_cast<Dart_Handle (*)(Dart_Handle)>(Dart_GetApiFunction(Dart_kNewPersistentHandle));
  Dart_HandleFromPersistent_ = reinterpret_cast<void (*)(Dart_PersistentHandle, Dart_Handle*)>(Dart_GetApiFunction(Dart_kHandleFromPersistent));
  Dart_DeletePersistentHandle_ = reinterpret_cast<void (*)(Dart_PersistentHandle)>(Dart_GetApiFunction(Dart_kDeletePersistentHandle));
  Dart_Invoke_ = reinterpret_cast<Dart_Handle (*)(Dart_Handle, Dart_Handle, int, Dart_Handle*)>(Dart_GetApiFunction(Dart_kInvoke));
  Dart_Null_ = reinterpret_cast<Dart_Handle (*)(void)>(Dart_GetApiFunction(Dart_kNull));
  Dart_RootLibrary_ = reinterpret_cast<Dart_Handle (*)(void)>(Dart_GetApiFunction(Dart_kRootLibrary));
  Dart_GetClosure_ = reinterpret_cast<Dart_Handle (*)(Dart_Handle, Dart_Handle)>(Dart_GetApiFunction(Dart_kGetClosure));
  Dart_IsError_ = reinterpret_cast<bool (*)(Dart_Handle)>(Dart_GetApiFunction(Dart_kIsError));
  Dart_GetError_ = reinterpret_cast<const char* (*)(Dart_Handle)>(Dart_GetApiFunction(Dart_kGetError));
  return true;
}

struct ThreadArgs {
  Dart_Port port;
  Dart_PersistentHandle function;
};

// 原生线程函数，接收一个 Dart 函数的 Handle
void* nativeThread(void* arg) {
  ThreadArgs* args = static_cast<ThreadArgs*>(arg);
  Dart_PersistentHandle persistentHandle = args->function;
  Dart_Port port = args->port;
  free(args); // 释放参数内存

  Dart_Handle dartFunction;
  Dart_HandleFromPersistent_(persistentHandle, &dartFunction);

  Dart_Handle result = Dart_Invoke(Dart_RootLibrary_(), Dart_NewStringFromCString("runInNativeThread"), 1, &dartFunction);

  if (Dart_IsError_(result)) {
      fprintf(stderr, "Error in native thread: %sn", Dart_GetError_(result));
  }

  Dart_DeletePersistentHandle_(persistentHandle);

  return NULL;
}

extern "C" {
  // 供 Dart 调用的函数，用于创建原生线程
  int pthread_create(pthread_t *thread, const pthread_attr_t *attr,
                     void *(*start_routine) (void *), void *arg) {
    return ::pthread_create(thread, attr, start_routine, arg);
  }

  // 初始化 Dart API，此函数必须在 Dart 代码调用任何其他 Dart API 之前调用
  bool initializeDartApi(void* data) {
    return initDartApi(data);
  }

  // 创建一个原生线程
  intptr_t create_native_thread(Dart_Port port, Dart_Handle function) {
    pthread_t thread;
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); // 设置为 detached 线程

    // 创建一个结构体来传递参数
    ThreadArgs* args = (ThreadArgs*)malloc(sizeof(ThreadArgs));
    args->port = port;
    args->function = Dart_NewPersistentHandle_(function);

    int result = pthread_create(&thread, &attr, nativeThread, args);
    pthread_attr_destroy(&attr);

    if (result != 0) {
        fprintf(stderr, "Failed to create thread: %dn", result);
        return -1;
    }
    return (intptr_t)thread;
  }

}

3. Dart 代码，调用原生函数创建线程 (Dart):

import 'dart:ffi';
import 'dart:isolate';
import 'dart:async';

import 'native_library.dart'; // 包含 NativeLibrary 的定义

// 声明原生函数
typedef CreateNativeThreadNative = Int64 Function(Int64 port, Handle function);
typedef CreateNativeThreadDart = int Function(int port, Object function);

typedef InitializeDartApiNative = Int8 Function(Pointer<Void> data);
typedef InitializeDartApiDart = int Function(Pointer<Void> data);

class NativeThreadWrapper {
  static final CreateNativeThreadDart createNativeThread = NativeLibrary.lib
      .lookupFunction<CreateNativeThreadNative, CreateNativeThreadDart>('create_native_thread');

  static final InitializeDartApiDart initializeDartApi = NativeLibrary.lib
      .lookupFunction<InitializeDartApiNative, InitializeDartApiDart>('initializeDartApi');

  static void runInNativeThread(Function function) {
    print("Running in native thread");
    function();
  }

  static Future<void> startNativeThread(Function function) async {
    final receivePort = ReceivePort();
    final sendPort = receivePort.sendPort.nativePort;

    // 初始化 Dart API
    final Pointer<Void> initFunctions = Pointer.fromFunction(NativeLibrary.lib.nativeLibrary.address);
    final int initResult = initializeDartApi(initFunctions);
    if (initResult == 0) {
      print("Failed to initialize Dart API in native code");
      return;
    }

    // 调用原生函数创建线程
    final int threadId = createNativeThread(sendPort, function);
    if (threadId == -1) {
      print("Failed to create native thread");
      return;
    }
    print("Native thread created with ID: $threadId");

    // 等待线程完成 (可选)
    //receivePort.listen((message) {
    //  print("Received message from native thread: $message");
    //  receivePort.close();
    //});
  }
}

void main() async {
  // 要在原生线程中执行的 Dart 函数
  void myDartFunction() {
    print("Hello from Dart function in native thread!");
    // 可以在这里执行其他 Dart 代码
    for (int i = 0; i < 5; i++) {
        print("Iteration: $i");
        await Future.delayed(Duration(milliseconds: 100));
    }
  }

  // 创建并启动原生线程
  await NativeThreadWrapper.startNativeThread(myDartFunction);

  print("Main function continues...");
  await Future.delayed(Duration(seconds: 1)); // 确保原生线程有时间运行
}

4. Makefile (可选，用于编译原生代码):

LIB_NAME = native
CC = g++
CFLAGS = -fPIC -std=c++17 -I./include  # 确保包含 dart_api.h 和 dart_native_api.h 的路径
LDFLAGS = -shared -lpthread

all: $(LIB_NAME).so

$(LIB_NAME).o: $(LIB_NAME).c include/dart_api.h include/dart_native_api.h
    $(CC) $(CFLAGS) -c $(LIB_NAME).c -o $(LIB_NAME).o

$(LIB_NAME).so: $(LIB_NAME).o
    $(CC) $(LDFLAGS) $(LIB_NAME).o -o $(LIB_NAME).so

clean:
    rm -f $(LIB_NAME).o $(LIB_NAME).so

重要注意事项:

Dart API 初始化: 必须在原生线程中使用 Dart API 之前初始化 Dart API。这通常通过 Dart_InitializeApiDL 函数完成。本例中通过initializeDartApi函数完成初始化.
线程安全: Dart VM 不是线程安全的。这意味着不能在多个原生线程中同时访问 Dart 对象或调用 Dart 函数。需要使用适当的同步机制（如互斥锁）来保护共享资源。
Handle: 需要在 Dart 和原生代码之间传递 Dart 对象时，使用 Dart_Handle 和 Dart_PersistentHandle。Dart_Handle 是一个临时引用，而 Dart_PersistentHandle 是一个持久引用，可以跨线程使用。务必在用完后删除 Dart_PersistentHandle 以避免内存泄漏。
异常处理: 原生线程中发生的 Dart 异常不会自动传播到 Dart 代码。需要在原生线程中捕获异常，并将其传递回 Dart 代码，例如通过 SendPort。
生命周期管理: 确保原生线程在 Dart VM 关闭之前退出。否则，可能会导致崩溃。
detached 线程: 例子中使用了 detached 线程, 意味着主线程不需要等待子线程结束，资源会在线程结束后自动释放。

上下文管理的挑战

在原生线程中使用 Dart VM 的上下文是一个复杂的问题。需要考虑以下几个方面：

线程安全: Dart VM 不是线程安全的。这意味着不能在多个原生线程中同时访问 Dart 对象或调用 Dart 函数。需要使用适当的同步机制（如互斥锁）来保护共享资源。
Handle: 需要在 Dart 和原生代码之间传递 Dart 对象时，使用 Dart_Handle。Dart_Handle 是一个临时引用，而 Dart_PersistentHandle 是一个持久引用，可以跨线程使用。
异常处理: 原生线程中发生的 Dart 异常不会自动传播到 Dart 代码。需要在原生线程中捕获异常，并将其传递回 Dart 代码，例如通过 SendPort。
生命周期管理: 确保原生线程在 Dart VM 关闭之前退出。否则，可能会导致崩溃。

线程同步机制

为了在原生线程和 Dart Isolate 之间安全地共享数据，需要使用适当的同步机制。常见的同步机制包括：

互斥锁 (Mutex): 用于保护共享资源，防止多个线程同时访问。
条件变量 (Condition Variable): 用于线程之间的信号传递和等待。
原子操作 (Atomic Operations): 用于对单个变量进行原子操作，避免竞争条件。

代码示例：使用互斥锁同步线程

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <iostream>

#include "include/dart_api.h"
#include "include/dart_native_api.h"

// 全局互斥锁
pthread_mutex_t mutex;

// 共享数据
int sharedData = 0;

// 原生线程函数
void* nativeThread(void* arg) {
  // 加锁
  pthread_mutex_lock(&mutex);

  // 访问共享数据
  sharedData++;
  printf("Native thread: sharedData = %dn", sharedData);

  // 解锁
  pthread_mutex_unlock(&mutex);

  return NULL;
}

extern "C" {
  // 供 Dart 调用的函数，用于创建原生线程
  intptr_t create_native_thread_with_mutex() {
    pthread_t thread;
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

    int result = pthread_create(&thread, &attr, nativeThread, NULL);
    pthread_attr_destroy(&attr);

    if (result != 0) {
      fprintf(stderr, "Failed to create thread: %dn", result);
      return -1;
    }
    return (intptr_t)thread;
  }

  // 初始化互斥锁
  int initialize_mutex() {
    return pthread_mutex_init(&mutex, NULL);
  }

  // 销毁互斥锁
  int destroy_mutex() {
    return pthread_mutex_destroy(&mutex);
  }
}

在 Dart 代码中，需要调用 initialize_mutex() 在程序启动时初始化互斥锁，并在程序退出时调用 destroy_mutex() 销毁互斥锁。

表格：Isolate vs. 原生线程

特性	Dart Isolate	原生线程
内存模型	独立堆，消息传递	共享堆
线程安全	是 (Isolate 内部)	否
上下文切换开销	较高	较低
适用场景	并发，隔离性要求高	并行，共享内存需求高
复杂性	较低	较高

一些更高级的用法

线程池： 可以创建一个线程池来管理多个原生线程，提高资源利用率。
异步操作： 可以使用原生线程来执行异步操作，避免阻塞 Dart UI 线程。
GPU 计算： 可以使用原生线程来调用 GPU 计算 API，加速计算密集型任务。

总结

我们讨论了如何在 Dart FFI 中创建和管理原生线程，重点介绍了 pthreads API 的使用、Dart API 的初始化、Handle 的管理、线程同步机制以及上下文管理的关键挑战。理解这些概念对于构建高性能的 Dart 应用至关重要。

POSIX 线程与 Dart Isolate：在 FFI 中创建原生线程的上下文管理

发表回复 取消回复

发表回复取消回复