Android BlueDroid分析: OSI中的reactor的实现与使用分析

2024-03-04 12:18
文章标签 分析 实现 android 使用 reactor osi bluedroid

本文主要是介绍Android BlueDroid分析: OSI中的reactor的实现与使用分析，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

说明

actor相当于内核中的worker, 用于监控与执行任务. reactor可以认为是: re+actor.

监控使用的是epoll, 而执行任务使用的是对应的epoll_wait返回后的event type, 然后调用相关的read或者write函数来完成对应event type的处理.而epoll_wait对某个fd是监控和执行一次还是多次是使用eventfd来进行控制.
epoll可以看下面的recator_change_registration这个函数的实现.

而eventfd的控制可以search event_read与event_write来找到对应的控制点.

actor的类型

不断监控: 即epoll_wait返回后, 再继续监控. 体现在函数reactor_start,这个时候可以叫做reactor

一次性的: event发生后就被移除,不再监控. 体现在函数reactor_run_once, 这个时候可以叫做actor

epoll_wait的线程休眠问题

epoll_wait在reactor中是永远等待,在event来临之前不会timeout而返回(最后一个参数-1决定的):

do {ret = epoll_wait(reactor->epoll_fd, events, MAX_EVENTS, -1);} while (ret == -1 && errno == EINTR);
因此调用这个epoll_wait后, thread会进行到休眠等待状态.

函数

核心函数的实现

reactor关键的有两个函数

  1. reactor_start/reactor_run_once,run_reactor : epoll_wait
  2. reactor_new : eventfd, epoll_create, epoll_ctl(EPOLL_CTL_ADD)

结构体

struct reactor_t {int epoll_fd;  //用于epoll waitint event_fd;  // 用于reactor的控制, 例如停止监控pthread_mutex_t list_lock;  // protects invalidation_list.list_t *invalidation_list;  // reactor objects that have been unregistered.pthread_t run_thread;       // the pthread on which reactor_run is executing.bool is_running;            // indicates whether |run_thread| is valid.bool object_removed;
};struct reactor_object_t {int fd;                              // the file descriptor to monitor for events.void *context;                       // a context that's passed back to the *_ready functions.reactor_t *reactor;                  // the reactor instance this object is registered with.pthread_mutex_t lock;                // protects the lifetime of this object and all variables.void (*read_ready)(void *context);   // function to call when the file descriptor becomes readable.void (*write_ready)(void *context);  // function to call when the file descriptor becomes writeable.
};

每一个reactor都是使用reactor_object_t来定义, 即reactor_object_t注册到reactor_t中

需要注意的是里面有一个List, 存放着所有没有注册的reactor, 这个是在unregister的时候将前面注册过的reactor放入到这个List中.

新创建与注册

reactor_t *reactor_new(void) {reactor_t *ret = (reactor_t *)osi_calloc(sizeof(reactor_t));if (!ret)return NULL;ret->epoll_fd = INVALID_FD;ret->event_fd = INVALID_FD;// epoll用来监控ret->epoll_fd = epoll_create(MAX_EVENTS);if (ret->epoll_fd == INVALID_FD) {LOG_ERROR("%s unable to create epoll instance: %s", __func__, strerror(errno));goto error;}// eventfd作为semophore,用来协调和控制是否继续进行监控ret->event_fd = eventfd(0, 0);if (ret->event_fd == INVALID_FD) {LOG_ERROR("%s unable to create eventfd: %s", __func__, strerror(errno));goto error;}// 这个List用来将unregistered的reactor存放pthread_mutex_init(&ret->list_lock, NULL);ret->invalidation_list = list_new(NULL);if (!ret->invalidation_list) {LOG_ERROR("%s unable to allocate object invalidation list.", __func__);goto error;}// 将需要监控的fd放入到epoll中struct epoll_event event;memset(&event, 0, sizeof(event));event.events = EPOLLIN;event.data.ptr = NULL;if (epoll_ctl(ret->epoll_fd, EPOLL_CTL_ADD, ret->event_fd, &event) == -1) {LOG_ERROR("%s unable to register eventfd with epoll set: %s", __func__, strerror(errno));goto error;}return ret;error:;reactor_free(ret);return NULL;
}

