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);