From 8d55710a8c439c45737f30ccf626d04241b95303 Mon Sep 17 00:00:00 2001 From: Mika Westerberg Date: Tue, 15 Sep 2026 18:55:27 +0300 Subject: [PATCH] thunderbolt: stream: Improve CLOSE handling on write side If there is only a writer in the stream (i.e the other side just reads and this side just writes) then if the other side is closed the CLOSE packet does is never picked up from the writers RX ring because we do not complete the RX poll. Currently we do that only when something is actually reading the stream. To make this work on write-only stream, complete the RX poll also during ->write_iter(), and then make the stream return -EPIPE to the writer. While there make sdev->closed to use READ/WRITE_ONCE() analogous to other flags. Signed-off-by: Mika Westerberg --- drivers/thunderbolt/stream.c | 150 ++++++++++++++++++++++++++--------- 1 file changed, 112 insertions(+), 38 deletions(-) diff --git a/drivers/thunderbolt/stream.c b/drivers/thunderbolt/stream.c index f5b67c09ed8d..8a1e06148a14 100644 --- a/drivers/thunderbolt/stream.c +++ b/drivers/thunderbolt/stream.c @@ -133,13 +133,18 @@ struct tbstream_ring { * @busy_poll: Instead of interrupts, busy poll the rings * @rx_pending: Receive ring has completions that need to be advanced * @users: Number of times @cdev has been opened - * @closed: CLOSE packet was received + * @close_received: CLOSE packet appeared in the RX ring + * @closed: CLOSE packet was handled in the read side * @removed: Userspace removed the ConfigFS group underneath. * @wait: Waitqueue for open, read and write * @lock: Lock protecting this structure * @tx_ring: Transmit ring * @rx_ring: Receive ring * @list: Stream devices are linked through this + * + * @close_received is used on write side to notify the writer that the + * other side sent CLOSE. @closed on the other hand is used on the read + * side to make read(2) return EOF to the caller. */ struct tbstream_dev { struct config_group group; @@ -154,6 +159,7 @@ struct tbstream_dev { bool busy_poll; bool rx_pending; int users; + bool close_received; bool closed; bool removed; wait_queue_head_t wait; @@ -348,9 +354,19 @@ static inline bool tbstream_dev_removed(const struct tbstream_dev *sdev) return sdev->removed; } +static inline bool tbstream_dev_close_received(const struct tbstream_dev *sdev) +{ + return READ_ONCE(sdev->close_received); +} + static inline bool tbstream_dev_closed(const struct tbstream_dev *sdev) { - return sdev->closed; + return READ_ONCE(sdev->closed); +} + +static inline bool tbstream_dev_rx_pending(const struct tbstream_dev *sdev) +{ + return READ_ONCE(sdev->rx_pending); } static void @@ -366,12 +382,19 @@ tbstream_dev_rx_callback(struct tb_ring *ring, struct ring_frame *frame, sf->completed = true; sdev->rx_ring.prod++; - if (sf->frame.flags & RING_DESC_CRC_ERROR) + if (sf->frame.flags & RING_DESC_CRC_ERROR) { pr_warn("RX CRC error\n"); - else if (sf->frame.flags & RING_DESC_BUFFER_OVERRUN) + } else if (sf->frame.flags & RING_DESC_BUFFER_OVERRUN) { pr_warn("RX buffer overrun\n"); - else - wake_up_interruptible_poll(&sdev->wait, EPOLLIN | EPOLLRDNORM); + } else { + __poll_t mask = EPOLLIN | EPOLLRDNORM; + + if (sf->frame.eof == TBSTREAM_CLOSE) { + WRITE_ONCE(sdev->close_received, true); + mask |= EPOLLHUP; + } + wake_up_interruptible_poll(&sdev->wait, mask); + } } static struct tbstream_frame * @@ -553,23 +576,24 @@ tbstream_dev_send_data(struct tbstream_dev *sdev, struct iov_iter *from, static int tbstream_dev_send_close(struct tbstream_dev *sdev) { struct tbstream_frame *sf; + ktime_t timeout; - if (sdev->busy_poll) { + /* + * Wait for the ring to have available slots before we send the + * CLOSE packet. + */ + timeout = ktime_add_ms(ktime_get(), 500); + do { + if (tbstream_ring_available(&sdev->tx_ring)) + break; /* - * When busy polling it's the write(2) path that - * advances the completions so it is possible that the - * ring is full at this point. Advance the ring here so - * that there is room for the CLOSE packet to be sent. + * For busy polling we need to advance the ring here as + * well to make the slots available. */ - ktime_t timeout = ktime_add_ms(ktime_get(), 500); - - do { - if (tbstream_ring_available(&sdev->tx_ring)) - break; + if (sdev->busy_poll) tbstream_ring_poll(&sdev->tx_ring); - fsleep(15); - } while (ktime_before(ktime_get(), timeout)); - } + fsleep(15); + } while (ktime_before(ktime_get(), timeout)); sf = tbstream_dev_alloc_tx(sdev, TBSTREAM_CLOSE, NULL, SZ_256); if (IS_ERR(sf)) @@ -745,6 +769,13 @@ static int tbstream_dev_busy_poll_wait(struct tbstream_dev *sdev, return -ERESTARTSYS; if (tb_ring_poll_pending(ring->ring)) return 0; + /* + * For TX ring we need to check the RX side too because + * it might have received CLOSE packet. + */ + if (ring == &sdev->tx_ring && + tb_ring_poll_pending(sdev->rx_ring.ring)) + return 0; if (tbstream_dev_valid(sdev) != 0 || tbstream_dev_closed(sdev) || tbstream_dev_removed(sdev)) return 0; @@ -752,6 +783,22 @@ static int tbstream_dev_busy_poll_wait(struct tbstream_dev *sdev, } } +static bool +tbstream_dev_has_event(struct tbstream_dev *sdev, struct tbstream_ring *ring) +{ + if (tbstream_dev_valid(sdev) != 0) + return true; + if (tbstream_dev_closed(sdev)) + return true; + if (tbstream_dev_removed(sdev)) + return true; + if (tbstream_dev_rx_pending(sdev)) + return true; + if (tbstream_ring_available(ring)) + return true; + return tb_ring_poll_pending(sdev->rx_ring.ring); +} + static ssize_t tbstream_dev_fops_read_iter(struct kiocb *kiocb, struct iov_iter *to) { @@ -800,12 +847,7 @@ tbstream_dev_fops_read_iter(struct kiocb *kiocb, struct iov_iter *to) return ret; } else { ret = wait_event_interruptible(sdev->wait, - READ_ONCE(sdev->rx_pending) || - tb_ring_poll_pending(sdev->rx_ring.ring) || - tbstream_ring_available(&sdev->rx_ring) || - tbstream_dev_valid(sdev) != 0 || - tbstream_dev_closed(sdev) || - tbstream_dev_removed(sdev)); + tbstream_dev_has_event(sdev, &sdev->rx_ring)); if (ret) return ret; } @@ -832,7 +874,7 @@ tbstream_dev_fops_read_iter(struct kiocb *kiocb, struct iov_iter *to) if (sf->frame.eof == TBSTREAM_CLOSE) { if (!nbytes) { tbstream_dev_consume_rx(sdev); - sdev->closed = true; + WRITE_ONCE(sdev->closed, true); } break; } @@ -868,6 +910,25 @@ tbstream_dev_fops_read_iter(struct kiocb *kiocb, struct iov_iter *to) return nbytes; } +static void tbstream_dev_advance_both(struct tbstream_dev *sdev) +{ + /* + * When busy polling, advance TX completions manually. + * + * We also need to advance the RX side in both modes to be able + * to receive CLOSE packet from the other peer if there is no + * reader. + */ + if (sdev->busy_poll) { + tbstream_ring_poll(&sdev->tx_ring); + tbstream_dev_advance_rx(sdev); + } else if (tbstream_dev_rx_pending(sdev) || + tb_ring_poll_pending(sdev->rx_ring.ring)) { + tbstream_dev_advance_rx(sdev); + tbstream_dev_complete_rx(sdev); + } +} + static ssize_t tbstream_dev_fops_write_iter(struct kiocb *kiocb, struct iov_iter *from) { @@ -886,9 +947,8 @@ tbstream_dev_fops_write_iter(struct kiocb *kiocb, struct iov_iter *from) return ret; for (;;) { - /* When busy polling, advance any completions manually */ - if (sdev->busy_poll) - tbstream_ring_poll(&sdev->tx_ring); + /* Advance TX (and RX) completions */ + tbstream_dev_advance_both(sdev); ret = tbstream_dev_valid(sdev); if (ret) { @@ -896,7 +956,12 @@ tbstream_dev_fops_write_iter(struct kiocb *kiocb, struct iov_iter *from) return ret; } - if (tbstream_dev_closed(sdev) || tbstream_dev_removed(sdev)) { + if (tbstream_dev_close_received(sdev)) { + mutex_unlock(&sdev->lock); + return -EPIPE; + } + + if (tbstream_dev_removed(sdev)) { mutex_unlock(&sdev->lock); return -ENXIO; } @@ -915,10 +980,8 @@ tbstream_dev_fops_write_iter(struct kiocb *kiocb, struct iov_iter *from) return ret; } else { ret = wait_event_interruptible(sdev->wait, - tbstream_ring_available(&sdev->tx_ring) || - tbstream_dev_valid(sdev) != 0 || - tbstream_dev_closed(sdev) || - tbstream_dev_removed(sdev)); + tbstream_dev_has_event(sdev, &sdev->tx_ring) || + tbstream_dev_close_received(sdev)); if (ret) return ret; } @@ -978,6 +1041,8 @@ tbstream_dev_fops_poll(struct file *file, struct poll_table_struct *wait) */ tbstream_dev_advance_rx(sdev); + if (tbstream_dev_close_received(sdev)) + mask |= EPOLLHUP; if (tbstream_ring_available(&sdev->tx_ring)) mask |= EPOLLOUT | EPOLLWRNORM; if (tbstream_ring_available(&sdev->rx_ring)) { @@ -1037,6 +1102,8 @@ static int tbstream_dev_fops_open(struct inode *inode, struct file *file) sdev->users--; goto err_unlock; } + + sdev->close_received = false; sdev->closed = false; } @@ -1058,11 +1125,18 @@ static int tbstream_dev_fops_release(struct inode *inode, struct file *file) mutex_lock(&sdev->lock); if (--sdev->users == 0) { /* - * Send CLOSE tunneled packet to notify the other end - * that we are closing the file. We do this twice if the - * first one fails. + * Advance now in case there is CLOSE waiting in the RX + * ring. */ - tbstream_dev_send_close(sdev); + tbstream_dev_advance_both(sdev); + /* + * Send CLOSE tunneled packet to notify the other end + * that we are closing the file, if it is not closed + * already. + */ + if (!tbstream_dev_close_received(sdev) && + tbstream_dev_valid(sdev) == 0) + tbstream_dev_send_close(sdev); tbstream_dev_stop(sdev); } mutex_unlock(&sdev->lock);