多核通信

合抱之木，生于毫末。

《老子·道德经》

在智能手机平台中，为了保证多媒体业务在平台上的流畅运行，多采用专用的处理器来处理多媒体业务。如在Qucalcomm MSM 7K 平台上，就包含了4 个处理器内核，Qucalcomm MSM 7K 平台采用ARM 9（mARM，modemARM）处理基带业务；采用mDSP（Modem DSP）来处理协议栈业务；采用ARM 11（aARM，application ARM）来负责Linux 操作系统的运行；采用aDSP（Application DSP）来处理多媒体业务方面的编/解码加速工作。

在2010 年末，智能终端的CPU 得到了快速发展，高端的智能终端已经采用了Cortex-A8 的双核处理器。而在平板电脑上，Nvidia 开发的基于Cortex-A9 的双核处理器Tegra 2 则成了市场的宠儿。

通信离不开内存的操作，在Qucalcomm平台上，内存一般分为3种：基带内存（Modem Memory）、应用内存（Application Memory）和共享内存，其中系统MPU保护基带内存不被aARM接入，ARM MMU保护应用内存不被mARM接入。在本章中，将主要介绍基于共享内存的多核通信。

3.1 共享内存

在Linux 中，实现进程通信的机制有很多种，如信号、管道、信号量、消息队列、共享内存和套接字等，但共享内存的方式效率最高。

在Aurora 中，共享内存是多核通信的物理基础，其实现主要包括3 个部分：共享内存驱动（SMD，Shared Memory Driver）、共享内存状态机（SMSM，Shared Memory State Machine）和共享内存管理器（SMEM，Shared Memory Manager）。其中SMD 用于多核之间的数据通信；SMSM用于多核之间的状态通信；SMEM是一个底层的协议，是物理RAM共享内存的管理接口，是SMD和SMSM的基础。

SMEM 具有两种分配模式：动态SMEM 和静态SMEM，动态SMEM 根据需要实时分配，静态SMEM则会预先分配。SMEM的主要接口为：smem_alloc()、smem_find()、smem_init()等。

SMEM、SMD、SMSM的实现都需要硬件平台厂商提供支持

3.1.1 同步与互斥

在Aurora 中，共享内存用到了自旋锁和互斥锁的概念。

自旋锁是Linux 内核的同步机制之一，与互斥锁类似，但自旋锁使用者一般保持锁的时间非常短。自旋锁的效率远高于互斥锁。自旋锁的定义位于aurora\msm\工程msm\include\linux\spinlock_*.h文件中。在Aurora 中，并未引入在Kernel 2.6.25 中才引入的排队自旋锁（FIFO Ticket Spinlock）概念。排队自旋锁（FIFO Ticket Spinlock）通过保存执行线程申请锁的顺序信息，可以解决传统自旋锁的“不公平”问题，在设计之初仅支持X86 架构。

自旋锁的定义如下：

typedef struct {
volatile unsigned int lock; //无符号整数
} raw_spinlock_t;

lock虽然被定义为无符号整数，但是实际上被当做有符号整数使用。slock值为1 代表锁未被占用，值为0 或负数代表锁被占用。初始化时slock被置为1。

与信号量和读写信号量导致调用者睡眠不同，自旋锁不会引起调用者睡眠。如果自旋锁已被别的执行单元保持，调用者就一直循环以确定是否该自旋锁的保持者已经释放了锁。

由于自旋锁适用的访问共享资源的时间非常短， 导致自旋锁通常应用于中断上下文访问和对共享资源访问文件非常短的场景中，如果被保护的共享资源在进程上下文访问，则应使用信号量。

与信号量和读写信号量在保持期间可以被抢占的情况不同，自旋锁在保持期间是抢占失效的，自旋锁只有在内核可抢占或SMP（Symmetrical Multi-Processing）的情况下才真正需要，在单CPU且不可抢占的内核下，自旋锁的所有操作都是空操作。

