高端上海网站设计公司价格,品牌营销案例,贵阳网站建设公司排名,可以做科学模拟实验的网站Dart IO 源码剖析
许多Flutter新手#xff0c;特别是安卓、iOS原生开发转做Flutter的小伙伴#xff0c;一直对Flutter 单线程模型开发APP倍感不解#xff0c;他们总是喜欢本能的把网络请求、文件读写放到一个单独线程去做#xff0c;因为“耗时操作会阻塞UI线程嘛”。于是…Dart IO 源码剖析
许多Flutter新手特别是安卓、iOS原生开发转做Flutter的小伙伴一直对Flutter 单线程模型开发APP倍感不解他们总是喜欢本能的把网络请求、文件读写放到一个单独线程去做因为“耗时操作会阻塞UI线程嘛”。于是我看到有人把这些所谓耗时代码放到一个单独的Isolate中去做美其名曰优化性能提升帧率殊不知这是耗费资源降低性能。因为Isolate是内存隔离的它比操作系统线程要更重与其说它是Dart的线程不如说它更像进程。当你在两个Isolate之间通信时涉及内存的拷贝频繁的交互反而降低Dart 主隔离root isloate的性能这也是官方并不太推荐你在非计算密集型任务中创建子隔离的原因。
虽然大家都知道Flutter中不用创建单独的隔离去发起网络IO但是并没有资料详细解释为什么不需要今天我们就通过剖析Dart VM底层源码详细了解Dart IO的底层原理。
关于Dart IO 源码剖析我会用两篇文章来介绍本章以剖析文件IO为主下一篇我们剖析网络IO。发车了请系好安全带
文件IO
Dart 侧
在Dart中我们一般可以使用下面的代码将二进制数据写入到一个文件中
File(test.bin).writeAsBytes([96,97]);接下来我们就沿着这条调用链详细研究一下当在Dart 上层写文件时Dart VM到底发生了什么。
abstract interface class File implements FileSystemEntity {
...factory File(String path) {final IOOverrides? overrides IOOverrides.current;if (overrides null) {return new _File(path);}return overrides.createFile(path);}...FutureFile writeAsBytes(Listint bytes,{FileMode mode FileMode.write, bool flush false});
}由于File类是一个抽象接口并没有writeAsBytes方法的具体实现但是我们通过它的工厂构造方法可知其具体实现的子类是_File我们直接找到file_impl.dart文件查看_File源码 FutureFile writeAsBytes(Listint bytes,{FileMode mode FileMode.write, bool flush false}) {return open(mode: mode).then((file) {return file.writeFrom(bytes, 0, bytes.length).thenFile((_) {if (flush) return file.flush().then((_) this);return this;}).whenComplete(file.close);});}这里它又调用了内部的open函数返回了一个file对象并调用这个file的writeFrom()方法写入字节
// 这里为了紧凑删减部分代码
FutureRandomAccessFile open({FileMode mode FileMode.read}) {...return _dispatchWithNamespace(_IOService.fileOpen, [null, _rawPath, mode._mode]).then((response) {_checkForErrorResponse(response, Cannot open file, path);return _RandomAccessFile(response as int, path);});
}通过返回值类型我们知道file其实是一个_RandomAccessFile类的实例。注意此类的源码也在file_impl.dart文件中这里我们直接查看它的writeFrom实现
// 删减部分代码
FutureRandomAccessFile writeFrom(Listint buffer,[int start 0, int? end]) {...List request new Listdynamic.filled(4, null);request[0] null;request[1] result.buffer;request[2] result.start;request[3] end - (start - result.start);return _dispatch(_IOService.fileWriteFrom, request).then((response) {_checkForErrorResponse(response, writeFrom failed, path);_resourceInfo.addWrite(end! - (start - result.start));return this;});
}此方法内部调用了一个_dispatch方法我们通过方法名和参数名大致可以猜测出此处应该是向所谓的_IOService派发了一条类型为_IOService.fileWriteFrom的请求消息我们继续看一下这个方法的实现 FutureObject? _dispatch(int request, List data, {bool markClosed false}) {if (closed) {return new Future.error(new FileSystemException(File closed, path));}if (_asyncDispatched) {var msg An async operation is currently pending;return new Future.error(new FileSystemException(msg, path));}if (markClosed) {closed true;}_asyncDispatched true;data[0] _pointer();return _IOService._dispatch(request, data).whenComplete(() {_asyncDispatched false;});}可以看到此方法并没有太多处理只是继续调用_IOService._dispatch静态方法派发请求消息。这里我们找到io_service.dart源文件打开发现其并没有具体实现
// 省略部分常量定义
class _IOService {...static const int fileReadByte 18;static const int fileWriteByte 19;static const int fileRead 20;static const int fileReadInto 21;static const int fileWriteFrom 22;...external static FutureObject? _dispatch(int request, List data);
}看到这里很多人可能就只能无奈放弃了因为external修饰的方法一般是本地方法也就是说该方法是由VM底层的C来实现的。也许有人会去VM的C源码中搜索结果一无所获因为C代码中找不到名为_dispatch的函数或方法。
但是我看到此处就发觉了不对劲Dart层的方法也是必须与C层的函数映射关联起来才能调用的并不是简单的在方法名上面加个external修饰就大功告成的否则Dart VM怎么知道你这个external函数到底对应哪个C函数如果写过Java的JNI代码对此应该深有体会。另外在Dart的2.14版本以前是支持第三方开发者为Dart写本地扩展的简单说就是写一个C函数然后映射到Dart层供人调用这个机制与官方推荐的Dart FFI不同但与Java JNI最为相似。我曾在dart 2.5版本上试验过本机扩展机制里面在声明Dart 层的方法时是需要明确指定映射到C函数的名称的。Dart 2.15之后本机扩展的官方文档被删除了也就是不让第三方开发者使用此机制但是Dart VM和Dart上层仍然是使用此机制交互的学习Dart 本机扩展机制是有利于我们剖析理解Dart VM底层的。这里我找到了官方文档的备份想要探究本机扩展的可以查看 独立 Dart VM 的本机扩展。
现在回到我们的_dispatch方法这里没有任何标记用于指定映射到C的函数名所以这行代码绝对是有问题的不可能正确执行。我通过文件内容检索工具检索了Dart SDK的全部代码终于发现了其中的猫腻。
这里我找到了sdk\lib\_internal\vm\bin\io_service_patch.dart
patch
class _IOService {static _IOServicePorts _servicePorts new _IOServicePorts();static RawReceivePort? _receivePort;static late SendPort _replyToPort;static HashMapint, Completer _messageMap new HashMapint, Completer();static int _id 0;patchstatic FutureObject? _dispatch(int request, List data) {int id;do {id _getNextId();} while (_messageMap.containsKey(id));final SendPort servicePort _servicePorts._getPort(id);_ensureInitialize();final Completer completer new Completer();_messageMap[id] completer;try {servicePort.send(dynamic[id, _replyToPort, request, data]);} catch (error) {_messageMap.remove(id)!.complete(error);if (_messageMap.length 0) {_finalize();}}return completer.future;}// ... 删除部分代码
}我们发现其实真正的_IOService实现代码被Dart SDK给隐藏了并且还带有一定的误导_dispatch方法根本就不是一个本机方法就是一个普通的Dart方法而已。这里官方是通过补丁的方式在编译时将所有的patch修饰的代码与前面公开的_IOService类进行替换或合并。简单说最终真正的_IOService是将上面的两个_IOService实现合并起来的完整代码。这里我还简单研究了一下patch注解这个注解并不是一个简单的注解换句话说它不是用我们熟知的Dart的注解生成器去做的注解解析它的实现非常复杂代码是在Dart的编译前端那个包。也就是说它并不是去做Dart源码级别的处理而是在源码解析之后直接修改的AST相当于是修改了中间产物和闲鱼的那个AspectD框架类似。
这里真不得不吐槽一下Dart 官方的坑爹
继续我们今天的源码剖析_dispatch中的实现实际上不是直接去调用C的本机扩展函数它是获取了一个Native层面的端口然后向这个端口发消息这里的端口通信和Dart层的Isolate端口通信是一样的。看到此处请思考一个问题这里为什么要进行端口通信而不是直接调用C层的扩展函数
很简单理由和我们的Isolate通信一样这里肯定是为了跨线程看到这里_dispatch的实现我们就应该知道Dart VM层肯定是起了一个工作线程Dart层的调用和VM层的实现不在同一个线程了。
接下来我们注意到final SendPort servicePort _servicePorts._getPort(id);这行代码它是从_servicePorts中获取一个发送消息的端口让我们看看这个类的具体实现
class _IOServicePorts {static const int maxPorts 32;final ListSendPort _ports [];final Listint _useCounts [];final Listint _freePorts [];final Mapint, int _usedPorts HashMapint, int();_IOServicePorts();SendPort _getPort(int forRequestId) {assert(!_usedPorts.containsKey(forRequestId));if (_freePorts.isEmpty _ports.length maxPorts) {final SendPort port _newServicePort();_ports.add(port);_useCounts.add(0);_freePorts.add(_ports.length - 1);}final index _freePorts.isNotEmpty? _freePorts.removeLast(): forRequestId % maxPorts;_usedPorts[forRequestId] index;_useCounts[index];return _ports[index];}void _returnPort(int forRequestId) {final index _usedPorts.remove(forRequestId)!;if (--_useCounts[index] 0) {_freePorts.add(index);}}pragma(vm:external-name, IOService_NewServicePort)external static SendPort _newServicePort();
}整体代码很少逻辑也很清晰主要就是调用一个本机扩展方法_newServicePort()在C层面创建了一个接收消息的服务端口然后把这个端口的SendPort保存起来复用。这里的_newServicePort才是一个真正的external方法它使用pragma注解将Dart层的方法声明与底层的IOService_NewServicePort函数名关联起来。至此我们才终于有了继续向底层探索的线索
我们继续在虚拟机的C源码目录sdk\runtime\中搜索IOService_NewServicePort我们可能会在sdk\runtime\bin\io_natives.cc中找到一些基于宏的声明这些声明的目的主要是自动生成符合Dart 本机扩展机制的C代码源码里面的这些宏定义主要就是为了简少编写一些模版代码的工作量。
#define IO_NATIVE_LIST(V)
...V(InternetAddress_RawAddrToString, 1) \V(IOService_NewServicePort, 0) \V(Namespace_Create, 2) \
...这里函数名后面的数值是代表函数的参数个数。这个函数的真正实现是在sdk\sdk\runtime\bin\io_service.cc中
namespace dart {
namespace bin {#define CASE_REQUEST(type, method, id) \case IOService::k##type##method##Request: \response type::method##Request(data); \break;void IOServiceCallback(Dart_Port dest_port_id, Dart_CObject* message) {Dart_Port reply_port_id ILLEGAL_PORT;CObject* response CObject::IllegalArgumentError();CObjectArray request(message);if ((message-type Dart_CObject_kArray) (request.Length() 4) request[0]-IsInt32() request[1]-IsSendPort() request[2]-IsInt32() request[3]-IsArray()) {CObjectInt32 message_id(request[0]);CObjectSendPort reply_port(request[1]);CObjectInt32 request_id(request[2]);CObjectArray data(request[3]);reply_port_id reply_port.Value();switch (request_id.Value()) {IO_SERVICE_REQUEST_LIST(CASE_REQUEST);default:UNREACHABLE();}}CObjectArray result(CObject::NewArray(2));result.SetAt(0, request[0]);result.SetAt(1, response);ASSERT(reply_port_id ! ILLEGAL_PORT);Dart_PostCObject(reply_port_id, result.AsApiCObject());
}Dart_Port IOService::GetServicePort() {return Dart_NewNativePort(IOService, IOServiceCallback, true);
}void FUNCTION_NAME(IOService_NewServicePort)(Dart_NativeArguments args) {Dart_SetReturnValue(args, Dart_Null());Dart_Port service_port IOService::GetServicePort();if (service_port ! ILLEGAL_PORT) {// Return a send port for the service port.Dart_Handle send_port Dart_NewSendPort(service_port);Dart_SetReturnValue(args, send_port);}
}} // namespace bin
} // namespace dart可以看到整个io_service.cc中的代码并不多逻辑也不难理解。IOService_NewServicePort函数首先调用IOService::GetServicePort()创建了一个本地端口而GetServicePort()又调用了Dart_NewNativePort函数这里的Dart_NewNativePort是Dart VM公开给第三方的虚拟机API我们可以直接查看它在dart_native_api.h中的文档注释了解含义。注意这里Dart_前缀的函数都是VM公开的API接着它又调用Dart_NewSendPort为这个本地端口创建了一个发送端口句柄然后将发送端口句柄作为返回值进行了返回。我们看到IOService_NewServicePort函数似乎没有返回值但是请注意这里的返回值是对应上层的Dart函数声明的我们再看一眼Dart端的函数声明 pragma(vm:external-name, IOService_NewServicePort)external static SendPort _newServicePort();所以当调用_newServicePort()完之后Dart 层就可以获得一个用于向底层发送消息的发送端口句柄。
Dart 侧的梳理
到这里我们再来回顾梳理一下流程 Dart 层的File类是一个接口具体实现是一个私有的_File子类 _File子类也没有真正处理它是对用起来更加繁琐的RandomAccessFile的简化封装 RandomAccessFile也是一个接口它的具体实现在私有的_RandomAccessFile子类中 _RandomAccessFile类也不是最终目的地它是通过调用_IOService._dispatch静态方法向虚拟机底层发消息的方式与VM中的C方法进行交互 不同的消息类型就代表了不同的IO操作 class _IOService {...static const int fileReadByte 18;static const int fileWriteByte 19;static const int fileRead 20;static const int fileReadInto 21;static const int fileWriteFrom 22;...external static FutureObject? _dispatch(int request, List data);
}总结Dart 文件IO操作的真正实现是在VM的C函数中。
C 侧
接下来就只有一个关键问题需要搞明白了那就是C层是怎么接收并处理消息的
这里我们再回看一个细节
Dart_Port IOService::GetServicePort() {return Dart_NewNativePort(IOService, IOServiceCallback, true);
}当使用Dart_NewNativePort创建一个本地端口时它还注册了一个回调函数IOServiceCallback我们仔细观察这个回调函数就会发现它实际上就是Dart上层发来的消息处理器
#define CASE_REQUEST(type, method, id) \case IOService::k##type##method##Request: \response type::method##Request(data); \break;void IOServiceCallback(Dart_Port dest_port_id, Dart_CObject* message) {Dart_Port reply_port_id ILLEGAL_PORT;CObject* response CObject::IllegalArgumentError();CObjectArray request(message);if ((message-type Dart_CObject_kArray) (request.Length() 4) request[0]-IsInt32() request[1]-IsSendPort() request[2]-IsInt32() request[3]-IsArray()) {CObjectInt32 message_id(request[0]);CObjectSendPort reply_port(request[1]);CObjectInt32 request_id(request[2]);CObjectArray data(request[3]);reply_port_id reply_port.Value();switch (request_id.Value()) {IO_SERVICE_REQUEST_LIST(CASE_REQUEST);default:UNREACHABLE();}}CObjectArray result(CObject::NewArray(2));result.SetAt(0, request[0]);result.SetAt(1, response);ASSERT(reply_port_id ! ILLEGAL_PORT);Dart_PostCObject(reply_port_id, result.AsApiCObject());
}这个函数的参数是不是与external static FutureObject? _dispatch(int request, List data)方法很相似它前面的代码很好理解其实就是对参数的提取和转换最后得到的request_id就是消息类型然后通过switch选择执行对应的函数。这里的IO_SERVICE_REQUEST_LIST()宏定义在sdk\runtime\bin\io_service.h文件中
// This list must be kept in sync with the list in sdk/lib/io/io_service.dart
#define IO_SERVICE_REQUEST_LIST(V) \...V(File, ReadByte, 18) \V(File, WriteByte, 19) \V(File, Read, 20) \V(File, ReadInto, 21) \V(File, WriteFrom, 22) \...V(SSLFilter, ProcessFilter, 43)#define DECLARE_REQUEST(type, method, id) k##type##method##Request id,这里我们其实可以根据CASE_REQUEST宏把实际调用的C函数名拼接出来。我们最开始调用的writeAsBytes()方法对应的消息类型是 static const int fileWriteFrom 22消息类型的值是22这里正好对应宏定义中的V(File, WriteFrom, 22)。注意了这里括号中的数值就不是参数个数了而是赋值。
再根据CASE_REQUEST宏中response type::method##Request(data);我们拼出来的函数名应该是File_WriteFrom。如果你对C/C中的宏不了解你完全可以把它理解成纯粹的字符串替换##号就是一个粘连符号。我继续在C源码中搜索File_WriteFrom()函数的具体实现sdk\runtime\bin\file.cc
// 删除部分代码精简结构
void FUNCTION_NAME(File_WriteFrom)(Dart_NativeArguments args) {File* file GetFile(args);...Dart_Handle buffer_obj Dart_GetNativeArgument(args, 1);intptr_t start DartUtils::GetNativeIntptrArgument(args, 2);intptr_t end DartUtils::GetNativeIntptrArgument(args, 3);Dart_TypedData_Type type;intptr_t length end - start;intptr_t buffer_len 0;void* buffer NULL;Dart_Handle result Dart_TypedDataAcquireData(buffer_obj, type, buffer, buffer_len);...char* byte_buffer reinterpret_castchar*(buffer);bool success file-WriteFully(byte_buffer start, length);if (!success) {Dart_SetReturnValue(args, DartUtils::NewDartOSError(os_error));} else {Dart_SetReturnValue(args, Dart_Null());}
}可以看到文件的写入其实就是调用的C的File类进行操作的。也就是说Dart层的所谓文件操作其实就是把数据发送给C函数让C干活。
看到此处大家可能要质疑了你前面不是说Dart的文件操作是在一个子线程进行的吗所以才需要搞端口通信但是这里没有看到线程呀确实这个地方是有一些绕的并不是非常的直接关键问题在于IOServiceCallback()这个回调是谁调用的是在哪里调用的
我们先找到Dart_NewNativePort函数的具体实现sdk\runtime\vm\native_api_impl.cc
DART_EXPORT Dart_Port Dart_NewNativePort(const char* name,Dart_NativeMessageHandler handler,bool handle_concurrently) {if (name NULL) {name UnnamedNativePort;}if (handler NULL) {OS::PrintErr(%s expects argument handler to be non-null.\n,CURRENT_FUNC);return ILLEGAL_PORT;}if (!Dart::SetActiveApiCall()) {return ILLEGAL_PORT;}IsolateLeaveScope saver(Isolate::Current());// 核心代码NativeMessageHandler* nmh new NativeMessageHandler(name, handler);Dart_Port port_id PortMap::CreatePort(nmh);if (port_id ! ILLEGAL_PORT) {PortMap::SetPortState(port_id, PortMap::kLivePort);if (!nmh-Run(Dart::thread_pool(), NULL, NULL, 0)) {PortMap::ClosePort(port_id);port_id ILLEGAL_PORT;}}Dart::ResetActiveApiCall();return port_id;
}这里的核心代码就是创建了一个NativeMessageHandler对象然后调用了它的Run()方法。这里的NativeMessageHandler还持有了我们前面注册的IOServiceCallback回调。我们来看一下该类的声明其完整的源码内容也很少 。
sdk\runtime\vm\native_message_handler.h
// NativeMessageHandler 接收消息并将它们分派给本机 C 处理程序
class NativeMessageHandler : public MessageHandler {
public:NativeMessageHandler(const char* name, Dart_NativeMessageHandler func);~NativeMessageHandler();const char* name() const { return name_; }Dart_NativeMessageHandler func() const { return func_; }// ... 省略部分代码private:char* name_;Dart_NativeMessageHandler func_;
};该类继承自MessageHandler所以真正的Run方法是在其父类中实现的。我们找到sdk\runtime\vm\message_handler.cc
bool MessageHandler::Run(ThreadPool* pool,StartCallback start_callback,EndCallback end_callback,CallbackData data) {MonitorLocker ml(monitor_);if (FLAG_trace_isolates) {OS::PrintErr([] Starting message handler:\n\thandler: %s\n,name());}ASSERT(pool_ NULL);ASSERT(!delete_me_);pool_ pool;start_callback_ start_callback;end_callback_ end_callback;callback_data_ data;task_running_ true;bool result pool_-RunMessageHandlerTask(this);if (!result) {pool_ nullptr;start_callback_ nullptr;end_callback_ nullptr;callback_data_ 0;task_running_ false;}return result;
}这里关键的代码只有一行bool result pool_-RunMessageHandlerTask(this)调用线程池来执行一个任务。关于Dart VM线程池的剖析可以看我的另一篇剖析文章。这里的线程池对象是Dart VM在初始化时创建的一个全局线程池Dart::thread_pool()那么这里的线程池是在哪里创建的呢同样在上一篇线程池剖析的文章已经说明请移步阅读 Dart VM 线程池剖析。
继续我们的主题线程池的Run函数是一个模版函数 template typename T, typename... Argsbool Run(Args... args) {return RunImpl(std::unique_ptrTask(new T(std::forwardArgs(args)...)));}这里的T我们理解成Dart的泛型即可那么这里的new T(std::forwardArgs(args)...)其实就是new MessageHandlerTask()封装一个任务然后把参数透传进去所以接下来要看看MessageHandlerTask的声明以及构造方法
sdk\runtime\vm\message_handler.cc
class MessageHandlerTask : public ThreadPool::Task {
public:explicit MessageHandlerTask(MessageHandler* handler) : handler_(handler) {ASSERT(handler ! NULL);}virtual void Run() {ASSERT(handler_ ! NULL);handler_-TaskCallback();}private:MessageHandler* handler_;DISALLOW_COPY_AND_ASSIGN(MessageHandlerTask);
};可见MessageHandlerTask继承自线程池的ThreadPool::Task在上篇关于线程池的剖析文章中我们知道当一个任务Task对象被线程池调度执行时其实就是调用Task的Run方法所以这里的MessageHandlerTask任务被执行时其Run方法被工作线程执行。
那么这里的handler是什么呢其实就是前面调用pool_-RunMessageHandlerTask(this);传进去的this指针也就是NativeMessageHandler类实例的指针。而NativeMessageHandler类没有实现TaskCallback()方法这里其实是调用的父类实现最后我们来看看MessageHandler中该方法的具体实现
void MessageHandler::TaskCallback() {...// Handle any pending messages for this message handler.if (status ! kShutdown) {status HandleMessages(ml, (status kOK), true);}}...
}MessageHandler::MessageStatus MessageHandler::HandleMessages(MonitorLocker* ml,bool allow_normal_messages,bool allow_multiple_normal_messages) {...Message::Priority min_priority ((allow_normal_messages !paused()) ? Message::kNormalPriority: Message::kOOBPriority);std::unique_ptrMessage message DequeueMessage(min_priority);while (message ! nullptr) {...{DisableIdleTimerScope disable_idle_timer(idle_time_handler);status HandleMessage(std::move(message));}...
}以上方法省略大量代码只保留关键代码。首先是在TaskCallback()中调用了本类的HandleMessages()方法在HandleMessages()中又调用了一个虚函数HandleMessage()。注意这两个方法一个带有s结尾一个没有 virtual MessageStatus HandleMessage(std::unique_ptrMessage message) 0;既然是虚函数那么肯定是交给子类去实现的这里我们到子类NativeMessageHandler中找实现
MessageHandler::MessageStatus NativeMessageHandler::HandleMessage(std::unique_ptrMessage message) {if (message-IsOOB()) {UNREACHABLE();}ApiNativeScope scope;Dart_CObject* object ReadApiMessage(scope.zone(), message.get());(*func())(message-dest_port(), object);return kOK;
}到这里我们终于找到了(*func())(message-dest_port(), object);这行代码还记得func是什么吗它就是我们通过Dart_NewNativePort注册的回调函数的指针这里就是真正的调用IOServiceCallback回调的地方。这里传的参数也与IOServiceCallback回调的完全一致。
C 侧的梳理
简单回顾梳理一下C 端的流程
响应Dart层的_newServicePort()方法C侧对应的函数是IOService_NewServicePort()调用Dart_NewNativePort函数创建本地端口在创建本地端口的同时还创建了一个NativeMessageHandler对象并传入了一个处理消息的回调函数IOServiceCallback调用NativeMessageHandler的Run方法将消息处理封装成了一个线程池的任务在工作线程中执行TaskCallback()函数通过一些封装的调用最终执行处理消息的回调函数IOServiceCallback()
至此我们彻底搞明白了Dart 文件IO的底层细节明确了Dart的文件操作都是在C的工作线程中完成的当工作线程执行完了对应的文件操作就会向Dart的单线程模型返回结果。这就说明在Dart层面做应用开发是不需要担心文件操作耗时会阻塞Dart的主线程的因为虚拟机底层已经帮你开辟了子线程。
总结
画一个示意图做总结 关注公众号编程之路从0到1
