【原创】xenomai内核解析-xenomai与普通linux进程之间通讯XDDP(二)-实时与非实时关联(bind流程)

1.概述

上篇文章介绍了实时端socket创建和配置的流程，本篇文章来看bind操作，实时端与非实时端是如何关联起来的？

XDDP通讯的底层设备为xnpipe，是linux任务与xenomai任务通讯的核心，在linux看来是一个字符设备，xnpipe在xenomai内核初始化过程初始化，并完成linux端xnipipe字符设备注册。

bind的主要操作就是**根据socket配置，分配资源，如指定通讯过程中分配释放的内存池（xnheap）、缓冲区大小等,并根据端口号，分配对应的xnpipe设备，并将rtdm_fd与xnipipe设备通过数组关联（用次设备号作为数组下标，端口号即次设备号）。**下面来看详细过程。

##　２. 解析bind函数

与前面函数一样，用户空间实时任务对socket调用

bind()

函数，先进入实时库licobalt，再由实时库libcobalt来发起实时内核系统调用：

saddr.sipc_family = AF_RTIPC;saddr.sipc_port = XDDP_PORT;ret = bind(s, (struct sockaddr *)&saddr, sizeof(saddr));

/*lib\\cobalt\\rtdm.c*/COBALT_IMPL(int, bind, (int fd, const struct sockaddr *my_addr, socklen_t addrlen)){.....ret = do_ioctl(fd, _RTIOC_BIND, &args);if (ret != -EBADF && ret != -ENOSYS)return set_errno(ret);return __STD(bind(fd, my_addr, addrlen));}static int do_ioctl(int fd, unsigned int request, void *arg){....ret = XENOMAI_SYSCALL3(sc_cobalt_ioctl,	fd, request, arg);....return ret;}

进入系统调用后执行

__xddp_ioctl()

static int __xddp_ioctl(struct rtdm_fd *fd,unsigned int request, void *arg){struct rtipc_private *priv = rtdm_fd_to_private(fd);struct sockaddr_ipc saddr, *saddrp = &saddr;struct xddp_socket *sk = priv->state;int ret = 0;switch (request) {......COMPAT_CASE(_RTIOC_BIND):/*bind操作*/ret = rtipc_get_sockaddr(fd, &saddrp, arg);.......ret = __xddp_bind_socket(priv, saddrp);break;......}return ret;}

前面文章看了

__xddp_ioctl()

中的

COMPAT_CASE(_RTIOC_SETSOCKOPT)

分支，现在来看

COMPAT_CASE(_RTIOC_BIND)

__xddp_bind_socket()

