从Java NIO到TCP协议栈：一次Socket读操作的全链路源码剖析

引言

在Java NIO编程中，SocketChannel.read(ByteBuffer)是最常见的网络数据读取接口。当应用程序调用该方法时，一条跨越用户态Java虚拟机、C标准库、Linux内核VFS层、套接字层，最终抵达TCP协议栈的漫长调用链便由此展开。本文基于OpenJDK 17、glibc 2.39和Linux内核6.18的源码，逐层深入，完整还原一次Socket读操作的全过程，解析每一环节的设计意图与实现细节。

一、Java层：从ByteBuffer到本地内存地址

入口是sun.nio.ch.SocketDispatcher的read方法，但实际逻辑封装在父类FileDispatcherImpl的私有静态方法readIntoNativeBuffer中：

java

private static int readIntoNativeBuffer(FileDescriptor fd, ByteBuffer bb, long position, boolean directIO, boolean async, int alignment, NativeDispatcher nd) throws IOException { int pos = bb.position(); int lim = bb.limit(); int rem = (pos <= lim ? lim - pos : 0); if (directIO) { Util.checkBufferPositionAligned(bb, pos, alignment); Util.checkRemainingBufferSizeAligned(rem, alignment); } if (rem == 0) return 0; int n = 0; var handle = acquireScope(bb, async); try { if (position != -1) { n = nd.pread(fd, bufferAddress(bb) + pos, rem, position); } else { n = nd.read(fd, bufferAddress(bb) + pos, rem); } } finally { releaseScope(handle); } if (n > 0) bb.position(pos + n); return n; }

关键点解析：

• 缓冲区状态管理：记录当前position和limit，计算剩余可读字节数rem。若directIO开启（如使用O_DIRECT标志的文件），则通过Util工具类验证缓冲区地址和长度是否与磁盘块大小对齐，这是直接I/O的硬件强制要求。

• 内存作用域控制：acquireScope与releaseScope是Java 17ScopedMemoryAccess机制的一部分。在异步I/O或多线程环境下，该机制防止ByteBuffer所引用的堆外内存被垃圾回收器过早回收，确保底层C代码操作的内核地址始终有效。

• 读写派发：由于套接字不支持随机访问，position始终为-1，因此调用nd.read(fd, bufferAddress(bb) + pos, rem)。其中bufferAddress(bb)返回DirectByteBuffer的堆外内存起始地址（long类型），该地址将作为参数传递给JNI层。

NativeDispatcher.read由具体子类实现，SocketDispatcher的实现十分简洁：

java

int read(FileDescriptor fd, long address, int len) throws IOException { return read0(fd, address, len); } private static native int read0(FileDescriptor fd, long address, int len) throws IOException;

至此，Java层的准备工作完成，控制权转入本地方法。

二、JNI层：参数转换与系统调用封装

JNI函数Java_sun_nio_ch_SocketDispatcher_read0负责将Java对象转换为C语言可用的数据类型：

c

