引言一次异常引发的思考在分布式系统中使用 Redis 的BRPOP命令实现轻量级消息队列是一种常见做法。然而当业务日志中出现JedisDataException: ERR list value is too large时我们往往只关注数据大小本身却忽略了消费者的实现细节。在一次排查中我们发现消费者线程的堆栈停在SocketInputStream.socketRead0上状态为RUNNABLE。这引发了三个疑问socketRead0明明在“等待”为何 JVM 认为它是RUNNABLEsetSoTimeout(0)真的能让连接“永远”阻塞吗底层如何保证不超时如果网络中断或服务端主从切换线程会被唤醒吗谁来唤醒它本文将从 Jedis 客户端源码出发一路穿越 JNI、系统调用、TCP 协议栈直至内核等待队列深入剖析BRPOP阻塞的完整脉络。一、Jedis 层的“无限等待”魔法1.1 brpop 的实现Jedis 的brpop方法最终调用javapublic ListString brpop(final String... args) { client.brpop(args); client.setTimeoutInfinite(); // ① 设置无限超时 try { return client.getMultiBulkReply(); } finally { client.rollbackTimeout(); // ② 恢复原超时 } }① 处setTimeoutInfinite()将底层 Socket 的soTimeout设置为0infiniteSoTimeout 0表示读取操作永不超时。② 在命令执行后恢复原值。1.2setSoTimeout(0)的真相soTimeout是 Java Socket 的读取超时参数当值为0时socketRead0本地方法中的timeout参数为 0表示无超时阻塞。但这仅仅是“读取操作”的阻塞并不保证 TCP 连接本身永久存活。真正维持连接的是操作系统的 TCP Keep-Alive 和 Redis 服务端的timeout配置默认 0 表示不主动断开。1.3 线程状态之谜当socketRead0阻塞时jstack 显示线程状态为RUNNABLE。这是因为 JVM 无法区分本地方法内的 CPU 计算和 I/O 等待只要线程未因synchronized或wait()而阻塞JVM 就统一标记为RUNNABLE。实际上在操作系统层面该线程处于TASK_INTERRUPTIBLE睡眠状态。二、从 JNI 到系统调用进入内核的桥梁2.1 socketRead0 的 JNI 实现SocketInputStream.socketRead0是 native 方法其 C 实现如下简化cJNIEXPORT jint JNICALL Java_java_net_SocketInputStream_socketRead0(...) { // ... 获取文件描述符 fd if (timeout) { nread NET_ReadWithTimeout(env, fd, bufP, len, timeout); } else { nread NET_Read(fd, bufP, len); // ① 无超时读取 } // ... }① 处的NET_Read是一个宏最终调用recv(fd, buf, len, 0)——系统调用。2.2 系统调用recv的入口recv系统调用在内核中定义为SYSCALL_DEFINE4(recv, ...)其调用链为textsys_recv └─ __sys_recvfrom └─ sock_recvmsg └─ sock_recvmsg_nosec └─ inet_recvmsg (根据协议族) └─ tcp_recvmsg (TCP协议) └─ tcp_recvmsg_locked三、TCP 层的阻塞逻辑tcp_recvmsg_locked3.1 循环等待数据tcp_recvmsg_locked是核心函数它在一个do-while循环中尝试从接收队列sk_receive_queue取数据。关键代码片段已添加中文注释cstatic int tcp_recvmsg_locked(struct sock *sk, ...) { // ... 初始化 timeo sock_rcvtimeo(sk, flags MSG_DONTWAIT); // 获取超时时间 do { // 遍历接收队列查找可读的数据段 skb_queue_walk(sk-sk_receive_queue, skb) { // 若找到数据跳转到 found_ok_skb 拷贝数据 } // 没有数据可读判断是否需要睡眠 if (copied target !sk-sk_backlog.tail) break; if (copied) { // 如果已经读取了一些数据根据条件决定是否继续等待 if (!timeo || sk-sk_err || ...) break; } else { // 完全没有数据检查各种错误条件 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)) { ... break; } } // 准备睡眠 if (copied target) { __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; } } } while (len 0); // ... }当接收队列为空且无错误时程序进入sk_wait_data这正是线程挂起的地方。四、等待队列机制sk_wait_data解剖4.1sk_wait_data实现cint sk_wait_data(struct sock *sk, long *timeo, const struct sk_buff *skb) { DEFINE_WAIT_FUNC(wait, woken_wake_function); // 定义等待队列项 int rc; add_wait_queue(sk_sleep(sk), wait); // 将当前进程加入套接字的等待队列 sk_set_bit(SOCKWQ_ASYNC_WAITDATA, sk); rc sk_wait_event(sk, timeo, skb_peek_tail(sk-sk_receive_queue) ! skb, wait); sk_clear_bit(SOCKWQ_ASYNC_WAITDATA, sk); remove_wait_queue(sk_sleep(sk), wait); // 唤醒后移除 return rc; }sk_wait_event是一个宏展开后如下已注释c({ int __rc, __dis sk-sk_disconnects; release_sock(sk); // 释放锁允许其他上下文操作 __rc (接收队列尾部 ! skb); // 检查条件是否有新数据 if (!__rc) { // 条件不满足需要睡眠 *(timeo) wait_woken(wait, TASK_INTERRUPTIBLE, *(timeo)); } sched_annotate_sleep(); lock_sock(sk); // 重新获取锁 __rc (__dis sk-sk_disconnects) ? (条件) : -EPIPE; __rc; })4.2wait_woken与schedule_timeoutwait_woken将进程状态设为TASK_INTERRUPTIBLE然后调用schedule_timeoutclong wait_woken(struct wait_queue_entry *wq_entry, unsigned mode, long timeout) { set_current_state(mode); // 设置 TASK_INTERRUPTIBLE if (!(wq_entry-flags WQ_FLAG_WOKEN) !kthread_should_stop_or_park()) timeout schedule_timeout(timeout); // ⭐ 真正让出CPU __set_current_state(TASK_RUNNING); // 清除 WQ_FLAG_WOKEN 标志 return timeout; }schedule_timeout最终调用schedule()触发进程切换。对于BRPOPtimeo通常为MAX_SCHEDULE_TIMEOUT无限因此进程会一直睡眠直到被显式唤醒。五、谁唤醒了沉睡的线程—— 数据到达的软中断路径唤醒线程的不是poll而是网卡中断处理。当 TCP 数据包到达时内核经过如下路径text网卡中断 └─ 软中断 (NET_RX_SOFTIRQ) └─ tcp_v4_rcv └─ tcp_v4_do_rcv └─ tcp_rcv_established (或 tcp_data_queue) └─ 将 skb 加入 sk_receive_queue └─ sk_data_ready(sk) // 回调函数 └─ sock_def_readable(sk) └─ wake_up_interruptible(sk_sleep(sk)) └─ __wake_up_common └─ woken_wake_function └─ try_to_wake_up5.1sock_def_readable唤醒等待队列cvoid sock_def_readable(struct sock *sk) { struct socket_wq *wq rcu_dereference(sk-sk_wq); if (wq_has_sleeper(wq)) wake_up_interruptible(wq-wait); // 也可能会触发 sk-sk_data_ready 的其他处理 }wake_up_interruptible遍历等待队列对每个等待项调用回调函数这里是woken_wake_function该函数设置WQ_FLAG_WOKEN标志并调用try_to_wake_up将进程状态设为TASK_RUNNING并将其加入 CPU 运行队列。5.2 唤醒后的恢复被唤醒的进程从schedule_timeout返回接着wait_woken清除标志sk_wait_event重新获取锁并再次检查条件接收队列不为空此时条件为真跳出等待tcp_recvmsg_locked继续执行将数据拷贝到用户空间。六、为什么 JedisCluster 的 BRPOP 存在隐患我们深入理解了阻塞唤醒机制但这建立在连接稳定、服务端正常的前提上。在 Redis 集群主从切换时旧主节点会强制中断阻塞命令返回UNBLOCKED而 JedisCluster 的brpop实现不会自动在新的主节点上重建阻塞上下文。更严重的是setSoTimeout(0)使得客户端读取永远等待如果服务端未主动关闭连接或 TCP Keep-Alive 未及时探测到断连线程可能永远阻塞在sk_wait_data导致队列消息积压。七、总结与建议层级关键函数/机制作用应用层 (Jedis)setTimeoutInfinite()设置soTimeout0使读取永久阻塞JNIsocketRead0调用系统调用recv系统调用__sys_recvfrom→tcp_recvmsg进入 TCP 层接收逻辑内核 TCPtcp_recvmsg_locked循环等待调用sk_wait_data等待队列sk_wait_event→wait_woken→schedule_timeout进程睡眠直到被唤醒唤醒数据包到达 → 软中断 →sock_def_readable→wake_up_interruptible将进程置为可运行建议不要依赖soTimeout0来实现“无限阻塞”这会导致线程在连接异常时永久挂起。使用 Redisson 的RBlockingQueue它正确处理了集群切换和连接重试。若坚持使用 Jedis务必为brpop设置合理的超时如 30 秒并在超时后重新建立连接。监控消费者线程如果长期无消费及时报警。附图阻塞唤醒流程text[Jedis brpop] ↓ [SocketInputStream.socketRead0] // JNI ↓ recv 系统调用 [__sys_recvfrom] ↓ [sock_recvmsg] → [inet_recvmsg] → [tcp_recvmsg] → [tcp_recvmsg_locked] ↓ [sk_wait_data] ↓ [sk_wait_event 宏] ↓ [wait_woken] → [schedule_timeout] → [schedule()] // 进程睡眠 ↑ | | | (数据包到达) | ↓ | [软中断] → [tcp_v4_rcv] → [sk_data_ready] | ↓ ----- [sock_def_readable] → [wake_up_interruptible] → [try_to_wake_up] | 唤醒进程继续执行通过对源码的逐层剖析我们不仅解答了最初的三个问题更揭示了 Redis 阻塞命令背后的高效设计——它并非“忙等待”而是基于事件驱动的等待队列。然而这种优雅的机制在客户端实现不当时反而会成为陷阱。希望本文能帮助读者在享受BRPOP便利的同时规避潜在的“永久阻塞”风险。#源码/* * Receive a datagram from a socket. */ SYSCALL_DEFINE4(recv, int, fd, void __user *, ubuf, size_t, size, unsigned int, flags) { return __sys_recvfrom(fd, ubuf, size, flags, NULL, NULL); } /* * Receive a frame from the socket and optionally record the address of the * sender. We verify the buffers are writable and if needed move the * sender address from kernel to user space. */ int __sys_recvfrom(int fd, void __user *ubuf, size_t size, unsigned int flags, struct sockaddr __user *addr, int __user *addr_len) { struct sockaddr_storage address; struct msghdr msg { /* Save some cycles and dont copy the address if not needed */ .msg_name addr ? (struct sockaddr *)address : NULL, }; struct socket *sock; int err, err2; int fput_needed; err import_ubuf(ITER_DEST, ubuf, size, msg.msg_iter); if (unlikely(err)) return err; sock sockfd_lookup_light(fd, err, fput_needed); if (!sock) goto out; if (sock-file-f_flags O_NONBLOCK) flags | MSG_DONTWAIT; err sock_recvmsg(sock, msg, flags); if (err 0 addr ! NULL) { err2 move_addr_to_user(address, msg.msg_namelen, addr, addr_len); if (err2 0) err err2; } fput_light(sock-file, fput_needed); out: return err; } /** * 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); /** * 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 * shouldnt 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; } /** * sk_wait_data - wait for data to arrive at sk_receive_queue * sk: sock to wait on * timeo: for how long * skb: last skb seen on sk_receive_queue * * Now socket state including sk-sk_err is changed only under lock, * hence we may omit checks after joining wait queue. * We check receive queue before schedule() only as optimization; * it is very likely that release_sock() added new data. */ int sk_wait_data(struct sock *sk, long *timeo, const struct sk_buff *skb) { DEFINE_WAIT_FUNC(wait, woken_wake_function); int rc; add_wait_queue(sk_sleep(sk), wait); sk_set_bit(SOCKWQ_ASYNC_WAITDATA, sk); rc sk_wait_event(sk, timeo, skb_peek_tail(sk-sk_receive_queue) ! skb, wait); sk_clear_bit(SOCKWQ_ASYNC_WAITDATA, sk); remove_wait_queue(sk_sleep(sk), wait); return rc; } EXPORT_SYMBOL(sk_wait_data); #define sk_wait_event(__sk,__timeo,__condition,__wait) ({ int __rc, __dis __sk-sk_disconnects; release_sock(__sk); __rc __condition; if (!__rc) { *(__timeo) wait_woken(__wait, TASK_INTERRUPTIBLE, *(__timeo)); } sched_annotate_sleep(); lock_sock(__sk); __rc __dis __sk-sk_disconnects ? __condition : -EPIPE; __rc; }) Expands to: ({ int __rc, __dis sk-sk_disconnects; release_sock(sk); __rc skb_peek_tail(sk-sk_receive_queue) ! skb; if (!__rc) { *(timeo) wait_woken(wait, TASK_INTERRUPTIBLE, *(timeo)); } sched_annotate_sleep(); lock_sock(sk); __rc __dis sk-sk_disconnects ? skb_peek_tail(sk-sk_receive_queue) ! skb : -32; __rc; }) /* * DEFINE_WAIT_FUNC(wait, woken_wake_func); * * add_wait_queue(wq_head, wait); * for (;;) { * if (condition) * break; * * // in wait_woken() // in woken_wake_function() * * p-state mode; wq_entry-flags | WQ_FLAG_WOKEN; * smp_mb(); // A try_to_wake_up(): * if (!(wq_entry-flags WQ_FLAG_WOKEN)) full barrier * schedule() if (p-state mode) * p-state TASK_RUNNING; p-state TASK_RUNNING; * wq_entry-flags ~WQ_FLAG_WOKEN; ~~~~~~~~~~~~~~~~~~ * smp_mb(); // B condition true; * } smp_mb(); // C * remove_wait_queue(wq_head, wait); wq_entry-flags | WQ_FLAG_WOKEN; */ long wait_woken(struct wait_queue_entry *wq_entry, unsigned mode, long timeout) { /* * The below executes an smp_mb(), which matches with the full barrier * executed by the try_to_wake_up() in woken_wake_function() such that * either we see the store to wq_entry-flags in woken_wake_function() * or woken_wake_function() sees our store to current-state. */ set_current_state(mode); /* A */ if (!(wq_entry-flags WQ_FLAG_WOKEN) !kthread_should_stop_or_park()) timeout schedule_timeout(timeout); __set_current_state(TASK_RUNNING); /* * The below executes an smp_mb(), which matches with the smp_mb() (C) * in woken_wake_function() such that either we see the wait condition * being true or the store to wq_entry-flags in woken_wake_function() * follows ours in the coherence order. */ smp_store_mb(wq_entry-flags, wq_entry-flags ~WQ_FLAG_WOKEN); /* B */ return timeout; } EXPORT_SYMBOL(wait_woken); /** * schedule_timeout - sleep until timeout * timeout: timeout value in jiffies * * Make the current task sleep until timeout jiffies have elapsed. * The function behavior depends on the current task state * (see also set_current_state() description): * * %TASK_RUNNING - the scheduler is called, but the task does not sleep * at all. That happens because sched_submit_work() does nothing for * tasks in %TASK_RUNNING state. * * %TASK_UNINTERRUPTIBLE - at least timeout jiffies are guaranteed to * pass before the routine returns unless the current task is explicitly * woken up, (e.g. by wake_up_process()). * * %TASK_INTERRUPTIBLE - the routine may return early if a signal is * delivered to the current task or the current task is explicitly woken * up. * * The current task state is guaranteed to be %TASK_RUNNING when this * routine returns. * * Specifying a timeout value of %MAX_SCHEDULE_TIMEOUT will schedule * the CPU away without a bound on the timeout. In this case the return * value will be %MAX_SCHEDULE_TIMEOUT. * * Returns 0 when the timer has expired otherwise the remaining time in * jiffies will be returned. In all cases the return value is guaranteed * to be non-negative. */ signed long __sched schedule_timeout(signed long timeout) { struct process_timer timer; unsigned long expire; switch (timeout) { case MAX_SCHEDULE_TIMEOUT: /* * These two special cases are useful to be comfortable * in the caller. Nothing more. We could take * MAX_SCHEDULE_TIMEOUT from one of the negative value * but I d like to return a valid offset (0) to allow * the caller to do everything it want with the retval. */ schedule(); goto out; default: /* * Another bit of PARANOID. Note that the retval will be * 0 since no piece of kernel is supposed to do a check * for a negative retval of schedule_timeout() (since it * should never happens anyway). You just have the printk() * that will tell you if something is gone wrong and where. */ if (timeout 0) { printk(KERN_ERR schedule_timeout: wrong timeout value %lx\n, timeout); dump_stack(); __set_current_state(TASK_RUNNING); goto out; } } expire timeout jiffies; timer.task current; timer_setup_on_stack(timer.timer, process_timeout, 0); __mod_timer(timer.timer, expire, MOD_TIMER_NOTPENDING); schedule(); del_timer_sync(timer.timer); /* Remove the timer from the object tracker */ destroy_timer_on_stack(timer.timer); timeout expire - jiffies; out: return timeout 0 ? 0 : timeout; } EXPORT_SYMBOL(schedule_timeout);
引言一次异常引发的思考在分布式系统中使用 Redis 的BRPOP命令实现轻量级消息队列是一种常见做法。然而当业务日志中出现JedisDataException: ERR list value is too large时我们往往只关注数据大小本身却忽略了消费者的实现细节。在一次排查中我们发现消费者线程的堆栈停在SocketInputStream.socketRead0上状态为RUNNABLE。这引发了三个疑问socketRead0明明在“等待”为何 JVM 认为它是RUNNABLEsetSoTimeout(0)真的能让连接“永远”阻塞吗底层如何保证不超时如果网络中断或服务端主从切换线程会被唤醒吗谁来唤醒它本文将从 Jedis 客户端源码出发一路穿越 JNI、系统调用、TCP 协议栈直至内核等待队列深入剖析BRPOP阻塞的完整脉络。一、Jedis 层的“无限等待”魔法1.1 brpop 的实现Jedis 的brpop方法最终调用javapublic ListString brpop(final String... args) { client.brpop(args); client.setTimeoutInfinite(); // ① 设置无限超时 try { return client.getMultiBulkReply(); } finally { client.rollbackTimeout(); // ② 恢复原超时 } }① 处setTimeoutInfinite()将底层 Socket 的soTimeout设置为0infiniteSoTimeout 0表示读取操作永不超时。② 在命令执行后恢复原值。1.2setSoTimeout(0)的真相soTimeout是 Java Socket 的读取超时参数当值为0时socketRead0本地方法中的timeout参数为 0表示无超时阻塞。但这仅仅是“读取操作”的阻塞并不保证 TCP 连接本身永久存活。真正维持连接的是操作系统的 TCP Keep-Alive 和 Redis 服务端的timeout配置默认 0 表示不主动断开。1.3 线程状态之谜当socketRead0阻塞时jstack 显示线程状态为RUNNABLE。这是因为 JVM 无法区分本地方法内的 CPU 计算和 I/O 等待只要线程未因synchronized或wait()而阻塞JVM 就统一标记为RUNNABLE。实际上在操作系统层面该线程处于TASK_INTERRUPTIBLE睡眠状态。二、从 JNI 到系统调用进入内核的桥梁2.1 socketRead0 的 JNI 实现SocketInputStream.socketRead0是 native 方法其 C 实现如下简化cJNIEXPORT jint JNICALL Java_java_net_SocketInputStream_socketRead0(...) { // ... 获取文件描述符 fd if (timeout) { nread NET_ReadWithTimeout(env, fd, bufP, len, timeout); } else { nread NET_Read(fd, bufP, len); // ① 无超时读取 } // ... }① 处的NET_Read是一个宏最终调用recv(fd, buf, len, 0)——系统调用。2.2 系统调用recv的入口recv系统调用在内核中定义为SYSCALL_DEFINE4(recv, ...)其调用链为textsys_recv └─ __sys_recvfrom └─ sock_recvmsg └─ sock_recvmsg_nosec └─ inet_recvmsg (根据协议族) └─ tcp_recvmsg (TCP协议) └─ tcp_recvmsg_locked三、TCP 层的阻塞逻辑tcp_recvmsg_locked3.1 循环等待数据tcp_recvmsg_locked是核心函数它在一个do-while循环中尝试从接收队列sk_receive_queue取数据。关键代码片段已添加中文注释cstatic int tcp_recvmsg_locked(struct sock *sk, ...) { // ... 初始化 timeo sock_rcvtimeo(sk, flags MSG_DONTWAIT); // 获取超时时间 do { // 遍历接收队列查找可读的数据段 skb_queue_walk(sk-sk_receive_queue, skb) { // 若找到数据跳转到 found_ok_skb 拷贝数据 } // 没有数据可读判断是否需要睡眠 if (copied target !sk-sk_backlog.tail) break; if (copied) { // 如果已经读取了一些数据根据条件决定是否继续等待 if (!timeo || sk-sk_err || ...) break; } else { // 完全没有数据检查各种错误条件 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)) { ... break; } } // 准备睡眠 if (copied target) { __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; } } } while (len 0); // ... }当接收队列为空且无错误时程序进入sk_wait_data这正是线程挂起的地方。四、等待队列机制sk_wait_data解剖4.1sk_wait_data实现cint sk_wait_data(struct sock *sk, long *timeo, const struct sk_buff *skb) { DEFINE_WAIT_FUNC(wait, woken_wake_function); // 定义等待队列项 int rc; add_wait_queue(sk_sleep(sk), wait); // 将当前进程加入套接字的等待队列 sk_set_bit(SOCKWQ_ASYNC_WAITDATA, sk); rc sk_wait_event(sk, timeo, skb_peek_tail(sk-sk_receive_queue) ! skb, wait); sk_clear_bit(SOCKWQ_ASYNC_WAITDATA, sk); remove_wait_queue(sk_sleep(sk), wait); // 唤醒后移除 return rc; }sk_wait_event是一个宏展开后如下已注释c({ int __rc, __dis sk-sk_disconnects; release_sock(sk); // 释放锁允许其他上下文操作 __rc (接收队列尾部 ! skb); // 检查条件是否有新数据 if (!__rc) { // 条件不满足需要睡眠 *(timeo) wait_woken(wait, TASK_INTERRUPTIBLE, *(timeo)); } sched_annotate_sleep(); lock_sock(sk); // 重新获取锁 __rc (__dis sk-sk_disconnects) ? (条件) : -EPIPE; __rc; })4.2wait_woken与schedule_timeoutwait_woken将进程状态设为TASK_INTERRUPTIBLE然后调用schedule_timeoutclong wait_woken(struct wait_queue_entry *wq_entry, unsigned mode, long timeout) { set_current_state(mode); // 设置 TASK_INTERRUPTIBLE if (!(wq_entry-flags WQ_FLAG_WOKEN) !kthread_should_stop_or_park()) timeout schedule_timeout(timeout); // ⭐ 真正让出CPU __set_current_state(TASK_RUNNING); // 清除 WQ_FLAG_WOKEN 标志 return timeout; }schedule_timeout最终调用schedule()触发进程切换。对于BRPOPtimeo通常为MAX_SCHEDULE_TIMEOUT无限因此进程会一直睡眠直到被显式唤醒。五、谁唤醒了沉睡的线程—— 数据到达的软中断路径唤醒线程的不是poll而是网卡中断处理。当 TCP 数据包到达时内核经过如下路径text网卡中断 └─ 软中断 (NET_RX_SOFTIRQ) └─ tcp_v4_rcv └─ tcp_v4_do_rcv └─ tcp_rcv_established (或 tcp_data_queue) └─ 将 skb 加入 sk_receive_queue └─ sk_data_ready(sk) // 回调函数 └─ sock_def_readable(sk) └─ wake_up_interruptible(sk_sleep(sk)) └─ __wake_up_common └─ woken_wake_function └─ try_to_wake_up5.1sock_def_readable唤醒等待队列cvoid sock_def_readable(struct sock *sk) { struct socket_wq *wq rcu_dereference(sk-sk_wq); if (wq_has_sleeper(wq)) wake_up_interruptible(wq-wait); // 也可能会触发 sk-sk_data_ready 的其他处理 }wake_up_interruptible遍历等待队列对每个等待项调用回调函数这里是woken_wake_function该函数设置WQ_FLAG_WOKEN标志并调用try_to_wake_up将进程状态设为TASK_RUNNING并将其加入 CPU 运行队列。5.2 唤醒后的恢复被唤醒的进程从schedule_timeout返回接着wait_woken清除标志sk_wait_event重新获取锁并再次检查条件接收队列不为空此时条件为真跳出等待tcp_recvmsg_locked继续执行将数据拷贝到用户空间。六、为什么 JedisCluster 的 BRPOP 存在隐患我们深入理解了阻塞唤醒机制但这建立在连接稳定、服务端正常的前提上。在 Redis 集群主从切换时旧主节点会强制中断阻塞命令返回UNBLOCKED而 JedisCluster 的brpop实现不会自动在新的主节点上重建阻塞上下文。更严重的是setSoTimeout(0)使得客户端读取永远等待如果服务端未主动关闭连接或 TCP Keep-Alive 未及时探测到断连线程可能永远阻塞在sk_wait_data导致队列消息积压。七、总结与建议层级关键函数/机制作用应用层 (Jedis)setTimeoutInfinite()设置soTimeout0使读取永久阻塞JNIsocketRead0调用系统调用recv系统调用__sys_recvfrom→tcp_recvmsg进入 TCP 层接收逻辑内核 TCPtcp_recvmsg_locked循环等待调用sk_wait_data等待队列sk_wait_event→wait_woken→schedule_timeout进程睡眠直到被唤醒唤醒数据包到达 → 软中断 →sock_def_readable→wake_up_interruptible将进程置为可运行建议不要依赖soTimeout0来实现“无限阻塞”这会导致线程在连接异常时永久挂起。使用 Redisson 的RBlockingQueue它正确处理了集群切换和连接重试。若坚持使用 Jedis务必为brpop设置合理的超时如 30 秒并在超时后重新建立连接。监控消费者线程如果长期无消费及时报警。附图阻塞唤醒流程text[Jedis brpop] ↓ [SocketInputStream.socketRead0] // JNI ↓ recv 系统调用 [__sys_recvfrom] ↓ [sock_recvmsg] → [inet_recvmsg] → [tcp_recvmsg] → [tcp_recvmsg_locked] ↓ [sk_wait_data] ↓ [sk_wait_event 宏] ↓ [wait_woken] → [schedule_timeout] → [schedule()] // 进程睡眠 ↑ | | | (数据包到达) | ↓ | [软中断] → [tcp_v4_rcv] → [sk_data_ready] | ↓ ----- [sock_def_readable] → [wake_up_interruptible] → [try_to_wake_up] | 唤醒进程继续执行通过对源码的逐层剖析我们不仅解答了最初的三个问题更揭示了 Redis 阻塞命令背后的高效设计——它并非“忙等待”而是基于事件驱动的等待队列。然而这种优雅的机制在客户端实现不当时反而会成为陷阱。希望本文能帮助读者在享受BRPOP便利的同时规避潜在的“永久阻塞”风险。#源码/* * Receive a datagram from a socket. */ SYSCALL_DEFINE4(recv, int, fd, void __user *, ubuf, size_t, size, unsigned int, flags) { return __sys_recvfrom(fd, ubuf, size, flags, NULL, NULL); } /* * Receive a frame from the socket and optionally record the address of the * sender. We verify the buffers are writable and if needed move the * sender address from kernel to user space. */ int __sys_recvfrom(int fd, void __user *ubuf, size_t size, unsigned int flags, struct sockaddr __user *addr, int __user *addr_len) { struct sockaddr_storage address; struct msghdr msg { /* Save some cycles and dont copy the address if not needed */ .msg_name addr ? (struct sockaddr *)address : NULL, }; struct socket *sock; int err, err2; int fput_needed; err import_ubuf(ITER_DEST, ubuf, size, msg.msg_iter); if (unlikely(err)) return err; sock sockfd_lookup_light(fd, err, fput_needed); if (!sock) goto out; if (sock-file-f_flags O_NONBLOCK) flags | MSG_DONTWAIT; err sock_recvmsg(sock, msg, flags); if (err 0 addr ! NULL) { err2 move_addr_to_user(address, msg.msg_namelen, addr, addr_len); if (err2 0) err err2; } fput_light(sock-file, fput_needed); out: return err; } /** * 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); /** * 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 * shouldnt 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; } /** * sk_wait_data - wait for data to arrive at sk_receive_queue * sk: sock to wait on * timeo: for how long * skb: last skb seen on sk_receive_queue * * Now socket state including sk-sk_err is changed only under lock, * hence we may omit checks after joining wait queue. * We check receive queue before schedule() only as optimization; * it is very likely that release_sock() added new data. */ int sk_wait_data(struct sock *sk, long *timeo, const struct sk_buff *skb) { DEFINE_WAIT_FUNC(wait, woken_wake_function); int rc; add_wait_queue(sk_sleep(sk), wait); sk_set_bit(SOCKWQ_ASYNC_WAITDATA, sk); rc sk_wait_event(sk, timeo, skb_peek_tail(sk-sk_receive_queue) ! skb, wait); sk_clear_bit(SOCKWQ_ASYNC_WAITDATA, sk); remove_wait_queue(sk_sleep(sk), wait); return rc; } EXPORT_SYMBOL(sk_wait_data); #define sk_wait_event(__sk,__timeo,__condition,__wait) ({ int __rc, __dis __sk-sk_disconnects; release_sock(__sk); __rc __condition; if (!__rc) { *(__timeo) wait_woken(__wait, TASK_INTERRUPTIBLE, *(__timeo)); } sched_annotate_sleep(); lock_sock(__sk); __rc __dis __sk-sk_disconnects ? __condition : -EPIPE; __rc; }) Expands to: ({ int __rc, __dis sk-sk_disconnects; release_sock(sk); __rc skb_peek_tail(sk-sk_receive_queue) ! skb; if (!__rc) { *(timeo) wait_woken(wait, TASK_INTERRUPTIBLE, *(timeo)); } sched_annotate_sleep(); lock_sock(sk); __rc __dis sk-sk_disconnects ? skb_peek_tail(sk-sk_receive_queue) ! skb : -32; __rc; }) /* * DEFINE_WAIT_FUNC(wait, woken_wake_func); * * add_wait_queue(wq_head, wait); * for (;;) { * if (condition) * break; * * // in wait_woken() // in woken_wake_function() * * p-state mode; wq_entry-flags | WQ_FLAG_WOKEN; * smp_mb(); // A try_to_wake_up(): * if (!(wq_entry-flags WQ_FLAG_WOKEN)) full barrier * schedule() if (p-state mode) * p-state TASK_RUNNING; p-state TASK_RUNNING; * wq_entry-flags ~WQ_FLAG_WOKEN; ~~~~~~~~~~~~~~~~~~ * smp_mb(); // B condition true; * } smp_mb(); // C * remove_wait_queue(wq_head, wait); wq_entry-flags | WQ_FLAG_WOKEN; */ long wait_woken(struct wait_queue_entry *wq_entry, unsigned mode, long timeout) { /* * The below executes an smp_mb(), which matches with the full barrier * executed by the try_to_wake_up() in woken_wake_function() such that * either we see the store to wq_entry-flags in woken_wake_function() * or woken_wake_function() sees our store to current-state. */ set_current_state(mode); /* A */ if (!(wq_entry-flags WQ_FLAG_WOKEN) !kthread_should_stop_or_park()) timeout schedule_timeout(timeout); __set_current_state(TASK_RUNNING); /* * The below executes an smp_mb(), which matches with the smp_mb() (C) * in woken_wake_function() such that either we see the wait condition * being true or the store to wq_entry-flags in woken_wake_function() * follows ours in the coherence order. */ smp_store_mb(wq_entry-flags, wq_entry-flags ~WQ_FLAG_WOKEN); /* B */ return timeout; } EXPORT_SYMBOL(wait_woken); /** * schedule_timeout - sleep until timeout * timeout: timeout value in jiffies * * Make the current task sleep until timeout jiffies have elapsed. * The function behavior depends on the current task state * (see also set_current_state() description): * * %TASK_RUNNING - the scheduler is called, but the task does not sleep * at all. That happens because sched_submit_work() does nothing for * tasks in %TASK_RUNNING state. * * %TASK_UNINTERRUPTIBLE - at least timeout jiffies are guaranteed to * pass before the routine returns unless the current task is explicitly * woken up, (e.g. by wake_up_process()). * * %TASK_INTERRUPTIBLE - the routine may return early if a signal is * delivered to the current task or the current task is explicitly woken * up. * * The current task state is guaranteed to be %TASK_RUNNING when this * routine returns. * * Specifying a timeout value of %MAX_SCHEDULE_TIMEOUT will schedule * the CPU away without a bound on the timeout. In this case the return * value will be %MAX_SCHEDULE_TIMEOUT. * * Returns 0 when the timer has expired otherwise the remaining time in * jiffies will be returned. In all cases the return value is guaranteed * to be non-negative. */ signed long __sched schedule_timeout(signed long timeout) { struct process_timer timer; unsigned long expire; switch (timeout) { case MAX_SCHEDULE_TIMEOUT: /* * These two special cases are useful to be comfortable * in the caller. Nothing more. We could take * MAX_SCHEDULE_TIMEOUT from one of the negative value * but I d like to return a valid offset (0) to allow * the caller to do everything it want with the retval. */ schedule(); goto out; default: /* * Another bit of PARANOID. Note that the retval will be * 0 since no piece of kernel is supposed to do a check * for a negative retval of schedule_timeout() (since it * should never happens anyway). You just have the printk() * that will tell you if something is gone wrong and where. */ if (timeout 0) { printk(KERN_ERR schedule_timeout: wrong timeout value %lx\n, timeout); dump_stack(); __set_current_state(TASK_RUNNING); goto out; } } expire timeout jiffies; timer.task current; timer_setup_on_stack(timer.timer, process_timeout, 0); __mod_timer(timer.timer, expire, MOD_TIMER_NOTPENDING); schedule(); del_timer_sync(timer.timer); /* Remove the timer from the object tracker */ destroy_timer_on_stack(timer.timer); timeout expire - jiffies; out: return timeout 0 ? 0 : timeout; } EXPORT_SYMBOL(schedule_timeout);