register

reactor_object_t *reactor_register(reactor_t *reactor,int fd, void *context,void (*read_ready)(void *context),void (*write_ready)(void *context)) {assert(reactor != NULL);assert(fd != INVALID_FD);reactor_object_t *object = (reactor_object_t *)osi_calloc(sizeof(reactor_object_t));if (!object) {LOG_ERROR("%s unable to allocate reactor object: %s", __func__, strerror(errno));return NULL;}object->reactor = reactor;object->fd = fd;object->context = context;object->read_ready = read_ready;object->write_ready = write_ready;pthread_mutex_init(&object->lock, NULL);// 和reactor_change类似, 也是给给回调.struct epoll_event event;memset(&event, 0, sizeof(event));if (read_ready)event.events |= (EPOLLIN | EPOLLRDHUP);if (write_ready)event.events |= EPOLLOUT;event.data.ptr = object;// 重点就在于这里, 和上面的reactor_new类似if (epoll_ctl(reactor->epoll_fd, EPOLL_CTL_ADD, fd, &event) == -1) {LOG_ERROR("%s unable to register fd %d to epoll set: %s", __func__, fd, strerror(errno));pthread_mutex_destroy(&object->lock);osi_free(object);return NULL;}return object;
}

修改reactor

bool reactor_change_registration(reactor_object_t *object, // 需要被修改的reactor可以从object获取void (*read_ready)(void *context), //传入函数指针,函数参数为void * contextvoid (*write_ready)(void *context)) {assert(object != NULL);struct epoll_event event;memset(&event, 0, sizeof(event));if (read_ready)  // 如果需要监控读的话,设置对应的FLAGevent.events |= (EPOLLIN | EPOLLRDHUP);if (write_ready)  // 如果需要监控写的话,设置对应的FLAGevent.events |= EPOLLOUT;event.data.ptr = object;// 更改epoll FLAGif (epoll_ctl(object->reactor->epoll_fd, EPOLL_CTL_MOD, object->fd, &event) == -1) {LOG_ERROR("%s unable to modify interest set for fd %d: %s", __func__, object->fd, strerror(errno));return false;}pthread_mutex_lock(&object->lock);object->read_ready = read_ready;  // 更改read event发生后的 callbackobject->write_ready = write_ready;pthread_mutex_unlock(&object->lock);return true;
}

run_reactor是实施监控的核心