由于智能终端平台上通常存在多个CPU 或DSP，自旋锁的运用就显得非常重要。在Aurora 的SMD 和SMSM的实现上，自旋锁主要运用于中断处理、信道列表和信道状态的变更过程中，自旋锁的定义如下：

static DEFINE_SPINLOCK(smd_lock);
static DEFINE_SPINLOCK(smem_lock);

互斥锁主要用于实现Linux 内核中的互斥访问功能，在Aurora 的SMD 的实现上，互斥锁主要用于SMD 信道的打开或关闭过程。定义如下：

static DEFINE_MUTEX(smd_creation_mutex);

关于自旋锁和互斥锁的更多内容请参考文献。

3.1.2 SMD 数据通信（1）

在Linux 中，基于SMD 的数据通信是以信道的形式作为一个设备存在的，作为一种双向信道，其接口的实现遵循Linux 设备驱动规范。在Qucalcomm 平台上，SMD 的缓冲大小为8192bit，最大信道数为64，SMD的头大小为20bit。

SMD 的相关代码实现主要位于aurora\msm\msm\arch\arm\mach-msm目录下。主要文件包括： smd.c、smd_nmea.c、smd_qmi.c、smd_rpcrouter.c、smd_rpcrouter_clients.c、smd_rpcrouter_device.c、smd_rpcrouter_servers.c、smd_tty.c等。

SMD 信道需要同时维护接收信道、发送信道的状态和数据信息，SMD 的信道定义如下：

struct smd_channel {
volatile struct smd_half_channel *send; //发送握手信道
volatile struct smd_half_channel *recv; //接收握手信道
unsigned char *send_buf; //发送信道数据
unsigned char *recv_buf; //接收信道数据
unsigned buf_size;
struct list_head ch_list; //信道列表
unsigned current_packet;
unsigned n;
void *priv;
void (*notify)(void *priv, unsigned flags);
int (*read)(smd_channel_t *ch, void *data, int len); //读数据
int (*write)(smd_channel_t *ch, const void *data, int len); //写数据
int (*read_avail)(smd_channel_t *ch); //是否可读
int (*write_avail)(smd_channel_t *ch); //是否可写
int (*read_from_cb)(smd_channel_t *ch, void *data, int len);
void (*update_state)(smd_channel_t *ch);
unsigned last_state;
char name[20];
struct platform_device pdev;
unsigned type;
};

共享信道的信道状态在其握手信道中记录，握手信道的定义如下：

struct smd_half_channel {
unsigned state;
unsigned char fDSR;
unsigned char fCTS;
unsigned char fCD;
unsigned char fRI;
unsigned char fHEAD; //头部
unsigned char fTAIL; //尾部
unsigned char fSTATE; //状态
unsigned char fUNUSED;
unsigned tail;
unsigned head;
};

在实际实现中，SMD信道分配被封装在SMEM（Shared Memory Manager）模块中，系统提供了smem_init()、smem_alloc()、smem_get_entry()等内存操作函数供SMD和SMSM操作。

SMD 的状态共有SMD_SS_CLOSED、SMD_SS_OPENING、SMD_SS_OPENED、SMD_SS_FLUSHING、SMD_SS_CLOSING、SMD_SS_RESET、SMD_SS_RESET_OPENING等。其变化过程如图3-1所示。

 

图3-1 SMD的状态机

下面结合SMD信道的实现简要介绍SMD 信道的分配、打开、读取、写入、关闭等操作。

1．分配信道

SMD 信道根据数据的类型可以分为流信道和包信道，其中包信道具有比流信道更强的流控制能力，包含头信息。在创建SMD 信道时，会根据信道类型的不同，确定创建的是FIFO信道还是普通信道，是流信道还是包信道，然后为SMD 进行设备注册。SMD 分配信道的实现如下：

代码3-1 SMD分配信道的过程

