一起撸linux内核24-内核IPC机制(02)

SYSCALL_DEFINE2(msgget, key_t, key, int, msgflg)
{
  struct ipc_namespace *ns;
  static const struct ipc_ops msg_ops = {
    .getnew = newque,
    .associate = msg_security,
  };
  struct ipc_params msg_params;


  获取当前进程namespace中的ipc命令空间
//  struct nsproxy {
//    atomic_t count;
//    struct uts_namespace *uts_ns;//unix timesharing system包含内存、版本、底层体系架构
//    struct ipc_namespace *ipc_ns;//ipc空间
//    struct mnt_namespace *mnt_ns;//挂载的文件系统
//    struct pid_namespace *pid_ns_for_children;//进程pid相关
//    struct net     *net_ns;//网络命名空间
//  };
  ns = current->nsproxy->ipc_ns;


  msg_params.key = key;
  msg_params.flg = msgflg;
  获取一个ipc对象
  return ipcget(ns, &msg_ids(ns), &msg_ops, &msg_params);
}

SYSCALL_DEFINE4(msgsnd, int, msqid, struct msgbuf __user *, msgp, size_t, msgsz,
    int, msgflg)
{
  long mtype;
  检测消息type并拷贝给mtype
  if (get_user(mtype, &msgp->mtype))
    return -EFAULT;
  将用户态的消息内容拷贝到内核空间
  return do_msgsnd(msqid, mtype, msgp->mtext, msgsz, msgflg);
}

long do_msgsnd(int msqid, long mtype, void __user *mtext,
    size_t msgsz, int msgflg)
{
  struct msg_queue *msq;
  struct msg_msg *msg;
  int err;
  struct ipc_namespace *ns;
  获取当前进程ipc的命名空间
  ns = current->nsproxy->ipc_ns;


  //该消息长度超过一条消息总长8192字节则返回错误
  //消息长度小于0或者msqid小于0均返回错误
  if (msgsz > ns->msg_ctlmax || (long) msgsz < 0 || msqid < 0)
    return -EINVAL;
  //消息类型不能小于1
  if (mtype < 1)
    return -EINVAL;
  //申请一个消息空间并将消息写入该内存
  msg = load_msg(mtext, msgsz);
  if (IS_ERR(msg))
    return PTR_ERR(msg);
  //设置消息类型
  msg->m_type = mtype;
  msg->m_ts = msgsz;


  rcu_read_lock();
  //做一些安全检测
  … …
  for (;;) {
    struct msg_sender s;
    … …
    //消息队列满,发送参数有IPC_NOWAIT,则返回错误
    /* queue full, wait: */
    if (msgflg & IPC_NOWAIT) {
      err = -EAGAIN;
      goto out_unlock0;
    }
    … …
  }
  msq->q_lspid = task_tgid_vnr(current);
  msq->q_stime = get_seconds();


  //查看是否有接收进程接收该消息,如果有则直接发送
  if (!pipelined_send(msq, msg)) {
    /* no one 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, &ns->msg_bytes);
    atomic_inc(&ns->msg_hdrs);
  }


  err = 0;
  msg = NULL;


out_unlock0:
  ipc_unlock_object(&msq->q_perm);
out_unlock1:
  rcu_read_unlock();
  if (msg != NULL)
    free_msg(msg);
  return err;
}

static inline int pipelined_send(struct msg_queue *msq, struct msg_msg *msg)
{
  struct msg_receiver *msr, *t;


  list_for_each_entry_safe(msr, t, &msq->q_receivers, r_list) {
    if (testmsg(msg, msr->r_msgtype, msr->r_mode) &&
        !security_msg_queue_msgrcv(msq, msg, msr->r_tsk,
                 msr->r_msgtype, msr->r_mode)) {
      //轮询消息队列中的每个q_receivers，如果type一致则尝试发送
      //上述security_msg_queue_msgrcv发送成功则进行后续处理
      list_del(&msr->r_list);
      //消息太大,唤醒对应的接收进程并返回接收失败
      if (msr->r_maxsize < msg->m_ts) {
        /* initialize pipelined send ordering */
        msr->r_msg = NULL;
        wake_up_process(msr->r_tsk);
        /* barrier (B) see barrier comment below */
        //内存屏障-多处理器相关，避免编译器优化代码
        smp_wmb();
        msr->r_msg = ERR_PTR(-E2BIG);
      } else {
        msr->r_msg = NULL;
        msq->q_lrpid = task_pid_vnr(msr->r_tsk);
        msq->q_rtime = get_seconds();
        //发送成功唤醒接收进程
        wake_up_process(msr->r_tsk);
        /*
         * Ensure that the wakeup is visible before
         * setting r_msg, as the receiving can otherwise
         * exit - once r_msg is set, the receiver can
         * continue. See lockless receive part 1 and 2
         * in do_msgrcv(). Barrier (B).
         */
        smp_wmb();
        msr->r_msg = msg;


        return 1;
      }
    }
  }
  return 0;
}