从内核看socketpair的实现（基于5.9.9）

theanarkh

发布于 2021-07-08 16:05:35

5920

发布于 2021-07-08 16:05:35

文章被收录于专栏：原创分享

前言：本文介绍socketpair的实现和通信。

Unix域支持服务器、客户端的模式，这种模式的好处就是任意进程都可以和服务器进程通信，这种模式通常需要一个文件路径作为地址，使得任意进程都能通过该文件路径标识找到服务器地址。而socketpair虽然也类似，但它不需要地址的概念，因为它用于有继承关系的进程间通信，通常是主进程调用socketpair拿到两个fd，然后fork出子进程，这样两个进程就可以通信了，不需要寻址的过程，也就不需要地址的概念了。下面我从内核角度看看socketpair的实现。

// 分配两个关联的socketint __sys_socketpair(int family, int type, int protocol, int __user *usockvec){    struct socket *sock1, *sock2;
    int fd1, fd2, err;
    struct file *newfile1, *newfile2;
    int flags;
    // 参数校验
    flags = type & ~SOCK_TYPE_MASK;
    if (flags & ~(SOCK_CLOEXEC | SOCK_NONBLOCK))
        return -EINVAL;
    type &= SOCK_TYPE_MASK;

    if (SOCK_NONBLOCK != O_NONBLOCK && (flags & SOCK_NONBLOCK))
        flags = (flags & ~SOCK_NONBLOCK) | O_NONBLOCK;
    // 获取两个fd
    fd1 = get_unused_fd_flags(flags);
    fd2 = get_unused_fd_flags(flags);
    // 获取两个socket
    err = sock_create(family, type, protocol, &sock1);
    err = sock_create(family, type, protocol, &sock2);
    // 调用钩子函数
    err = sock1->ops->socketpair(sock1, sock2);
    // 分配两个file，每个file和socket关联
    newfile1 = sock_alloc_file(sock1, flags, NULL);
    newfile2 = sock_alloc_file(sock2, flags, NULL);
    // 关联fd和file
    fd_install(fd1, newfile1);
    fd_install(fd2, newfile2);
    return 0;}

我们知道在网络的实现中，为了符合虚拟文件系统的设计，socket的设计中，也遵循fd->file->inode的结构，__sys_socketpair就是申请了两份这样的数据结构，然后调用钩子函数socketpair。再调用socketpair之前，架构如下。

我们接下来看socketpair钩子函数的实现。

static int unix_socketpair(struct socket *socka, struct socket *sockb){
    struct sock *ska = socka->sk, *skb = sockb->sk;

    sock_hold(ska);
    sock_hold(skb);
    // 互相关联
    unix_peer(ska) = skb;
    unix_peer(skb) = ska;
    init_peercred(ska);
    init_peercred(skb);
    return 0;}

我们看到socketpair的实现非常简单，就是把两个socket关联起来，这时候的架构如下。

下面我们看看数据通信，以数据报模式为例。

static int unix_dgram_sendmsg(struct socket *sock, struct msghdr *msg,
                  size_t len){
    struct sock *sk = sock->sk;
    struct net *net = sock_net(sk);
    struct unix_sock *u = unix_sk(sk);
    struct sock *other = NULL;
    // 找到对端socket
    other = unix_peer_get(sk);
    // 单个包不能太大
    err = -EMSGSIZE;
    if (len > sk->sk_sndbuf - 32)
        goto out;
    // 分配个skb承载消息
    skb = sock_alloc_send_pskb(sk, len - data_len, data_len,
                   msg->msg_flags & MSG_DONTWAIT, &err,
                   PAGE_ALLOC_COSTLY_ORDER);

    skb->data_len = data_len;
    skb->len = len;
    // 把数据从msg复制到skb
    err = skb_copy_datagram_from_iter(skb, 0, &msg->msg_iter, len);
    // 阻塞时的阻塞时间
    timeo = sock_sndtimeo(sk, msg->msg_flags & MSG_DONTWAIT);

    // 接收队列满了并且设置了超时时间则阻塞
    if (other != sk &&
        unlikely(unix_peer(other) != sk &&
        unix_recvq_full_lockless(other))) {
        if (timeo) {
            // 阻塞
            timeo = unix_wait_for_peer(other, timeo);
            err = sock_intr_errno(timeo);
            if (signal_pending(current))
                goto out_free;

            goto restart;
        }
    }
    // 插入消息队列
    skb_queue_tail(&other->sk_receive_queue, skb);
    // 唤醒对端，对端可能在阻塞
    other->sk_data_ready(other);
    return len;}

我们看到unix_dgram_sendmsg的逻辑主要是分配一个承载消息的skb结构体，然后通过关联关系找到对端，最后插入到对端的消息队列，下面我们看接收端如何处理，对应函数是unix_dgram_recvmsg。

static int unix_dgram_recvmsg(struct socket *sock, struct msghdr *msg,
                  size_t size, int flags){
    struct scm_cookie scm;
    struct sock *sk = sock->sk;
    struct unix_sock *u = unix_sk(sk);
    struct sk_buff *skb, *last;
    long timeo;
    int skip;
    int err;
    // 没数据时的阻塞时长，可以设置为非阻塞
    timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);

    do {
        mutex_lock(&u->iolock);
        // 跳过的字节
        skip = sk_peek_offset(sk, flags);
        // 从接收队列中获取一个节点
        skb = __skb_try_recv_datagram(sk, &sk->sk_receive_queue, flags,
                          &skip, &err, &last);
        mutex_unlock(&u->iolock);
        // 成功则break
        if (err != -EAGAIN)
            break;
    // 没有数据则判断是否要阻塞，阻塞的话看多久
    } while (timeo &&
         !__skb_wait_for_more_packets(sk, &sk->sk_receive_queue,
                          &err, &timeo, last));
    // 是否有等待可写的对端（因为接收队列满了），有则唤醒
    if (wq_has_sleeper(&u->peer_wait))
        wake_up_interruptible_sync_poll(&u->peer_wait,
                        EPOLLOUT | EPOLLWRNORM |
                        EPOLLWRBAND);

    // 把skb的数据复制到msg
    skb_copy_datagram_msg(skb, skip, msg, size);}

数据接收就是从消息队列中逐个返回给用户，每次调用返回一个。

后记：无。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2021-06-19，如有侵权请联系 cloudcommunity@tencent.com 删除

kafka