static void smd_alloc_channel(struct smd_alloc_elm *alloc_elm)
{
struct smd_channel *ch;
uint32_t *smd_ver;
//分配SMEM 内存
smd_ver=smem_alloc(SMEM_VERSION_SMD, 32 * sizeof(uint32_t));
if (smd_ver && ((smd_ver[VERSION_MODEM]>>16)>=1))
ch=smd_alloc_channel_v2(alloc_elm->cid); //FIFO信道
else
ch=smd_alloc_channel_v1(alloc_elm->cid); //普通信道
if (ch==0)
return;
ch->type=SMD_CHANNEL_TYPE(alloc_elm->type);
memcpy(ch->name, alloc_elm->name, 20);
ch->name[19]=0;
if (smd_is_packet(alloc_elm)) { //包信道
ch->read=smd_packet_read;
ch->write=smd_packet_write;
ch->read_avail=smd_packet_read_avail;
ch->write_avail=smd_packet_write_avail;
ch->update_state=update_packet_state;
ch->read_from_cb=smd_packet_read_from_cb;
} else { //流信道
ch->read=smd_stream_read;
ch->write=smd_stream_write;
ch->read_avail=smd_stream_read_avail;
ch->write_avail=smd_stream_write_avail;
ch->update_state=update_stream_state;
ch->read_from_cb=smd_stream_read;
}
ch->pdev.name=ch->name;
ch->pdev.id=ch->type;
pr_info("smd_alloc_channel() '%s' cid=%d\n",
ch->name, ch->n);
mutex_lock(&smd_creation_mutex); //互斥锁
//将信道添加到“smd_ch_closed_list”列表中
list_add(&ch->ch_list, &smd_ch_closed_list); mutex_unlock(&smd_creation_mutex);
platform_device_register(&ch->pdev); //注册设备
}
3.1.2 SMD 数据通信（2）2．打开信道为了打开一个信道，首先要判断SMD 信道是否已经初始化。如果SMD 信道已经初始化，就根据信道名获得信道，将信道加入到“smd_ch_list”信道列表中并设置该信道的状态为SMD_SS_OPENING，然后调用notify_other_smd()函数通知其他的信道该信道已经激活。在默认情况下，其信道类型为SMD_APPS_MODEM，打开一个SMD信道的实现如下：代码3-2 SMD打开信道的过程int smd_named_open_on_edge(const char *name, uint32_t edge,
smd_channel_t **_ch,
void *priv, void (*notify)(void *, unsigned))
{
struct smd_channel *ch;
unsigned long flags;
if (smd_initialized==0) { //判断SMD 信道是否已初始化
printk(KERN_INFO "smd_open() before smd_init()\n");
return -ENODEV;
}
D("smd_open('%s', %p, %p)\n", name, priv, notify);
ch=smd_get_channel(name, edge); //获取信道
if (!ch)
return -ENODEV;
if (notify==0)
notify=do_nothing_notify;
ch->notifynotify=notify;
ch->current_packet=0;
ch->last_state=SMD_SS_CLOSED;
ch->privpriv=priv;
*_ch=ch;
D("smd_open: opening '%s'\n", ch->name);
spin_lock_irqsave(&smd_lock, flags); //自旋锁
list_add(&ch->ch_list, &smd_ch_list); //将信道添加到“smd_ch_list”列表中
D("%s: opening ch %d\n", __func__, ch->n);
smd_state_change(ch, ch->last_state, SMD_SS_OPENING); //信道状态变更
spin_unlock_irqrestore(&smd_lock, flags);
return 0;
}3．关闭信道关闭信道的操作相对简单，首先将信道从“smd_ch_list”信道列表中删除，然后将信道状态设置为SMD_SS_CLOSED，并将信道添加到“smd_ch_closed_list”信道列表中即可。关闭SMD 信道的实现如下：代码3-3 SMD关闭信道的过程int smd_close(smd_channel_t *ch)
{
unsigned long flags;
printk(KERN_INFO "smd_close(%p)\n", ch);
if (ch==0)
return -1;
spin_lock_irqsave(&smd_lock, flags); //自旋锁
ch->notify=do_nothing_notify;
list_del(&ch->ch_list); //从打开信道列表中去除该信道
ch_set_state(ch, SMD_SS_CLOSED); //设置信道状态
spin_unlock_irqrestore(&smd_lock, flags);
mutex_lock(&smd_creation_mutex); //互斥锁
list_add(&ch->ch_list, &smd_ch_closed_list);//将信道添加至关闭信道列表
mutex_unlock(&smd_creation_mutex);
return 0;
}4．信道读取包信道的内容读取涉及缓冲的复制、与其他SMD 的消息通信和包状态的更新。从包信道读取数据的实现如下：代码3-4 SMD包信道读取数据的过程static int smd_packet_read(smd_channel_t *ch, void *data, int len)
{
unsigned long flags;
int r;
if (len<0)
return -EINVAL;
if (len>ch->current_packet)
len=ch->current_packet;
r=ch_read(ch, data, len); //读取数据
if (r>0)
notify_other_smd(ch->type);
spin_lock_irqsave(&smd_lock, flags); //自旋锁
ch->current_packet-=r;
update_packet_state(ch); //更新包状态
spin_unlock_irqrestore(&smd_lock, flags);
return r;
}流信道的内容读取非常简单，只需要调用ch_read()函数读取数据并通知其他SMD 该信道处于打开状态即可。流信道读取的实现如下：代码3-5 SMD流信道读取数据的过程{
int r;
if (len<0)
return -EINVAL;
r=ch_read(ch, data, len); //读取数据
if (r>0)
notify_other_smd(ch->type); //通知其他SMD，该信道处于激活状态。
return r;
}流信道和包信道在读取信道数据时，都需要调用ch_read()函数来实现真正的数据读取，ch_read()函数的实现如下：代码3-6 SMD信道读取数据的过程static int ch_read(struct smd_channel *ch, void *_data, int len)
{
void *ptr;
unsigned n;
unsigned char *datadata=data;
int orig_len=len;
while (len>0) {
n=ch_read_buffer(ch, &ptr); //读取缓冲
if (n==0)
break;
if (n>len)
n=len;
if (data)
memcpy(data, ptr, n); //数据复制
data+=n;
len-=n;
ch_read_done(ch,n); //读取完成
}
return orig_len-len;
}5．信道写入在信道的数据写入方面，同样分为流信道数据的写入和包信道数据的写入两种类型。向包信道中写入数据的过程和读取数据不同，在写入数据前，要首先利用smd_stream_write_avail()函数判断是否有数据可供写入，在确定有可供写入的数据的情况下才调用smd_stream_write()函数执行数据的写入操作。包信道写入数据的实现如下：代码3-7 SMD包信道写入数据的过程static int smd_packet_write(smd_channel_t *ch,const void *_data, int len)
{
int ret;
unsigned hdr[5];
D("smd_packet_write() %d->ch%d\n",len, ch->n);
if (len<0)
return-EINVAL;
if (smd_stream_write_avail(ch)<(len+SMD_HEADER_SIZE)) //判断数据写入进度
return-ENOMEM;
hdr[0]=len;
hdr[1]=hdr[2]=hdr[3]=hdr[4]=0;
ret=smd_stream_write(ch, hdr, sizeof(hdr)); //流写入
if (ret<0 || ret!=sizeof(hdr)) {
D("%s failed to write pkt header: "
"%d returned\n", __func__, ret);
return -1;
}
ret=smd_stream_write(ch, _data, len);
if (ret<0 || ret != len) {
D("%s failed to write pkt data: "
"%d returned\n", __func__, ret);
return ret;
}
return len;
}流信道数据的写入和包信道数据的写入不同，其首先获得下一段可用缓冲的指针，然后执行内存复制，流信道写入数据的实现如下：代码3-8 SMD流信道写入数据的过程static int smd_stream_write(smd_channel_t *ch, const void *_data, int len)
{
void *ptr;
const unsigned char *buf=data;
unsigned xfer;
int orig_len=len;
D("smd_stream_write() %d->ch%d\n", len, ch->n);
if (len<0)
return -EINVAL;
while ((xfer=ch_write_buffer(ch, &ptr)) != 0) { //写入数据
if (!ch_is_open(ch))
break;
if (xfer>len)
xfer=len;
memcpy(ptr, buf, xfer); //内存复制
ch_write_done(ch, xfer); //完成写入数据
len-=xfer;
buf+=xfer;
if (len==0)
break;
}
if (orig_len-len)
notify_other_smd(ch->type); //通知其他SMD，该信道处于激活状态
return orig_len-len;
}SMD 是多核通信的基础，在SMD 之上是一个叫做RPC路由器（RPC Router）的封装。关于RPC路由器，将在3.2 节过程调用中详细介绍。通过SMD，可以为系统提供RPC、DIAG、AT命令、NMEA（GPS数据）、数据服务、拨号等服务。3.1.3 SMSM 状态通信SMSM为处理状态而非数据的共享内存，主要就电源管理、共享内存、定时器、进程状态等信息在多核之间进行通信。SMSM的实现框架和SMD 类似。SMSM 的状态共有：SMSM_INIT、SMSM_OSENTERED 、SMSM_SMDWAIT、SMSM_SMDINIT、SMSM_RPCWAIT、SMSM_RPCINIT、SMSM_RESET、SMSM_RSA、SMSM_RUN、SMSM_PWRC、SMSM_TIMEWAIT、SMSM_TIMENIT、SMSM_PWRC_EARLY_EXIT、SMSM_WFPI、SMSM_SLEEP、SMSM_SLEEPEXIT、SMSM_OEMSBL_RELEASE、SMSM_APPS_REBOOT、SMSM_SYSTEM_POWER_DOWN、SMSM_SYSTEM_REBOOT 、SMSM_SYSTEM_DOWNLOAD 、SMSM_PWRC_SUSPEND 、SMSM_APPS_SHUTDOWN、SMSM_SMD_LOOPBACK、SMSM_RUN_QUIET、SMSM_MODEM_WAIT、SMSM_MODEM_BREAK、SMSM_MODEM_CONTINUE、SMSM_UNKNOWN等。在实际应用上，SMSM主要在处理器发生状态变化，以及发生中断、重设基带处理器时运行，下面结合处理器状态变化和中断等两种情况简要介绍SMSM的处理过程。处理处理器的变化，首先判断接入点的有效性，是应用处理器还是基带处理器等。如果接入点有效，就分配共享内存，并将相应的信息通知给其他的SMSM，下面是smsm_change_state()函数的实现：代码3-9 SMSM改变状态的过程int smsm_change_state(uint32_t smsm_entry,
uint32_t clear_mask, uint32_t set_mask)
{
unsigned long flags;
uint32_t *smsm;
uint32_t old_state;
if (smsm_entry >= SMSM_NUM_ENTRIES) { //判断有效性
printk(KERN_ERR "smsm_change_state: Invalid entry %d",
smsm_entry);
return -EINVAL;
}
spin_lock_irqsave(&smem_lock, flags);
smsm=smem_alloc(ID_SHARED_STATE, //分配共享内存
SMSM_NUM_ENTRIES * sizeof(uint32_t));
if (smsm) {
old_state=smsm[smsm_entry];
smsm[smsm_entry]=(smsm[smsm_entry] & ~clear_mask) | set_mask;
if (msm_smd_debug_mask & MSM_SMSM_DEBUG)
printk(KERN_INFO "smsm_change_state %x\n",
smsm[smsm_entry]);
notify_other_smsm(SMSM_APPS_STATE, old_state, smsm[smsm_entry]);
} //通知其他SMSM
spin_unlock_irqrestore(&smem_lock, flags);
if (smsm==NULL) {
printk(KERN_ERR "smsm_change_state<SM NO STATE>\n");
return -EIO;
}
return 0;
}当INT_A9_M2A_5 和INT_ADSP_A11 中断发生时，会触发SMSM 的中断处理函数，SMSM处理中断的过程如下：代码3-10 SMSM 处理中断的过程static irqreturn_t smsm_irq_handler(int irq, void *data)
{
unsigned long flags;
uint32_t *smsm;
static uint32_t prev_smem_q6_apps_smsm;
if (irq==INT_ADSP_A11) {
smsm=smem_alloc(SMEM_SMD_SMSM_INTR_MUX, //分配内存
SMSM_NUM_INTR_MUX * sizeof(uint32_t));
if (!smsm ||
(smsm[SMEM_Q6_APPS_SMSM]==prev_smem_q6_apps_smsm))
return IRQ_HANDLED;
prev_smem_q6_apps_smsm=smsm[SMEM_Q6_APPS_SMSM];
}
spin_lock_irqsave(&smem_lock, flags);
smsm=smem_alloc(ID_SHARED_STATE, //分配内存
SMSM_NUM_ENTRIES * sizeof(uint32_t));
if (smsm==0) {
printk(KERN_INFO "<SM NO STATE>\n");
} else {
unsigned old_apps, apps;
unsigned modm=smsm[SMSM_MODEM_STATE];
old_apps=apps=smsm[SMSM_APPS_STATE];
if (msm_smd_debug_mask & MSM_SMSM_DEBUG)
printk(KERN_INFO "<SM %08x %08x>\n", apps, modm);
if (apps & SMSM_RESET) {
apps &=~SMSM_RESET;
smd_fake_irq_handler(0); //让SMD响应假中断
modem_queue_start_reset_notify(); //发起重设提醒
} else if (modm & SMSM_RESET) {
apps |=SMSM_RESET;
} else {
apps|=SMSM_INIT;
if (modm & SMSM_SMDINIT)
apps|=SMSM_SMDINIT;
if (modm & SMSM_RPCINIT)
apps|=SMSM_RPCINIT;
if ((apps & (SMSM_INIT|SMSM_SMDINIT|SMSM_RPCINIT))==
(SMSM_INIT|SMSM_SMDINIT|SMSM_RPCINIT))
apps|=SMSM_RUN;
}
if (smsm[SMSM_APPS_STATE]!=apps) {
if (msm_smd_debug_mask & MSM_SMSM_DEBUG)
printk(KERN_INFO "<SM %08x NOTIFY>\n", apps);
smsm[SMSM_APPS_STATE]=apps;
do_smd_probe();
notify_other_smsm(SMSM_APPS_STATE, old_apps, apps); //通知其他SMSM
}
}
spin_unlock_irqrestore(&smem_lock, flags);
return IRQ_HANDLED;
}3.2 过程调用在Android中，远程过程调用是基于ONC RPC来实现的，在Android上层的接口为IBinder,在框架层面，其实现主要包括RPC路由器（RPC Router）、RPC服务器（RPC Server）和RPC管道等3 部分。关于多核之间远程过程调用的内容，在Qucalcomm 平台上，相关的代码实现主要位于urora\msm\msm\arch\arm\mach-msm 和aurora\msm\msm\net\sunrpc 目录下。关于ONC RPC协议的实现位于aurora\msm7k\msm7k\librpc 目录下。关于RDMA 的内容位于aurora\msm\msm\net\sunrpc\xprtrdma 目录下。协议的具体实现不是本书关注的重点，这里就不再进行过多的描述，有兴趣的读者可以参考文献。远程过程调用物理上是基于3.1.2 节数据通信中介绍的共享内存（SMD，Shared MemoryDriver）的。事实上，Aurora 的鼠标、键盘、全速USB、高速USB等设备都是基于ONC RPC来实现通信的。相关的实现分布在rpcmouse.c 、rpckbd.c 、rpc_fsusb.c、rpc_hsusb.c等文件中。