所载送的信息是无格式的字节流,不知道分界线在哪,也没通信规范,另外缺乏控制手段,比如保温优先级,管道机制的大小只有1页,管道很容易写满而读取没有及时,发送者只能休眠,强化了管道机制同步要求.另外管道机制开销不少,尤其是当发送信息量很少时,平均每个字节所耗费的代价非常高
Linux内核为系统IPC提供了一个统一的系统调用ipc()
int ipc(unsigned int call,int firtst,int second,int third,void*ptr,int firth);
其中第一个参数为具体的操作码
#define SEMOP 1#define SEMGET 2#define SEMCTL 3#define MSGSND 11#define MSGRCV 12#define MSGGET 13#define MSGCTL 14#define SHMAT 21#define SHMDT 22#define SHMGET 23#define SHMCTL 24
操作码SEM开头的都是信号量设定,MSG开头则是消息队列,"SHM"开头则是共享内存区设定,其他参数因具体操作而不同,不过为了方便使用,c语言库函数分别提供了semget,msgget,msgsnd等库函数,这些库函数把其调用统一成了系统调用ipc.
内核入口为sys_ipc()
消息队列并没有纳入文件系统范畴,所以并不占用打开文件号,内核有个全局的数据结构msg_ids.专门用来管理报文队列
static struct ipc_ids msg_ids;
struct ipc_ids {int size;int in_use;int max_id;unsigned short seq;unsigned short seq_max;struct semaphore sem;spinlock_t ary;struct ipc_id* entries;//指向一个结构数组};
//ipc_id如下定义
struct ipc_id {struct kern_ipc_perm* p;};
kern_ipc_perm结构如下定义
/* used by in-kernel data structures */struct kern_ipc_perm{key_t key;//建值uid_t uid;gid_t gid;uid_t cuid;gid_t cgid;mode_t mode;unsigned long seq;};
每个报文队列都有个队列头,msg_queue数据结构
/* one msq_queue structure for each present queue on the system */struct msg_queue {struct kern_ipc_perm q_perm;time_t q_stime; /* last msgsnd time */time_t q_rtime; /* last msgrcv time */time_t q_ctime; /* last change time */unsigned long q_cbytes; /* current number of bytes on queue */unsigned long q_qnum; /* number of messages in queue */unsigned long q_qbytes; /* max number of bytes on queue */pid_t q_lspid; /* pid of last msgsnd */pid_t q_lrpid; /* last receive pid */struct list_head q_messages;struct list_head q_receivers;struct list_head q_senders;};
每个msg_queue数据结构唯一对应着一个报文队列,这些数据结构以及结构之间的关系可以如下总结:
全局ipc_ids数据结构的msg_ids是系统中所有报文队列的总根,总跟的指针entries指向一个ipc_id结构数组,数组中的每个元素都是ipc_id数据结构,结构中有个指针p指向一个kern_ipc_perm数据结构,由于kern_ipc_perm数据结构是报文队列头msg_queue的第一个成分,数组元素的指针p实际指向一个报文对哦,数组大小已经决定了可以建立的报文队列数目
消息队列的创建或获取(msgget)
//可以用于2个目的,通过给定的key创建队列,通过给定的key查找已存在队列asmlinkage long sys_msgget (key_t key, int msgflg){int id, ret = -EPERM;struct msg_queue *msq;//队列down(&msg_ids.sem);if (key == IPC_PRIVATE) //自己私用,无条件创建一个报文队列ret = newque(key, msgflg);//根据key与msgflgelse if ((id = ipc_findkey(&msg_ids, key)) == -1) { /* key没有找到key not used */if (!(msgflg & IPC_CREAT))//没找到,但没设定IPC_CREAT那就返回错误ret = -ENOENT;else//设定了就创建报文队列ret = newque(key, msgflg);} else if (msgflg & IPC_CREAT && msgflg & IPC_EXCL) {//同时设定了IPC_CREAT与IPC_EXCL返回错误ret = -EEXIST;} else {msq = msg_lock(id);if(msq==NULL)BUG();if (ipcperms(&msq->q_perm, msgflg))//检查访问权限是否符合规则ret = -EACCES;elseret = msg_buildid(id, msq->q_perm.seq);//将数组下标转换一体化的标识号msg_unlock(id);}up(&msg_ids.sem);return ret;//返回标识号}
查看具体的队列创建函数newque()
static int newque (key_t key, int msgflg){int id;struct msg_queue *msq;//队列头msq = (struct msg_queue *) kmalloc (sizeof (*msq), GFP_KERNEL);//分配结构if (!msq)return -ENOMEM;id = ipc_addid(&msg_ids, &msq->q_perm, msg_ctlmni);//分配一个标识号if(id == -1) {kfree(msq);return -ENOSPC;}//以下是报文队列头各种初始化msq->q_perm.mode = (msgflg & S_IRWXUGO);msq->q_perm.key = key;//key值msq->q_stime = msq->q_rtime = 0;msq->q_ctime = CURRENT_TIME;msq->q_cbytes = msq->q_qnum = 0;msq->q_qbytes = msg_ctlmnb;msq->q_lspid = msq->q_lrpid = 0;INIT_LIST_HEAD(&msq->q_messages);INIT_LIST_HEAD(&msq->q_receivers);INIT_LIST_HEAD(&msq->q_senders);msg_unlock(id);//将标识号转换为一个一体化的标识号,因为实际分配的id实际是数组下标会重复使用return msg_buildid(id,msq->q_perm.seq);- }
每个已建立的报文队列由一个标识号来代表,报文队列的标识号是全局的,所以必须把这全局范围内的唯一性,标识号由ipc_addid()分配,
/*** ipc_addid - add an IPC identifier* @ids: IPC identifier set* @new: new IPC permission set* @size: new size limit for the id array** Add an entry 'new' to the IPC arrays. The permissions object is* initialised and the first free entry is set up and the id assigned* is returned. The list is returned in a locked state on success.* On failure the list is not locked and -1 is returned.*///全局队列管理结构 新创建的队列头的这个结构int ipc_addid(struct ipc_ids* ids, struct kern_ipc_perm* new, int size){int id;size = grow_ary(ids,size);//增加管理数组的大小for (id = 0; id < size; id++) {if(ids->entries[id].p == NULL)//总根看管理数组的哪个为空goto found;}return -1;found:ids->in_use++;//已使用++if (id > ids->max_id)//最大的key值ids->max_id = id;new->cuid = new->uid = current->euid;//uid赋值new->gid = new->cgid = current->egid;//gid赋值new->seq = ids->seq++;//序列号if(ids->seq > ids->seq_max)//超过了最大的限定ids->seq = 0;//从0开始继续spin_lock(&ids->ary);ids->entries[id].p = new;//挂钩,新创建的报文头与总跟挂钩成功return id;}
库函数msgsnd-报文发送
参与通信双方通过mssget创建了一个报文队列后或取得该队列的标识号以后就可以向该队列发送或接收报文了
发送格式为msgbuf
/* message buffer for msgsnd and msgrcv calls */struct msgbuf {long mtype; /* type of message */char mtext[1]; /* message text */};
内核保存消息队列的格式
/* one msg_msg structure for each message */struct msg_msg {struct list_head m_list;long m_type;//类型int m_ts; /*长度 message text size */struct msg_msgseg* next;/* the actual message follows immediately */};
struct msg_msgseg {struct msg_msgseg* next;/* the next part of the message follows immediately */};
分配一页,第一个除去msg_msg其他的用于保存消息,如果不够继续分配一页除去msg_msg_seq以此类推
//标识号 发送格式 //大小 //标志位asmlinkage long sys_msgsnd (int msqid, struct msgbuf *msgp, size_t msgsz, int msgflg){struct msg_queue *msq;//队列头struct msg_msg *msg;//内核保存信息的格式long mtype;int err;//消息不可以超过8kif (msgsz > msg_ctlmax || (long) msgsz < 0 || msqid < 0)return -EINVAL;if (get_user(mtype, &msgp->mtype))//从用户空间拷贝到内核return -EFAULT;if (mtype < 1)return -EINVAL;msg = load_msg(msgp->mtext, msgsz);//分配缓冲区保存消息(从用户拷贝到内核)if(IS_ERR(msg))return PTR_ERR(msg);msg->m_type = mtype;//消息类型msg->m_ts = msgsz;//消息大小msq = msg_lock(msqid);//根据给定的标号msg_msg找到相应的报文队列,将其数据结构上锁err=-EINVAL;if(msq==NULL)goto out_free;retry:err= -EIDRM;if (msg_checkid(msq,msqid))//验证下id号goto out_unlock_free;err=-EACCES;if (ipcperms(&msq->q_perm, S_IWUGO)) //检查是否有权限向这个队列发送报文goto out_unlock_free;//当前报文大小+当前队列统计的字节数超过了报文队列的总容量.或者报文的个数超过了限制,那就不可以发送了if(msgsz + msq->q_cbytes > msq->q_qbytes ||1 + msq->q_qnum > msq->q_qbytes) {struct msg_sender s;if(msgflg&IPC_NOWAIT) {//是否等待,不等待直接退出err=-EAGAIN;goto out_unlock_free;}ss_add(msq, &s);//挂载到报文队列q_sender链,这样可以通过此链找到休眠正在等待发送的进程msg_unlock(msqid);schedule();//调度current->state= TASK_RUNNING;msq = msg_lock(msqid);err = -EIDRM;if(msq==NULL)goto out_free;ss_del(&s);//删除if (signal_pending(current)) {err=-EINTR;goto out_unlock_free;}goto retry;//重新运行一遍}if(!pipelined_send(msq,msg)) {//如果有相关进程正在读这个报文就不用放入队列了/* noone is waiting for this message, enqueue it */list_add_tail(&msg->m_list,&msq->q_messages);//链入队列msq->q_cbytes += msgsz;//总数++msq->q_qnum++;//数目++atomic_add(msgsz,&msg_bytes);atomic_inc(&msg_hdrs);}err = 0;msg = NULL;msq->q_lspid = current->pid;msq->q_stime = CURRENT_TIME;out_unlock_free:msg_unlock(msqid);out_free:if(msg!=NULL)free_msg(msg);return err;}
load_msg用于消息分割,在内核保存,见上图
//报文的源static struct msg_msg* load_msg(void* src, int len){struct msg_msg* msg;struct msg_msgseg** pseg;int err;int alen;alen = len;if(alen > DATALEN_MSG)//一页减去msg结构大小alen = DATALEN_MSG;msg = (struct msg_msg *) kmalloc (sizeof(*msg) + alen, GFP_KERNEL);//一个报文的头if(msg==NULL)return ERR_PTR(-ENOMEM);msg->next = NULL;if (copy_from_user(msg+1, src, alen)) {//从msg的尾部开始拷贝err = -EFAULT;goto out_err;}len -= alen;//剩余长度src = ((char*)src)+alen;//剩余源头pseg = &msg->next;//指向下一个页while(len > 0) {//还有剩余长度struct msg_msgseg* seg;alen = len;if(alen > DATALEN_SEG)//是否超过page-msgsegalen = DATALEN_SEG;seg = (struct msg_msgseg *) kmalloc (sizeof(*seg) + alen, GFP_KERNEL);//获取一页if(seg==NULL) {err=-ENOMEM;goto out_err;}*pseg = seg;//链接seg->next = NULL;if(copy_from_user (seg+1, src, alen)) {//继续拷贝err = -EFAULT;goto out_err;}pseg = &seg->next;len -= alen;src = ((char*)src)+alen;}return msg;out_err:free_msg(msg);return ERR_PTR(err);}
/* one msg_sender for each sleeping sender */struct msg_sender {struct list_head list;struct task_struct* tsk;};
ss_add(msq, &s);//挂载到报文队列q_sender链,这样可以通过此链找到休眠正在等待发送的进程
static inline void ss_add(struct msg_queue* msq, struct msg_sender* mss){mss->tsk=current;//获取当前队列的进程运行的进程current->state=TASK_INTERRUPTIBLE;//设置为可中断睡眠状态list_add_tail(&mss->list,&msq->q_senders);//添加到队尾}
如果有相关进程正在读这个报文就不用放入队列了
int inline pipelined_send(struct msg_queue* msq, struct msg_msg* msg){struct list_head* tmp;tmp = msq->q_receivers.next;//聚集正在睡眠等待接收的读进程while (tmp != &msq->q_receivers) {//表示有struct msg_receiver* msr;msr = list_entry(tmp,struct msg_receiver,r_list);tmp = tmp->next;if(testmsg(msg,msr->r_msgtype,msr->r_mode)) {//类型是否匹配list_del(&msr->r_list);if(msr->r_maxsize < msg->m_ts) {//读的缓冲区是否够用msr->r_msg = ERR_PTR(-E2BIG);wake_up_process(msr->r_tsk);//不够用则将进程唤醒,让其出错返回} else {msr->r_msg = msg;//有的话,直接读取msq->q_lspid = msr->r_tsk->pid;msq->q_rtime = CURRENT_TIME;wake_up_process(msr->r_tsk);return 1;}}}return 0;}
消息队列接收msgrcv
//标识号 用户空间缓冲区 大小asmlinkage long sys_msgrcv (int msqid, struct msgbuf *msgp, size_t msgsz,long msgtyp, int msgflg)//msgtyp消息类型,{struct msg_queue *msq;//队列头struct msg_receiver msr_d;//接收的需要睡眠的进程struct list_head* tmp;struct msg_msg* msg, *found_msg;int err;int mode;if (msqid < 0 || (long) msgsz < 0)return -EINVAL;mode = convert_mode(&msgtyp,msgflg);msq = msg_lock(msqid);//根据报文队列标识号,找到具体队列if(msq==NULL)return -EINVAL;retry:err=-EACCES;if (ipcperms (&msq->q_perm, S_IRUGO))//检测是否具有权限goto out_unlock;tmp = msq->q_messages.next;found_msg=NULL;while (tmp != &msq->q_messages) {//遍历完msg = list_entry(tmp,struct msg_msg,m_list);//根据队列当前项找到其指针if(testmsg(msg,msgtyp,mode)) {found_msg = msg;//查找到了消息if(mode == SEARCH_LESSEQUAL && msg->m_type != 1) {found_msg=msg;msgtyp=msg->m_type-1;//将type减到比这个报文的类型值更小,看能否找到更小的} else {found_msg=msg;break;}}tmp = tmp->next;}if(found_msg) {msg=found_msg;if ((msgsz < msg->m_ts) && !(msgflg & MSG_NOERROR)) {//如果接收大小小于报文大小,出错err=-E2BIG;goto out_unlock;}list_del(&msg->m_list);//否则将该报文从队列脱链msq->q_qnum--;msq->q_rtime = CURRENT_TIME;msq->q_lrpid = current->pid;msq->q_cbytes -= msg->m_ts;atomic_sub(msg->m_ts,&msg_bytes);atomic_dec(&msg_hdrs);ss_wakeup(&msq->q_senders,0);//将发送的睡眠等待进程全部唤醒,因为拿出了一个报文msg_unlock(msqid);out_success:msgsz = (msgsz > msg->m_ts) ? msg->m_ts : msgsz;if (put_user (msg->m_type, &msgp->mtype) ||//实际接收的报文类型,通过put_user送回用户空间store_msg(msgp->mtext, msg, msgsz)) {//将实际接收到的复制到用户空间msgsz = -EFAULT;}free_msg(msg);//释放内核缓存return msgsz;} else//报文队列还没有报文可供接收{struct msg_queue *t;/* no message waiting. Prepare for pipelined* receive.*/if (msgflg & IPC_NOWAIT) {//不等待直接返回错误err=-ENOMSG;goto out_unlock;}list_add_tail(&msr_d.r_list,&msq->q_receivers);//链入等待接收队列进程msr_d.r_tsk = current;//保存各种信息msr_d.r_msgtype = msgtyp;msr_d.r_mode = mode;if(msgflg & MSG_NOERROR)msr_d.r_maxsize = INT_MAX;elsemsr_d.r_maxsize = msgsz;msr_d.r_msg = ERR_PTR(-EAGAIN);current->state = TASK_INTERRUPTIBLE;msg_unlock(msqid);//解锁并调度//当前进程一旦睡下,以下需要等待进程通过pipelined_send()向其发送报文,并且选择这个进程作为接收进程才会被唤醒schedule();current->state = TASK_RUNNING;msg = (struct msg_msg*) msr_d.r_msg;if(!IS_ERR(msg)) //表示已经成功接收goto out_success;//以下是因为缓冲区太小,唤醒了睡眠进程依旧无法接收,而是被信号唤醒的错误处理t = msg_lock(msqid);//对报文加锁,隐藏着等待,可能被其他进程抢先锁住该队列if(t==NULL)msqid=-1;msg = (struct msg_msg*)msr_d.r_msg;if(!IS_ERR(msg)) {//在锁住队列之前,还有可能接收到其他进程pipelined_send发来的报文/* our message arived while we waited for* the spinlock. Process it.*///所以还需要检查下是否成功接收到报文if(msqid!=-1)msg_unlock(msqid);goto out_success;}err = PTR_ERR(msg);if(err == -EAGAIN) {//要将本进程的msg_receiver结构拖链,并且看是否有信号处理if(msqid==-1)BUG();list_del(&msr_d.r_list);if (signal_pending(current))//如果没有信号处理,则跳转到retry重新开始err=-EINTR;elsegoto retry;}}out_unlock:if(msqid!=-1)msg_unlock(msqid);return err;}
msgctl()报文机制的控制与设置
管道的缺点在于缺乏对管道的控制手段,也缺乏获取其状态的信息(例如,有多少个字节已经在管道中等待读取)的手段,msgctl就是为报文队列提供了这样的手段
long sys_msgctl(int msqid,int cmd,struct msqid_ds*buf);
cmd为具体的操作码
/** Control commands used with semctl, msgctl and shmctl* see also specific commands in sem.h, msg.h and shm.h*/#define IPC_RMID 0 /* remove resource关闭报文队列 */#define IPC_SET 1 /* set ipc_perm options 改变ipc设施的相关状态与属性*/#define IPC_STAT 2 /* get ipc_perm options 获取状态*/#define IPC_INFO 3 /* see ipcs 统计信息*/
以上命令码并不是专门为报文队列设置,也适用于sysv ipc的其他两种机制,对于具体的机制还可能补充其他专用命令.
msg专用命令
/* ipcs ctl commands */#define MSG_STAT 11#define MSG_INFO 12
调用参数buf是一个msgid_ds结构指针,此结构适用于ipc_stat与ipc_set.
/* Obsolete, used only for backwards compatibility and libc5 compiles */struct msqid_ds {struct ipc_perm msg_perm;struct msg *msg_first; /* first message on queue,unused */struct msg *msg_last; /* last message in queue,unused */__kernel_time_t msg_stime; /* last msgsnd time */__kernel_time_t msg_rtime; /* last msgrcv time */__kernel_time_t msg_ctime; /* last change time */unsigned long msg_lcbytes; /* Reuse junk fields for 32 bit */unsigned long msg_lqbytes; /* ditto */unsigned short msg_cbytes; /* current number of bytes on queue */unsigned short msg_qnum; /* number of messages in queue */unsigned short msg_qbytes; /* max number of bytes on queue */__kernel_ipc_pid_t msg_lspid; /* pid of last msgsnd */__kernel_ipc_pid_t msg_lrpid; /* last receive pid */};
当命令为ipc_info时,指向一个msginfo结构
/* buffer for msgctl calls IPC_INFO, MSG_INFO */struct msginfo {int msgpool;int msgmap;int msgmax;int msgmnb;int msgmni;int msgssz;int msgtql;unsigned short msgseg;};
asmlinkage long sys_msgctl (int msqid, int cmd, struct msqid_ds *buf){int err, version;struct msg_queue *msq;struct msq_setbuf setbuf;struct kern_ipc_perm *ipcp;if (msqid < 0 || cmd < 0)return -EINVAL;version = ipc_parse_version(&cmd);//判断是64位版本还是32位版本switch (cmd) {//根据不同类型选择不同操作case IPC_INFO: //合在一起case MSG_INFO:{struct msginfo msginfo;int max_id;if (!buf)return -EFAULT;/* We must not return kernel stack data.* due to padding, it's not enough* to set all member fields.*/memset(&msginfo,0,sizeof(msginfo));msginfo.msgmni = msg_ctlmni;msginfo.msgmax = msg_ctlmax;msginfo.msgmnb = msg_ctlmnb;msginfo.msgssz = MSGSSZ;msginfo.msgseg = MSGSEG;down(&msg_ids.sem);if (cmd == MSG_INFO) {msginfo.msgpool = msg_ids.in_use;msginfo.msgmap = atomic_read(&msg_hdrs);msginfo.msgtql = atomic_read(&msg_bytes);} else {msginfo.msgmap = MSGMAP;msginfo.msgpool = MSGPOOL;msginfo.msgtql = MSGTQL;}max_id = msg_ids.max_id;up(&msg_ids.sem);if (copy_to_user (buf, &msginfo, sizeof(struct msginfo)))//从内核拷贝到用户空间return -EFAULT;return (max_id < 0) ? 0: max_id;}case MSG_STAT://stat状态相关操作case IPC_STAT:{struct msqid64_ds tbuf;int success_return;if (!buf)return -EFAULT;if(cmd == MSG_STAT && msqid > msg_ids.size)return -EINVAL;memset(&tbuf,0,sizeof(tbuf));msq = msg_lock(msqid);if (msq == NULL)return -EINVAL;if(cmd == MSG_STAT) {success_return = msg_buildid(msqid, msq->q_perm.seq);} else {err = -EIDRM;if (msg_checkid(msq,msqid))//id检查goto out_unlock;success_return = 0;}err = -EACCES;if (ipcperms (&msq->q_perm, S_IRUGO))//权限判断goto out_unlock;kernel_to_ipc64_perm(&msq->q_perm, &tbuf.msg_perm);tbuf.msg_stime = msq->q_stime;tbuf.msg_rtime = msq->q_rtime;tbuf.msg_ctime = msq->q_ctime;tbuf.msg_cbytes = msq->q_cbytes;tbuf.msg_qnum = msq->q_qnum;tbuf.msg_qbytes = msq->q_qbytes;tbuf.msg_lspid = msq->q_lspid;tbuf.msg_lrpid = msq->q_lrpid;msg_unlock(msqid);if (copy_msqid_to_user(buf, &tbuf, version))//从内核拷贝到用户return -EFAULT;return success_return;}case IPC_SET://set命令操作if (!buf)return -EFAULT;if (copy_msqid_from_user (&setbuf, buf, version))//从用户拷贝到内核return -EFAULT;break;case IPC_RMID:break;default:return -EINVAL;}down(&msg_ids.sem);msq = msg_lock(msqid);err=-EINVAL;if (msq == NULL)goto out_up;err = -EIDRM;if (msg_checkid(msq,msqid))goto out_unlock_up;ipcp = &msq->q_perm;err = -EPERM;if (current->euid != ipcp->cuid &¤t->euid != ipcp->uid && !capable(CAP_SYS_ADMIN))//是否有权限操作/* We _could_ check for CAP_CHOWN above, but we don't */goto out_unlock_up;switch (cmd) {case IPC_SET:{if (setbuf.qbytes > msg_ctlmnb && !capable(CAP_SYS_RESOURCE))goto out_unlock_up;msq->q_qbytes = setbuf.qbytes;ipcp->uid = setbuf.uid;ipcp->gid = setbuf.gid;ipcp->mode = (ipcp->mode & ~S_IRWXUGO) |(S_IRWXUGO & setbuf.mode);msq->q_ctime = CURRENT_TIME;/* sleeping receivers might be excluded by* stricter permissions.*/expunge_all(msq,-EAGAIN);//使所有正在等待此队列接收报文的进程都出错返回/* sleeping senders might be able to send* due to a larger queue size.*/ss_wakeup(&msq->q_senders,0);//将所有正在等待此队列发送报文的进程都唤醒,进行新一轮尝试msg_unlock(msqid);break;}case IPC_RMID:freeque (msqid); //将所有正在等待的发送还是接收的进程全部唤醒,让其出错返回break;}err = 0;out_up:up(&msg_ids.sem);return err;out_unlock_up:msg_unlock(msqid);goto out_up;out_unlock:msg_unlock(msqid);return err;}
ipc_rmid对应操作:freeque (msqid); //将所有正在等待的发送还是接收的进程全部唤醒,让其出错返回
static void freeque (int id){struct msg_queue *msq;struct list_head *tmp;msq = msg_rmid(id);expunge_all(msq,-EIDRM);//将所有读写的从链表拖链,让其出错返回ss_wakeup(&msq->q_senders,1);//唤醒msg_unlock(id);tmp = msq->q_messages.next;while(tmp != &msq->q_messages) {//将所有报文释放struct msg_msg* msg = list_entry(tmp,struct msg_msg,m_list);tmp = tmp->next;atomic_dec(&msg_hdrs);free_msg(msg);}atomic_sub(msq->q_cbytes, &msg_bytes);kfree(msq);//是否队列头}