// Runs the reactor loop for a maximum of |iterations|.
// 0 |iterations| means loop forever.
// |reactor| may not be NULL.
static reactor_status_t run_reactor(reactor_t *reactor, int iterations) {assert(reactor != NULL);reactor->run_thread = pthread_self();reactor->is_running = true;struct epoll_event events[MAX_EVENTS];for (int i = 0; iterations == 0 || i < iterations; ++i) {pthread_mutex_lock(&reactor->list_lock);list_clear(reactor->invalidation_list);pthread_mutex_unlock(&reactor->list_lock);int ret;do { // wait将block并等待Event发生ret = epoll_wait(reactor->epoll_fd, events, MAX_EVENTS, -1);} while (ret == -1 && errno == EINTR);if (ret == -1) {LOG_ERROR("%s error in epoll_wait: %s", __func__, strerror(errno));reactor->is_running = false;return REACTOR_STATUS_ERROR;}for (int j = 0; j < ret; ++j) {// The event file descriptor is the only one that registers with// a NULL data pointer. We use the NULL to identify it and break// out of the reactor loop.if (events[j].data.ptr == NULL) {eventfd_t value;eventfd_read(reactor->event_fd, &value); //监控的控制,即使用eventfd来完成epoll流程的更改,例如这里的退出监控reactor->is_running = false;return REACTOR_STATUS_STOP;}reactor_object_t *object = (reactor_object_t *)events[j].data.ptr;pthread_mutex_lock(&reactor->list_lock);if (list_contains(reactor->invalidation_list, object)) {pthread_mutex_unlock(&reactor->list_lock);continue;}// Downgrade the list lock to an object lock.pthread_mutex_lock(&object->lock);pthread_mutex_unlock(&reactor->list_lock);reactor->object_removed = false;if (events[j].events & (EPOLLIN | EPOLLHUP | EPOLLRDHUP | EPOLLERR) && object->read_ready)//根据Event type来调用回调object->read_ready(object->context);//调用read回调if (!reactor->object_removed && events[j].events & EPOLLOUT && object->write_ready)object->write_ready(object->context); //调用write回调完成处理pthread_mutex_unlock(&object->lock);if (reactor->object_removed) {pthread_mutex_destroy(&object->lock);osi_free(object);}}}reactor->is_running = false;return REACTOR_STATUS_DONE;
}

使用分析与示例

创建是在thread创建的时候做的, 然后注册则是在需要监控某个fd的时候使用.

例如HCI H4使用的是串口, 那么在打开tty后有一个fd, 然后就需要对这个fd进行监控;

例如对其进行读写操作, 从而完成HCI的信息传输.

对于这个例子,调用代码的顺序如下:

在hci_hal_h4.c中有:

uart_stream = eager_reader_new(uart_fd, &allocator_malloc, HCI_HAL_SERIAL_BUFFER_SIZE, SIZE_MAX, "hci_single_channel");

其中uart_fd,就是打开串口的fd, 然后: 

  ret->inbound_read_object = reactor_register(thread_get_reactor(ret->inbound_read_thread), // 从前面创建的thread中获取在thread中创建的reactor, 这样子因为reactor的epoll_wait会睡眠也就会在这个新创建的thread中睡眠了.fd_to_read,ret,inbound_data_waiting,NULL);

这里面说到的新创建的thread的创建位于eager_reader.c中: 

  ret->inbound_read_thread = thread_new(thread_name);
这个thread_new会创建reactor: 

osi/src/thread.c
调用流程如下:
 
thread_t *thread_new_sized(const char *name, size_t work_queue_capacity) {ret->reactor = reactor_new();
....
}
而
 
thread_t *thread_new(const char *name) {return thread_new_sized(name, DEFAULT_WORK_QUEUE_CAPACITY);
}


前面的fd变成了fd_to_read, 接下来就到了epoll_ctl中的fd:

(epoll_ctl(reactor->epoll_fd,EPOLL_CTL_ADD,fd, &event)== -1)

这个时候就被加入到监控列表中了, 但是还没有调用epoll_wait来进行监控,仅仅只是加入到epoll的监控fd中. 

void eager_reader_register(eager_reader_t *reader, reactor_t *reactor, eager_reader_cb read_cb, void *context) {assert(reader != NULL);assert(reactor != NULL);assert(read_cb != NULL);// Make sure the reader isn't currently registered.eager_reader_unregister(reader);reader->outbound_read_ready = read_cb;reader->outbound_context = context;reader->outbound_registration = reactor_register(reactor, reader->bytes_available_fd, reader, internal_outbound_read_ready, NULL);
}

下面是hci_hal_h4.c中的处理:

  return eager_reader_read(uart_stream, buffer, max_size, block);



// SEE HEADER FOR THREAD SAFETY NOTE
size_t eager_reader_read(eager_reader_t *reader, uint8_t *buffer, size_t max_size, bool block) {assert(reader != NULL);assert(buffer != NULL);// If the caller wants nonblocking behavior, poll to see if we have// any bytes available before reading.if (!block && !has_byte(reader))//只有非block才会去has_type(里面是select尝试读取),见下面函数的分析return 0;// Find out how many bytes we have available in our various buffers.eventfd_t bytes_available;if (eventfd_read(reader->bytes_available_fd, &bytes_available) == -1) {LOG_ERROR("%s unable to read semaphore for output data.", __func__);return 0;}// 上面的eventfd_read会block, 要等到有数据才会才会返回了.if (max_size > bytes_available)max_size = bytes_available;size_t bytes_consumed = 0;while (bytes_consumed < max_size) {if (!reader->current_buffer)reader->current_buffer = fixed_queue_dequeue(reader->buffers);// 这个queue里面的数据是在inbound_data_waiting中read并enqueue的,在以后的eager_reader.c分析中会有说明.size_t bytes_to_copy = reader->current_buffer->length - reader->current_buffer->offset;if (bytes_to_copy > (max_size - bytes_consumed))bytes_to_copy = max_size - bytes_consumed;memcpy(&buffer[bytes_consumed], &reader->current_buffer->data[reader->current_buffer->offset], bytes_to_copy);bytes_consumed += bytes_to_copy;reader->current_buffer->offset += bytes_to_copy;if (reader->current_buffer->offset >= reader->current_buffer->length) {reader->allocator->free(reader->current_buffer);reader->current_buffer = NULL;}}// 将没有读完即unconsumed的字节数使用eventfd_write写回去,这样子下次来读有可以继续从前面读完的位置继续读取bytes_available -= bytes_consumed;if (eventfd_write(reader->bytes_available_fd, bytes_available) == -1) {LOG_ERROR("%s unable to write back bytes available for output data.", __func__);}return bytes_consumed;
}


has_byte实现 

static bool has_byte(const eager_reader_t *reader) {assert(reader != NULL);fd_set read_fds;FD_ZERO(&read_fds);FD_SET(reader->bytes_available_fd, &read_fds);// Immediate timeoutstruct timeval timeout;timeout.tv_sec = 0;timeout.tv_usec = 0;
// 设置的timeout时间是0,所以仅仅是try read一把,不会block select(reader->bytes_available_fd + 1, &read_fds, NULL, NULL, &timeout);return FD_ISSET(reader->bytes_available_fd, &read_fds);
}

总结

以上就是reactor的分析. 总结起来便是:

  • 1. 创建一个reactor,这个被thread.c中封装在new thread中
  • 2. reactor_register完成reactor的注册
  • 3. 调用run_reactor的封装函数完成一次性或者不停的监控, 这个依然被thread.c中的run_thread进行了封装.
  • 4. 使用reactor_unregister来将fd从epoll中移除,放入到unregistered list中
  • 5. 线程stop/exit/kill的时候,使用reactor_stop来停止epoll监控.

这篇关于Android BlueDroid分析: OSI中的reactor的实现与使用分析的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/773140。

相关文章

中文分词jieba库的使用与实景应用（一）

中文分词jieba库的使用与实景应用（一）

知识星球：https://articles.zsxq.com/id_fxvgc803qmr2.html 目录 一.定义： 精确模式（默认模式）： 全模式： 搜索引擎模式： paddle 模式（基于深度学习的分词模式）： 二 自定义词典 三.文本解析   调整词出现的频率 四. 关键词提取 A. 基于TF-IDF算法的关键词提取 B. 基于TextRank算法的关键词提取
阅读更多...
使用SecondaryNameNode恢复NameNode的数据

使用SecondaryNameNode恢复NameNode的数据

1）需求： NameNode进程挂了并且存储的数据也丢失了，如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2）故障模拟 （1）kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 （2）删除NameNode存储的数据（/opt/module/hadoop-3.1.4/data/tmp/dfs/na
阅读更多...
Hadoop数据压缩使用介绍

Hadoop数据压缩使用介绍

一、压缩原则 （1）运算密集型的Job，少用压缩 （2）IO密集型的Job，多用压缩 二、压缩算法比较 三、压缩位置选择 四、压缩参数配置 1）为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器 2）要在Hadoop中启用压缩，可以配置如下参数
阅读更多...
Makefile简明使用教程

Makefile简明使用教程

文章目录 规则makefile文件的基本语法：加在命令前的特殊符号：.PHONY伪目标: Makefilev1 直观写法v2 加上中间过程v3 伪目标v4 变量 make 选项-f-n-C Make 是一种流行的构建工具，常用于将源代码转换成可执行文件或者其他形式的输出文件(如库文件、文档等)。Make 可以自动化地执行编译、链接等一系列操作。 规则 makefile文件
阅读更多...
hdu1043（八数码问题，广搜 + hash（实现状态压缩） ）

hdu1043（八数码问题，广搜 + hash（实现状态压缩） ）

利用康拓展开将一个排列映射成一个自然数，然后就变成了普通的广搜题。 #include<iostream>#include<algorithm>#include<string>#include<stack>#include<queue>#include<map>#include<stdio.h>#include<stdlib.h>#include<ctype.h>#inclu
阅读更多...
性能分析之MySQL索引实战案例

性能分析之MySQL索引实战案例

文章目录 一、前言二、准备三、MySQL索引优化四、MySQL 索引知识回顾五、总结 一、前言 在上一讲性能工具之 JProfiler 简单登录案例分析实战中已经发现SQL没有建立索引问题，本文将一起从代码层去分析为什么没有建立索引？ 开源ERP项目地址：https://gitee.com/jishenghua/JSH_ERP 二、准备 打开IDEA找到登录请求资源路径位置
阅读更多...
使用opencv优化图片（画面变清晰）

使用opencv优化图片（画面变清晰）

文章目录 需求影响照片清晰度的因素 实现降噪测试代码 锐化空间锐化Unsharp Masking频率域锐化对比测试 对比度增强常用算法对比测试 需求 对图像进行优化，使其看起来更清晰，同时保持尺寸不变，通常涉及到图像处理技术如锐化、降噪、对比度增强等 影响照片清晰度的因素 影响照片清晰度的因素有很多，主要可以从以下几个方面来分析 1. 拍摄设备 相机传感器：相机传
阅读更多...
【C++】_list常用方法解析及模拟实现

【C++】_list常用方法解析及模拟实现

相信自己的力量，只要对自己始终保持信心，尽自己最大努力去完成任何事，就算事情最终结果是失败了，努力了也不留遗憾。💓💓💓 目录   ✨说在前面 🍋知识点一：什么是list？ •🌰1.list的定义 •🌰2.list的基本特性 •🌰3.常用接口介绍 🍋知识点二：list常用接口 •🌰1.默认成员函数 🔥构造函数（⭐） 🔥析构函数 •🌰2.list对象
阅读更多...
【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

✨✨ 欢迎大家来到景天科技苑✨✨ 🎈🎈 养成好习惯，先赞后看哦~🎈🎈 🏆 作者简介：景天科技苑 🏆《头衔》：大厂架构师，华为云开发者社区专家博主，阿里云开发者社区专家博主，CSDN全栈领域优质创作者，掘金优秀博主，51CTO博客专家等。 🏆《博客》：Python全栈，前后端开发，小程序开发，人工智能，js逆向，App逆向，网络系统安全，数据分析，Django，fastapi
阅读更多...
让树莓派智能语音助手实现定时提醒功能

让树莓派智能语音助手实现定时提醒功能

最初的时候是想直接在rasa 的chatbot上实现，因为rasa本身是带有remindschedule模块的。不过经过一番折腾后，忽然发现，chatbot上实现的定时，语音助手不一定会有响应。因为，我目前语音助手的代码设置了长时间无应答会结束对话，这样一来，chatbot定时提醒的触发就不会被语音助手获悉。那怎么让语音助手也具有定时提醒功能呢？ 我最后选择的方法是用threading.Time
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2