static int __xddp_bind_socket(struct rtipc_private *priv,struct sockaddr_ipc *sa){struct xddp_socket *sk = priv->state;struct xnpipe_operations ops;rtdm_lockctx_t s;size_t poolsz;void *poolmem;...../*参数检查*/poolsz = sk->poolsz;if (poolsz > 0) {poolsz = xnheap_rounded_size(poolsz);//对齐poolsz += xnheap_rounded_size(sk->reqbufsz);poolmem = xnheap_vmalloc(poolsz); //ZONE_NORMAL中分配，分配后使用xnhead方式进行管理......ret = xnheap_init(&sk->privpool, poolmem, poolsz);/*初始化内存区*/.......sk->bufpool = &sk->privpool;} elsesk->bufpool = &cobalt_heap;if (sk->reqbufsz > 0) {sk->buffer = xnheap_alloc(sk->bufpool, sk->reqbufsz);/*从bufpool 分配sk->buffer*/......sk->curbufsz = sk->reqbufsz;}/*__xddp_bind_socket()剩余部分*/.......}

该函数中先检查相关参数的合法性，然后配置xddp本地内存池

privpool

，上篇文章

setsocketopt()

只是设置了内存池的大小

poolsz

，但是还没有真正分配内存，现在开始分配内存，先将内存大小向上页对齐（PAGE_SIZE为4K），由于xenomai内存池管理缘故，每个内存池至少为（2*PAGE_SIZE）；然后看看

poolsz

是否够分配

reqbufsz

，不够的话向

reqbufsz

对齐。

大小确定后正式调用linux接口分配，从ZONE_NORMAL中分配，分配后调用

xnheap_init()

将该内存初始化(具体流程参见文章xenomai内核解析–实时内存管理–xnheap)。然后将bufpool指向该内存池。接着分配数据缓冲区

bufpool

，从

bufpool

指向的内存池中分配缓冲区内存。

上面大部分都是关于缓冲区与内存池的设置，到此还没有看到关于数据真正传输控制的东西，

__xddp_bind_socket()

接着要完成bind相关工作：

static int __xddp_bind_socket(struct rtipc_private *priv,struct sockaddr_ipc *sa){struct xnpipe_operations ops;....../*接上部分*/sk->fd = rtdm_private_to_fd(priv);ops.output = &__xddp_output_handler;ops.input = &__xddp_input_handler;ops.alloc_ibuf = &__xddp_alloc_handler;ops.free_ibuf = &__xddp_free_handler;ops.free_obuf = &__xddp_free_handler;ops.release = &__xddp_release_handler;ret = xnpipe_connect(sa->sipc_port, &ops, sk);//将SK与OPS与sipc_port联系起来，绑定端口.......sk->minor = ret;sa->sipc_port = ret;sk->name = *sa;/*剩余部分*/}

先取出rtdm_fd，设置

struct xnpipe_operations

struct xnpipe_operations

中的ops为xddp通讯过程中buf分配释放的函数；

struct xnpipe_operations {void (*output)(struct xnpipe_mh *mh, void *xstate);int (*input)(struct xnpipe_mh *mh, int retval, void *xstate);void *(*alloc_ibuf)(size_t size, void *xstate);void (*free_ibuf)(void *buf, void *xstate);void (*free_obuf)(void *buf, void *xstate);void (*release)(void *xstate);};

谁会用到这些buf？xnpipe,xnpipe管理收发的数据包时需要动态管理buf，在具体通讯的时候，我们要为每一个数据包在内核空间临时申请一块内存来存放数据，这块内存的申请释放要足够快，而且不能影响实时性，所以得从xnheap中申请，也就是前面

xddp-socket->bufpool

指向的内存池,对每块内存的分配释放就是由这个回调函数来完成。需要注意的是，linux端读写数据的时候也是从

xddp-socket->bufpool

中分配释放内存，这会在后面文章中看到；

还有一些场合，执行内核用户线程需要在数据到来或发送的时候添加一些hook，这通过output()/input()来设置monitor函数。

接下来调用

xnpipe_connect(sa->sipc_port, &ops, sk)

将xddp_socket与linux端的xnipipe函数关联起来，由于xnpipe不是动态分配的，内核配置时确定xnpipe的数量，以数组的形式，这样确保了确定性，linux启动时，xenomai内核初始化过程中就已将xnpipe初始化。

2.1 xnpipe介绍

XNPIPE是xenomai内核提供的通讯层，是linux任务与xenomai任务通讯的核心。每个xddp socket对应一个XNPIPE，XNPIPE的个数XNPIPE_NDEVS在内核编译时配置,内核默认配置为32个XNPIPE对象保存在全局数组

xnpipe_states[XNPIPE_NDEVS]

中，全局bitmap

xnpipe_bitmap

中记录着XNPIPE对象分配情况，

xnpipe_states[]

内的xpipe对象在xenomai初始化时初始化,在linux VFS下生成对应的设备节点，后一节说明。

内核xnpipe数量配置menuconfig 项如下：

[*] Xenomai/cobalt —>

Sizes and static limits —>

(32) Number of pipe devices

XNPIPE对象结构

struct xnpipe_state

如下。

struct xnpipe_state {struct list_head slink;	/* Link on sleep queue */struct list_head alink;	/* Link on async queue */struct list_head inq;		/* in/out是从实时端看的类似USB的端口*/int nrinq;		     /*链表节点数，代指消息个数*/struct list_head outq;		/* From kernel to user-space */int nroutq;struct xnsynch synchbase;/*同步*/struct xnpipe_operations ops;/*执行一些hook函数，如释放消息节点的内存，有消息时执行monitor函数等*/void *xstate;		/* xddp是指向 xddp_socket *//* Linux kernel part */unsigned long status;/*状态标识*/struct fasync_struct *asyncq;wait_queue_head_t readq;	/* linux端读等待队列*/wait_queue_head_t syncq;	/*linux端写同步等待队列*/int wcount;			      /* 这个设备节点的进程数量*/size_t ionrd;             /*缓冲包数据长度*/};

最为linux任务与xenomai任务通讯的中间人，struct xnpipe_state成员分为两个部分，首先看xenomai相关成员

slink、alink 链接到xnpipe睡眠队列、async 队列。
inq 实时端接收数据包队列，其中的in是相对xenomai端来说的，每个链表节点表示一个数据包，包个数用成员
```
nrinq
```
记录。
outq 实时端发送数据包队列，其中的out是相对xenomai端来说的，每个链表节点表示一个数据包，包个数用成员
```
nroutq
```
记录。
synchbase xenomai资源同步对象，当没有数据时会阻塞在
```
xnsynch
```
等待资源可用。
ops 动态发送数据过程中执行的回调函数。
xstate 指向私有数据，对于xddp指向xddp_socket。

接着是linux相关成员：

status linux端收发操作状态码,各状态码定义如下

#define XNPIPE_KERN_CONN         0x1  	/*内核端(rt)已连接*/#define XNPIPE_KERN_LCLOSE       0x2	/*内核端(rt)关闭*/#define XNPIPE_USER_CONN         0x4	/*用户端(nrt)已链接*/#define XNPIPE_USER_SIGIO        0x8	/*用户(nrt)已设置异步通知*/#define XNPIPE_USER_WREAD        0x10 	/*用户(nrt)端读*/#define XNPIPE_USER_WREAD_READY  0x20	 /*用户端(nrt)读就绪*/#define XNPIPE_USER_WSYNC        0x40	/*用户端(nrt)写同步*/#define XNPIPE_USER_WSYNC_READY  0x80	/*rt端已读数据，待完成写同步唤醒nrt*/#define XNPIPE_USER_LCONN        0x100	/*(nrt)端正在执行连接操作*/

asyncq 异步通知队列用于linux端poll操作。
readq linux端读等待队列,当没有数据时会在该队列上阻塞，知道有数据可读。
syncq linux端写同步队列,对同步发送的数据包，会在该队列上阻塞知道数据包被实时端读取。
wcount 使用同一个xnpipe的linux端进程数。
ionrd 缓冲区数据包长度。

2.2 xnpipe与xddp_socket关联

回到

__xddp_bind_socket()

接着调用

xnpipe_connect()

开始执行bind工作，

sa->sipc_port

中保存着我们要使用的rtipc端口（XNPIPE），如果为-1表示自动分配，自动分配后Linux端可通过上节设置的label来找到该xddp。

int xnpipe_connect(int minor, struct xnpipe_operations *ops, void *xstate){struct xnpipe_state *state;int need_sched = 0, ret;spl_t s;minor = xnpipe_minor_alloc(minor);.....state = &xnpipe_states[minor];xnlock_get_irqsave(&nklock, s);ret = xnpipe_set_ops(state, ops);.....state->status |= XNPIPE_KERN_CONN;xnsynch_init(&state->synchbase, XNSYNCH_FIFO, NULL);state->xstate = xstate;state->ionrd = 0;if (state->status & XNPIPE_USER_CONN) {if (state->status & XNPIPE_USER_WREAD) {/** Wake up the regular Linux task waiting for* the kernel side to connect (xnpipe_open).*/state->status |= XNPIPE_USER_WREAD_READY;need_sched = 1;}if (state->asyncq) {	/* Schedule asynch sig. */state->status |= XNPIPE_USER_SIGIO;need_sched = 1;}}if (need_sched)xnpipe_schedule_request();xnlock_put_irqrestore(&nklock, s);return minor;}

在xnpipe_connect中首先根据传入的

sa->sipc_port

,分配对应的XNPIPE设备号

minor

。

static inline int xnpipe_minor_alloc(int minor){......if (minor == XNPIPE_MINOR_AUTO)//(-1)表示自动分配端口minor = find_first_zero_bit(xnpipe_bitmap, XNPIPE_NDEVS);if (minor == XNPIPE_NDEVS ||(xnpipe_bitmap[minor / BITS_PER_LONG] &(1UL << (minor % BITS_PER_LONG))))minor = -EBUSY;elsexnpipe_bitmap[minor / BITS_PER_LONG] |=(1UL << (minor % BITS_PER_LONG));.....return minor;}

xnpipe_minor_alloc()

就是去

xnpipe_bitmap

中查看我们要bind的

rtipc_port

是否已经被使用，指定-1则表示自动分配。得到可用的

minor

后，就去

xnpipe_states[]

中得到对应的

struct xnpipe_state

，配置到xnpipe的ops，初始化xenomai资源同步对象

state->synchbase

，设置状态掩码为rt已链接，如果nrt此时也处于open xddp设备状态，唤醒 Linux任务，以等待linux内核端连接。

接着

__xddp_bind_socket（）

剩余部分，如果我们设置的是使用label方式，自动分配的端口号，就调用

xnregistry_enter

注册一个实时对象xnregistry,以便linux端通过路径

/proc/xenomai/registry/rtipc/xddp/%s

来打开通讯端点。

将分配的XNPIPE minor与rddm_fd对应关系保存到

portmap[]

中；

static int __xddp_bind_socket(struct rtipc_private *priv,struct sockaddr_ipc *sa){/* Set default destination if unset at binding time.*/if (sk->peer.sipc_port < 0)sk->peer = *sa;if (poolsz > 0)xnheap_set_name(sk->bufpool, "xddp-pool@%d", sa->sipc_port);if (*sk->label) {/*使用xlabel*/ret = xnregistry_enter(sk->label, sk, &sk->handle,&__xddp_pnode.node);.......}cobalt_atomic_enter(s);portmap[sk->minor] = rtdm_private_to_fd(priv);__clear_bit(_XDDP_BINDING, &sk->status);__set_bit(_XDDP_BOUND, &sk->status);if (xnselect_signal(&priv->send_block, POLLOUT))xnsched_run();cobalt_atomic_leave(s);return 0;}

到此分配好了一个XNPIPE对象，内核所有数据结构初始化好，实时应用可以使用该socket发送接收数据了。

3. xnpipe设备注册流程

上面仅简单说明了xnpipe_state,没有看xnpipe在linux端注册的具体过程，其实就是注册一个字符设备，xnpipe在linux端的初始化是在xenomai内核初始化过程中调用

xnpipe_mount()

完成初始化。

static int __init xenomai_init(void){......ret = xnpipe_mount(); /*注册进程间通讯管道xnpipe*/......}

static struct file_operations xnpipe_fops = {.read = xnpipe_read,.write = xnpipe_write,.poll = xnpipe_poll,.unlocked_ioctl = xnpipe_ioctl,.open = xnpipe_open,.release = xnpipe_release,.fasync = xnpipe_fasync};int xnpipe_mount(void){struct xnpipe_state *state;struct device *cldev;int i;for (state = &xnpipe_states[0];state < &xnpipe_states[XNPIPE_NDEVS]; state++) {state->status = 0;state->asyncq = NULL;INIT_LIST_HEAD(&state->inq); /*初始化数据包链表*/state->nrinq = 0;INIT_LIST_HEAD(&state->outq);/*初始化数据包链表*/state->nroutq = 0;}/*创建class*/xnpipe_class = class_create(THIS_MODULE, "frtpipe");if (IS_ERR(xnpipe_class)) {printk(XENO_ERR "error creating rtpipe class, err=%ld\\n",PTR_ERR(xnpipe_class));return -EBUSY;}/*创建设备*/for (i = 0; i < XNPIPE_NDEVS; i++) {  /*创建rtp1-rtpn*/cldev = device_create(xnpipe_class, NULL,MKDEV(XNPIPE_DEV_MAJOR, i),NULL, "rtp%d", i);.......}/*注册字符设备*/if (register_chrdev(XNPIPE_DEV_MAJOR, "rtpipe", &xnpipe_fops)) {......}/*注册xenomai与linux间异步唤醒虚拟中断*/xnpipe_wakeup_apc =xnapc_alloc("pipe_wakeup", &xnpipe_wakeup_proc, NULL);return 0;}

3.1 xnpipe初始化与设备创建

xnpipe_mount()中,内核构建的时候我们在指定了多少个xnipipe就要注册多少个字符设备

将xnpipe_states[]内的xnpipe对象初始化。
创建设备类.

创建设备.

device_create()->device_create_vargs()->device_create_groups_vargs()->dev = kzalloc(sizeof(*dev), GFP_KERNEL);->retval = device_add(dev);

设备添加过程中,向用户空间发出uevent(添加对象)事件,用户空间的守护进程

systemd-udevd

监听到该事件后,

systemd-udevd

在

/dev

下生成设备节点

/dev/rtpX

3.2注册rtpipe设备

接着注册字符设备，将file_operation与cdev实列关联,其

file_operations

为

xnpipe_fops

.linux端最终通过这些接口来操作设备

/dev/rtpX

来与xenomai 应用通讯。

static struct file_operations xnpipe_fops = {.read = xnpipe_read,.write = xnpipe_write,.poll = xnpipe_poll,.unlocked_ioctl = xnpipe_ioctl,.open = xnpipe_open,.release = xnpipe_release,.fasync = xnpipe_fasync};

int __register_chrdev(unsigned int major, unsigned int baseminor,unsigned int count, const char *name,const struct file_operations *fops){struct char_device_struct *cd;struct cdev *cdev;int err = -ENOMEM;cd = __register_chrdev_region(major, baseminor, count, name);cdev = cdev_alloc();cdev->owner = fops->owner;cdev->ops = fops;kobject_set_name(&cdev->kobj, "%s", name);err = cdev_add(cdev, MKDEV(cd->major, baseminor), count);cd->cdev = cdev;return major ? 0 : cd->major;}

字符设备在内核设备数据库中由cdev结构体表示，字符设备驱动程序的主要工作就是创建并向内核注册cdev实例。注册的方式是调用 __register_chrdev_region,传入注册字符设备的主次设备号和名称(这里需要注意了,次设备号就是数组下标,也就是我们bind的端口号)，然后分配一个

struct cdev

结构，将 cdev 的 ops 成员变量指向这个模块声明的

file_operations

。然后，cdev_add 会将这个字符设备添加到内核中一个叫作

struct kobj_map *cdev_map

的结构，来统一管理所有字符设备。

其中，

MKDEV(cd->major, baseminor)

表示将主设备号和次设备号生成一个

dev_t

的整数，然后将这个整数

dev_t

和

cdev

关联起来。

int cdev_add(struct cdev *p, dev_t dev, unsigned count){int error;p->dev = dev;p->count = count;error = kobj_map(cdev_map, dev, count, NULL,exact_match, exact_lock, p);kobject_get(p->kobj.parent);return 0;}

3.3 注册xnpipe_wakeup_apc

接着注册一个异步过程调用(Asynchronous Procedure Call)

xnpipe_wakeup_apc

，apc基于ipipe虚拟中断。通过APC，Xenomai域中的活动可以让在Linux内核重新获得控制后,让要延迟处理的程序尽快的在linux域中调度。

xnpipe_wakeup_apc

是ipipe实现的一种虚拟中断机制，主要用于xenomai内核与linux内核的事件通知，其处理过程和ipipe处理硬件中断一致，所以实时性好。其具体实现会在ipipe系列文章中详细解析，敬请关注本博客。

现简单说明其作用：linux端一个任务$nrt$与xenomai实时任务$rt$使用xddp进行通讯，此时$nrt$读阻塞等待数据，当$rt$向$nrt$发送数据后，xenomai内核就会发送一个

xnpipe_wakeup_apc

，由于是基于ipipe虚拟中断实现，相当于给linux发送了一个中断，发送后会将该虚拟中断暂时在linux域挂起，当linux得到运行时才会去处理该虚拟中断的handler，进而知道可以唤醒阻塞的$nrt$，这个过程中完全是在xenomai域完成的，对xenomai实时性没有任何影响。

后续文章将从linux端、实时端的数据收发接口进行解析XDDP的详细通讯过程，请关注本博客。

【原创】xenomai内核解析–xenomai与普通linux进程之间通讯XDDP(二)–实时与非实时关联(bind流程)