JNIEXPORT jint JNICALL Java_sun_nio_ch_SocketDispatcher_read0(JNIEnv *env, jclass clazz, jobject fdo, jlong address, jint len) { jint fd = fdval(env, fdo); // 从Java FileDescriptor对象中取出整型fd void *buf = (void *)jlong_to_ptr(address); jint n = read(fd, buf, len); // 调用glibc的read if ((n == -1) && (errno == ECONNRESET || errno == EPIPE)) { JNU_ThrowByName(env, "sun/net/ConnectionResetException", "Connection reset"); return IOS_THROWN; } else { return convertReturnVal(env, n, JNI_TRUE); } }

• 文件描述符提取：fdval通过访问FileDescriptor对象的fd字段（int类型）获得内核级别的文件描述符编号。

• 地址转换：jlong_to_ptr将Java的64位地址转换为void*指针，该指针指向DirectByteBuffer的底层内存。

• 系统调用：直接调用read（由glibc提供）。若返回-1且错误码为ECONNRESET（TCP连接重置）或EPIPE（对端关闭写端），则抛出Java层的ConnectionResetException，这是网络编程中常见且需要特殊处理的异常。其他错误由convertReturnVal转换，负值表示I/O异常，正值表示成功读取的字节数。

三、glibc层：取消点与系统调用陷入

glibc中的read实际上是__libc_read的弱别名，其实现核心在于SYSCALL_CANCEL宏：

c

ssize_t __libc_read (int fd, void *buf, size_t nbytes) { return SYSCALL_CANCEL (read, fd, buf, nbytes); } weak_alias (__libc_read, read)

SYSCALL_CANCEL是glibc为支持POSIX线程取消（cancellation）而设计的包装器。它执行以下操作：

1. 检查当前线程的取消状态和类型。

2. 若线程处于可取消状态且取消了异步模式，则设置一个取消点，并在系统调用返回时根据errno判断是否被信号中断。

3. 使用syscall指令（或int $0x80）发起系统调用，陷入内核。

此机制使得Java的InterruptibleChannel能够通过中断线程来取消阻塞的I/O操作，是NIO可中断特性的底层支撑。

四、内核系统调用入口：sys_read与ksys_read

在内核端，系统调用read通过SYSCALL_DEFINE3宏声明：

c

SYSCALL_DEFINE3(read, unsigned int, fd, char __user *, buf, size_t, count) { return ksys_read(fd, buf, count); }

ksys_read负责获取文件对象并调用VFS层的读取函数：

c

ssize_t ksys_read(unsigned int fd, char __user *buf, size_t count) { struct fd f = fdget_pos(fd); ssize_t ret = -EBADF; if (f.file) { loff_t pos = file_pos_read(f.file); ret = vfs_read(f.file, buf, count, &pos); if (ret >= 0) file_pos_write(f.file, pos); fdput_pos(f); } return ret; }

• fdget_pos通过文件描述符编号从当前进程的文件描述符表中获取struct file，并增加引用计数，同时锁定文件位置（file->f_pos）。对于套接字，f_pos虽然无实际意义，但VFS统一管理。

• 调用vfs_read后，若成功则更新文件位置（套接字忽略此更新）。

五、VFS层：权限检查与操作派发

vfs_read是虚拟文件系统读取操作的核心枢纽：

c

ssize_t vfs_read(struct file *file, char __user *buf, size_t count, loff_t *pos) { ssize_t ret; if (!(file->f_mode & FMODE_READ)) return -EBADF; if (!(file->f_mode & FMODE_CAN_READ)) return -EINVAL; if (unlikely(!access_ok(buf, count))) return -EFAULT; ret = rw_verify_area(READ, file, pos, count); if (ret) return ret; if (count > MAX_RW_COUNT) count = MAX_RW_COUNT; if (file->f_op->read) ret = file->f_op->read(file, buf, count, pos); else if (file->f_op->read_iter) ret = new_sync_read(file, buf, count, pos); else ret = -EINVAL; if (ret > 0) { fsnotify_access(file); add_rchar(current, ret); } inc_syscr(current); return ret; }

关键步骤：

• 模式与权限检查：确保文件以读方式打开且具备读取能力。

• 用户地址校验：access_ok验证用户态缓冲区范围是否属于进程地址空间，防止内核访问非法地址。

• 区域检查：rw_verify_area核对读取偏移和长度是否超出文件大小或进程资源限制（如RLIMIT_FSIZE）。

• 操作派发：这是VFS多态性的体现。对于套接字，file->f_op指向socket_file_ops，该结构定义了read_iter成员，而未定义read（传统read已逐渐被read_iter取代）。因此，内核调用new_sync_read，该函数会构造struct kiocb并调用file->f_op->read_iter。

• 统计更新：成功读取后更新文件访问通知、进程I/O计数和系统调用计数。

socket_file_ops的定义片段：

c

static const struct file_operations socket_file_ops = { .read_iter = sock_read_iter, .write_iter = sock_write_iter, .poll = sock_poll, // ... 其他成员 };

因此，VFS将控制权移交给sock_read_iter。

六、套接字层：从VFS到协议无关接收

sock_read_iter将VFS的iov_iter（统一迭代器，封装用户缓冲区）转换为msghdr结构，并调用套接字层接收函数：

c

static ssize_t sock_read_iter(struct kiocb *iocb, struct iov_iter *to) { struct file *file = iocb->ki_filp; struct socket *sock = file->private_data; struct msghdr msg = {.msg_iter = *to, .msg_iocb = iocb}; ssize_t res; if (file->f_flags & O_NONBLOCK || (iocb->ki_flags & IOCB_NOWAIT)) msg.msg_flags = MSG_DONTWAIT; if (iocb->ki_pos != 0) return -ESPIPE; if (!iov_iter_count(to)) return 0; res = sock_recvmsg(sock, &msg, msg.msg_flags); *to = msg.msg_iter; return res; }

• 非阻塞标志传递：从文件标志或kiocb中提取非阻塞设置，转换为MSG_DONTWAIT。

• 偏移量检查：套接字不支持lseek，偏移量必须为0。

• 空数据快速返回：若用户请求长度为零，直接返回0，符合SYSV行为。

• 核心调用：sock_recvmsg，它先调用LSM钩子（security_socket_recvmsg）进行安全审计，然后进入sock_recvmsg_nosec。

sock_recvmsg_nosec通过INDIRECT_CALL_INET宏调用sock->ops->recvmsg。对于AF_INET的流式套接字（TCP），sock->ops为inet_stream_ops，其recvmsg成员指向inet_recvmsg。

七、INET层：协议族通用接收

inet_recvmsg是IPv4协议族套接字的通用接收入口：

c

int inet_recvmsg(struct socket *sock, struct msghdr *msg, size_t size, int flags) { struct sock *sk = sock->sk; int addr_len = 0; int err; if (likely(!(flags & MSG_ERRQUEUE))) sock_rps_record_flow(sk); err = INDIRECT_CALL_2(sk->sk_prot->recvmsg, tcp_recvmsg, udp_recvmsg, sk, msg, size, flags, &addr_len); if (err >= 0) msg->msg_namelen = addr_len; return err; }

• RPS（Receive Packet Steering）记录：如果未指定MSG_ERRQUEUE，则记录该流以优化多核调度。

• 协议派发：通过INDIRECT_CALL_2优化间接调用，优先尝试tcp_recvmsg或udp_recvmsg。对于TCP套接字，sk->sk_prot指向tcp_prot，其recvmsg即为tcp_recvmsg。

• 地址长度返回：若成功，则设置对端地址长度（用于recvfrom风格调用）。

八、TCP核心：tcp_recvmsg_locked的精细处理

tcp_recvmsg加锁后调用tcp_recvmsg_locked，这是整个调用链中最复杂且最核心的函数。我们深度剖析其关键部分。

8.1 初始化和状态校验

c

static int tcp_recvmsg_locked(struct sock *sk, struct msghdr *msg, size_t len, int flags, struct scm_timestamping_internal *tss, int *cmsg_flags) { struct tcp_sock *tp = tcp_sk(sk); int copied = 0; u32 peek_seq, *seq; unsigned long used; int err, target; long timeo; struct sk_buff *skb, *last; u32 urg_hole = 0; if (sk->sk_state == TCP_LISTEN) return -ENOTCONN; // 设置MSG_INQ控制消息（如果用户请求） if (tp->recvmsg_inq) { *cmsg_flags = TCP_CMSG_INQ; msg->msg_get_inq = 1; } timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT); if (flags & MSG_OOB) goto recv_urg; // 处理修复模式（TCP_REPAIR）...

• 监听套接字不可读数据，返回-ENOTCONN。

• 支持MSG_INQ扩展（查询接收队列中待读数据量）。

• 超时时间根据非阻塞标志计算，若MSG_DONTWAIT则超时为0。

• 带外数据（OOB）走单独路径recv_urg。

8.2 主循环：遍历接收队列

c

seq = &tp->copied_seq; if (flags & MSG_PEEK) { peek_seq = tp->copied_seq; seq = &peek_seq; } target = sock_rcvlowat(sk, flags & MSG_WAITALL, len); do { u32 offset; // 紧急数据处理（如果紧急指针到达当前序列号） if (unlikely(tp->urg_data) && tp->urg_seq == *seq) { if (copied) break; if (signal_pending(current)) { copied = -EAGAIN; break; } } last = skb_peek_tail(&sk->sk_receive_queue); skb_queue_walk(&sk->sk_receive_queue, skb) { last = skb; offset = *seq - TCP_SKB_CB(skb)->seq; if (unlikely(TCP_SKB_CB(skb)->tcp_flags & TCPHDR_SYN)) offset--; if (offset < skb->len) goto found_ok_skb; if (TCP_SKB_CB(skb)->tcp_flags & TCPHDR_FIN) goto found_fin_ok; // 如果offset >= skb->len，说明该skb已完全消费，继续下一个 } // 队列中无可用数据...

• copied_seq记录当前已读取并交付给用户的TCP序列号位置。若使用MSG_PEEK窥探模式，则使用临时变量peek_seq。

• target是低水位标记，当已拷贝数据达到该值，可提前返回（除非MSG_WAITALL要求全部读取）。

• 遍历接收队列（sk_receive_queue）中的每个sk_buff。每个skb包含一段TCP数据载荷。通过计算offset（当前读取位置在skb内部的偏移）判断该skb是否还有未读数据。

• 若skb包含SYN标志（很少见），则偏移需减1（SYN占用一个字节序列号但不含数据）。

• 若偏移小于skb长度，则进入数据拷贝分支；若skb带有FIN标志，则处理连接关闭。

8.3 数据拷贝分支（found_ok_skb）

c

found_ok_skb: used = skb->len - offset; if (len < used) used = len; if (unlikely(tp->urg_data)) { u32 urg_offset = tp->urg_seq - *seq; if (urg_offset < used) { if (!urg_offset) { if (!sock_flag(sk, SOCK_URGINLINE)) { WRITE_ONCE(*seq, *seq + 1); urg_hole++; offset++; used--; if (!used) goto skip_copy; } } else used = urg_offset; } } if (!(flags & MSG_TRUNC)) { err = skb_copy_datagram_msg(skb, offset, msg, used); if (err) { if (!copied) copied = -EFAULT; break; } } WRITE_ONCE(*seq, *seq + used); copied += used; len -= used; tcp_rcv_space_adjust(sk);

• 计算本次实际拷贝量used，不超过用户请求len和skb剩余数据。

• 处理紧急数据（URG）：如果紧急指针落在本次拷贝范围内，根据SOCK_URGINLINE标志决定是否将紧急字节内联到普通数据流中。标准TCP行为是跳过紧急字节（即urg_hole），仅通过MSG_OOB读取。

• 实际拷贝：调用skb_copy_datagram_msg，该函数将skb中的数据（可能是线性区域或分页区域）复制到msg->msg_iter指向的用户缓冲区。最终通过copy_to_user完成内核到用户空间的数据搬运。

• 更新copied_seq（或临时窥探指针），累加copied，减少剩余长度，并调用tcp_rcv_space_adjust根据接收速率动态调整接收窗口。

8.4 队列无数据时的等待逻辑

当遍历完整个接收队列仍无可读数据时，进入等待处理：

c

if (copied >= target && !READ_ONCE(sk->sk_backlog.tail)) break; if (copied) { if (!timeo || sk->sk_err || sk->sk_state == TCP_CLOSE || (sk->sk_shutdown & RCV_SHUTDOWN) || signal_pending(current)) break; } else { if (sock_flag(sk, SOCK_DONE)) break; if (sk->sk_err) { copied = sock_error(sk); break; } if (sk->sk_shutdown & RCV_SHUTDOWN) break; if (sk->sk_state == TCP_CLOSE) { copied = -ENOTCONN; break; } if (!timeo) { copied = -EAGAIN; break; } if (signal_pending(current)) { copied = sock_intr_errno(timeo); break; } } if (copied >= target) { __sk_flush_backlog(sk); // 处理backlog队列 } else { tcp_cleanup_rbuf(sk, copied); err = sk_wait_data(sk, &timeo, last); if (err < 0) { err = copied ? : err; goto out; } }

• 若已拷贝数据达到低水位标记，且无积压（backlog），则退出循环。

• 若已拷贝部分数据，则检查超时、错误、关闭、信号等条件决定是否退出。

• 若尚未拷贝任何数据，则必须等待新数据到达，调用sk_wait_data使当前进程进入睡眠（可被信号中断）。唤醒后继续循环。

8.5 FIN处理与收尾

若遇到FIN标志：

c

found_fin_ok: WRITE_ONCE(*seq, *seq + 1); if (!(flags & MSG_PEEK)) tcp_eat_recv_skb(sk, skb); break;

将copied_seq推进一位（FIN消耗一个序列号），并从接收队列移除该skb（非窥探模式）。最终，循环结束后调用tcp_cleanup_rbuf更新接收窗口并可能发送ACK，然后返回copied字节数（或错误码）。

九、数据拷贝的内核实现：skb_copy_datagram_iter

skb_copy_datagram_iter是skb数据拷贝的通用函数，它处理三种数据存储方式：

1. 线性数据区（skb->data，通常存储TCP头部和小量载荷）。

2. 分页区（skb_shinfo(skb)->frags，存储大块数据以节省内存拷贝）。

3. 分片链（frag_list，用于GSO/GRO合并的大包）。

它通过遍历这些区域，对每个片段调用simple_copy_to_iter，该函数最终使用copy_to_user将内核数据复制到用户态缓冲区。这是整个调用链中唯一的一次跨地址空间复制，效率至关重要。

十、总结与整体流程图

我们可以用以下简化的时序图概括全链路：

text

Java应用 -> SocketChannel.read(ByteBuffer) -> readIntoNativeBuffer (获取地址) -> SocketDispatcher.read0 (JNI) -> glibc read (SYSCALL_CANCEL) -> 系统调用 sys_read -> ksys_read (获取file结构) -> vfs_read (权限检查) -> sock_read_iter (构造msghdr) -> sock_recvmsg (安全钩子) -> inet_recvmsg (协议派发) -> tcp_recvmsg (加锁) -> tcp_recvmsg_locked [遍历sk_receive_queue] -> skb_copy_datagram_msg -> copy_to_user (返回用户空间) -> 更新copied_seq/窗口 -> 释放锁

从Java层的缓冲区管理、JNI参数转换，到glibc的取消点封装，再到内核VFS的多态派发、套接字层协议无关接口，最终进入TCP协议栈的复杂接收逻辑——每个阶段都紧密协作，确保了高效、可靠且可中断的网络数据读取。理解这条调用链，不仅有助于定位网络I/O性能瓶颈，更能为编写高并发、低延迟的Java网络应用提供坚实的底层认知基础。

##源码

private static int readIntoNativeBuffer(FileDescriptor fd, ByteBuffer bb, long position, boolean directIO, boolean async, int alignment, NativeDispatcher nd) throws IOException { int pos = bb.position(); int lim = bb.limit(); assert (pos <= lim); int rem = (pos <= lim ? lim - pos : 0); if (directIO) { Util.checkBufferPositionAligned(bb, pos, alignment); Util.checkRemainingBufferSizeAligned(rem, alignment); } if (rem == 0) return 0; int n = 0; var handle = acquireScope(bb, async); try { if (position != -1) { n = nd.pread(fd, bufferAddress(bb) + pos, rem, position); } else { n = nd.read(fd, bufferAddress(bb) + pos, rem); } } finally { releaseScope(handle); } if (n > 0) bb.position(pos + n); return n; } int read(FileDescriptor fd, long address, int len) throws IOException { return read0(fd, address, len); } /** * Reads up to len bytes from a socket with special handling for "connection * reset". * * @throws sun.net.ConnectionResetException if connection reset is detected * @throws IOException if another I/O error occurs */ int read(FileDescriptor fd, long address, int len) throws IOException { return read0(fd, address, len); } private static native int read0(FileDescriptor fd, long address, int len) throws IOException; JNIEXPORT jint JNICALL Java_sun_nio_ch_SocketDispatcher_read0(JNIEnv *env, jclass clazz, jobject fdo, jlong address, jint len) { jint fd = fdval(env, fdo); void *buf = (void *)jlong_to_ptr(address); jint n = read(fd, buf, len); if ((n == -1) && (errno == ECONNRESET || errno == EPIPE)) { JNU_ThrowByName(env, "sun/net/ConnectionResetException", "Connection reset"); return IOS_THROWN; } else { return convertReturnVal(env, n, JNI_TRUE); } } /* Read NBYTES into BUF from FD. Return the number read or -1. */ ssize_t __libc_read (int fd, void *buf, size_t nbytes) { return SYSCALL_CANCEL (read, fd, buf, nbytes); } libc_hidden_def (__libc_read) libc_hidden_def (__read) weak_alias (__libc_read, __read) libc_hidden_def (read) weak_alias (__libc_read, read) SYSCALL_DEFINE3(read, unsigned int, fd, char __user *, buf, size_t, count) { return ksys_read(fd, buf, count); } ssize_t vfs_read(struct file *file, char __user *buf, size_t count, loff_t *pos) { ssize_t ret; if (!(file->f_mode & FMODE_READ)) return -EBADF; if (!(file->f_mode & FMODE_CAN_READ)) return -EINVAL; if (unlikely(!access_ok(buf, count))) return -EFAULT; ret = rw_verify_area(READ, file, pos, count); if (ret) return ret; if (count > MAX_RW_COUNT) count = MAX_RW_COUNT; if (file->f_op->read) ret = file->f_op->read(file, buf, count, pos); else if (file->f_op->read_iter) ret = new_sync_read(file, buf, count, pos); else ret = -EINVAL; if (ret > 0) { fsnotify_access(file); add_rchar(current, ret); } inc_syscr(current); return ret; } /* * Socket files have a set of 'special' operations as well as the generic file ones. These don't appear * in the operation structures but are done directly via the socketcall() multiplexor. */ static const struct file_operations socket_file_ops = { .owner = THIS_MODULE, .llseek = no_llseek, .read_iter = sock_read_iter, .write_iter = sock_write_iter, .poll = sock_poll, .unlocked_ioctl = sock_ioctl, #ifdef CONFIG_COMPAT .compat_ioctl = compat_sock_ioctl, #endif .uring_cmd = io_uring_cmd_sock, .mmap = sock_mmap, .release = sock_close, .fasync = sock_fasync, .splice_write = splice_to_socket, .splice_read = sock_splice_read, .splice_eof = sock_splice_eof, .show_fdinfo = sock_show_fdinfo, }; static ssize_t sock_read_iter(struct kiocb *iocb, struct iov_iter *to) { struct file *file = iocb->ki_filp; struct socket *sock = file->private_data; struct msghdr msg = {.msg_iter = *to, .msg_iocb = iocb}; ssize_t res; if (file->f_flags & O_NONBLOCK || (iocb->ki_flags & IOCB_NOWAIT)) msg.msg_flags = MSG_DONTWAIT; if (iocb->ki_pos != 0) return -ESPIPE; if (!iov_iter_count(to)) /* Match SYS5 behaviour */ return 0; res = sock_recvmsg(sock, &msg, msg.msg_flags); *to = msg.msg_iter; return res; } /** * sock_recvmsg - receive a message from @sock * @sock: socket * @msg: message to receive * @flags: message flags * * Receives @msg from @sock, passing through LSM. Returns the total number * of bytes received, or an error. */ int sock_recvmsg(struct socket *sock, struct msghdr *msg, int flags) { int err = security_socket_recvmsg(sock, msg, msg_data_left(msg), flags); return err ?: sock_recvmsg_nosec(sock, msg, flags); } EXPORT_SYMBOL(sock_recvmsg); static inline int sock_recvmsg_nosec(struct socket *sock, struct msghdr *msg, int flags) { int ret = INDIRECT_CALL_INET(READ_ONCE(sock->ops)->recvmsg, inet6_recvmsg, inet_recvmsg, sock, msg, msg_data_left(msg), flags); if (trace_sock_recv_length_enabled()) call_trace_sock_recv_length(sock->sk, ret, flags); return ret; } int inet_recvmsg(struct socket *sock, struct msghdr *msg, size_t size, int flags) { struct sock *sk = sock->sk; int addr_len = 0; int err; if (likely(!(flags & MSG_ERRQUEUE))) sock_rps_record_flow(sk); err = INDIRECT_CALL_2(sk->sk_prot->recvmsg, tcp_recvmsg, udp_recvmsg, sk, msg, size, flags, &addr_len); if (err >= 0) msg->msg_namelen = addr_len; return err; } EXPORT_SYMBOL(inet_recvmsg); int tcp_recvmsg(struct sock *sk, struct msghdr *msg, size_t len, int flags, int *addr_len) { int cmsg_flags = 0, ret; struct scm_timestamping_internal tss; if (unlikely(flags & MSG_ERRQUEUE)) return inet_recv_error(sk, msg, len, addr_len); if (sk_can_busy_loop(sk) && skb_queue_empty_lockless(&sk->sk_receive_queue) && sk->sk_state == TCP_ESTABLISHED) sk_busy_loop(sk, flags & MSG_DONTWAIT); lock_sock(sk); ret = tcp_recvmsg_locked(sk, msg, len, flags, &tss, &cmsg_flags); release_sock(sk); if ((cmsg_flags || msg->msg_get_inq) && ret >= 0) { if (cmsg_flags & TCP_CMSG_TS) tcp_recv_timestamp(msg, sk, &tss); if (msg->msg_get_inq) { msg->msg_inq = tcp_inq_hint(sk); if (cmsg_flags & TCP_CMSG_INQ) put_cmsg(msg, SOL_TCP, TCP_CM_INQ, sizeof(msg->msg_inq), &msg->msg_inq); } } return ret; } EXPORT_SYMBOL(tcp_recvmsg); /* * This routine copies from a sock struct into the user buffer. * * Technical note: in 2.3 we work on _locked_ socket, so that * tricks with *seq access order and skb->users are not required. * Probably, code can be easily improved even more. */ static int tcp_recvmsg_locked(struct sock *sk, struct msghdr *msg, size_t len, int flags, struct scm_timestamping_internal *tss, int *cmsg_flags) { struct tcp_sock *tp = tcp_sk(sk); int copied = 0; u32 peek_seq; u32 *seq; unsigned long used; int err; int target; /* Read at least this many bytes */ long timeo; struct sk_buff *skb, *last; u32 urg_hole = 0; err = -ENOTCONN; if (sk->sk_state == TCP_LISTEN) goto out; if (tp->recvmsg_inq) { *cmsg_flags = TCP_CMSG_INQ; msg->msg_get_inq = 1; } timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT); /* Urgent data needs to be handled specially. */ if (flags & MSG_OOB) goto recv_urg; if (unlikely(tp->repair)) { err = -EPERM; if (!(flags & MSG_PEEK)) goto out; if (tp->repair_queue == TCP_SEND_QUEUE) goto recv_sndq; err = -EINVAL; if (tp->repair_queue == TCP_NO_QUEUE) goto out; /* 'common' recv queue MSG_PEEK-ing */ } seq = &tp->copied_seq; if (flags & MSG_PEEK) { peek_seq = tp->copied_seq; seq = &peek_seq; } target = sock_rcvlowat(sk, flags & MSG_WAITALL, len); do { u32 offset; /* Are we at urgent data? Stop if we have read anything or have SIGURG pending. */ if (unlikely(tp->urg_data) && tp->urg_seq == *seq) { if (copied) break; if (signal_pending(current)) { copied = timeo ? sock_intr_errno(timeo) : -EAGAIN; break; } } /* Next get a buffer. */ last = skb_peek_tail(&sk->sk_receive_queue); skb_queue_walk(&sk->sk_receive_queue, skb) { last = skb; /* Now that we have two receive queues this * shouldn't happen. */ if (WARN(before(*seq, TCP_SKB_CB(skb)->seq), "TCP recvmsg seq # bug: copied %X, seq %X, rcvnxt %X, fl %X\n", *seq, TCP_SKB_CB(skb)->seq, tp->rcv_nxt, flags)) break; offset = *seq - TCP_SKB_CB(skb)->seq; if (unlikely(TCP_SKB_CB(skb)->tcp_flags & TCPHDR_SYN)) { pr_err_once("%s: found a SYN, please report !\n", __func__); offset--; } if (offset < skb->len) goto found_ok_skb; if (TCP_SKB_CB(skb)->tcp_flags & TCPHDR_FIN) goto found_fin_ok; WARN(!(flags & MSG_PEEK), "TCP recvmsg seq # bug 2: copied %X, seq %X, rcvnxt %X, fl %X\n", *seq, TCP_SKB_CB(skb)->seq, tp->rcv_nxt, flags); } /* Well, if we have backlog, try to process it now yet. */ if (copied >= target && !READ_ONCE(sk->sk_backlog.tail)) break; if (copied) { if (!timeo || sk->sk_err || sk->sk_state == TCP_CLOSE || (sk->sk_shutdown & RCV_SHUTDOWN) || signal_pending(current)) break; } else { if (sock_flag(sk, SOCK_DONE)) break; if (sk->sk_err) { copied = sock_error(sk); break; } if (sk->sk_shutdown & RCV_SHUTDOWN) break; if (sk->sk_state == TCP_CLOSE) { /* This occurs when user tries to read * from never connected socket. */ copied = -ENOTCONN; break; } if (!timeo) { copied = -EAGAIN; break; } if (signal_pending(current)) { copied = sock_intr_errno(timeo); break; } } if (copied >= target) { /* Do not sleep, just process backlog. */ __sk_flush_backlog(sk); } else { tcp_cleanup_rbuf(sk, copied); err = sk_wait_data(sk, &timeo, last); if (err < 0) { err = copied ? : err; goto out; } } if ((flags & MSG_PEEK) && (peek_seq - copied - urg_hole != tp->copied_seq)) { net_dbg_ratelimited("TCP(%s:%d): Application bug, race in MSG_PEEK\n", current->comm, task_pid_nr(current)); peek_seq = tp->copied_seq; } continue; found_ok_skb: /* Ok so how much can we use? */ used = skb->len - offset; if (len < used) used = len; /* Do we have urgent data here? */ if (unlikely(tp->urg_data)) { u32 urg_offset = tp->urg_seq - *seq; if (urg_offset < used) { if (!urg_offset) { if (!sock_flag(sk, SOCK_URGINLINE)) { WRITE_ONCE(*seq, *seq + 1); urg_hole++; offset++; used--; if (!used) goto skip_copy; } } else used = urg_offset; } } if (!(flags & MSG_TRUNC)) { err = skb_copy_datagram_msg(skb, offset, msg, used); if (err) { /* Exception. Bailout! */ if (!copied) copied = -EFAULT; break; } } WRITE_ONCE(*seq, *seq + used); copied += used; len -= used; tcp_rcv_space_adjust(sk); skip_copy: if (unlikely(tp->urg_data) && after(tp->copied_seq, tp->urg_seq)) { WRITE_ONCE(tp->urg_data, 0); tcp_fast_path_check(sk); } if (TCP_SKB_CB(skb)->has_rxtstamp) { tcp_update_recv_tstamps(skb, tss); *cmsg_flags |= TCP_CMSG_TS; } if (used + offset < skb->len) continue; if (TCP_SKB_CB(skb)->tcp_flags & TCPHDR_FIN) goto found_fin_ok; if (!(flags & MSG_PEEK)) tcp_eat_recv_skb(sk, skb); continue; found_fin_ok: /* Process the FIN. */ WRITE_ONCE(*seq, *seq + 1); if (!(flags & MSG_PEEK)) tcp_eat_recv_skb(sk, skb); break; } while (len > 0); /* According to UNIX98, msg_name/msg_namelen are ignored * on connected socket. I was just happy when found this 8) --ANK */ /* Clean up data we have read: This will do ACK frames. */ tcp_cleanup_rbuf(sk, copied); return copied; out: return err; recv_urg: err = tcp_recv_urg(sk, msg, len, flags); goto out; recv_sndq: err = tcp_peek_sndq(sk, msg, len); goto out; } static inline int skb_copy_datagram_msg(const struct sk_buff *from, int offset, struct msghdr *msg, int size) { return skb_copy_datagram_iter(from, offset, &msg->msg_iter, size); } /** * skb_copy_datagram_iter - Copy a datagram to an iovec iterator. * @skb: buffer to copy * @offset: offset in the buffer to start copying from * @to: iovec iterator to copy to * @len: amount of data to copy from buffer to iovec */ int skb_copy_datagram_iter(const struct sk_buff *skb, int offset, struct iov_iter *to, int len) { trace_skb_copy_datagram_iovec(skb, len); return __skb_datagram_iter(skb, offset, to, len, false, simple_copy_to_iter, NULL); } EXPORT_SYMBOL(skb_copy_datagram_iter);