xref: /linux/drivers/thunderbolt/stream.c (revision 42bcac323cc3b16566fed8c5b250318ad02a0de0)

// SPDX-License-Identifier: GPL-2.0
/*
 * Stream data over Thunderbolt/USB4 cable
 *
 * Copyright (C) 2026, Intel Corporation
 * Authors: Alan Borzeszkowski <alan.borzeszkowski@linux.intel.com>
 *	    Mika Westerberg <mika.westerberg@linux.intel.com>
 */

#define pr_fmt(fmt) "tbstream: " fmt

#include <linux/delay.h>
#include <linux/configfs.h>
#include <linux/file.h>
#include <linux/fs.h>
#include <linux/idr.h>
#include <linux/ktime.h>
#include <linux/miscdevice.h>
#include <linux/module.h>
#include <linux/mutex.h>
#include <linux/poll.h>
#include <linux/sizes.h>
#include <linux/thunderbolt.h>
#include <linux/uaccess.h>
#include <linux/uio.h>
#include <linux/uuid.h>
#include <linux/wait.h>

/*
 * USB4STREAM - Stream data directly over Thunderbolt/USB4 cable
 *
 * HopIDs are configured by the user. In Linux this is done through
 * ConfigFS. Once that is done paths are be established the first time
 * the stream is opened. Typically the read side is opened first to make
 * sure all the data will be received.
 *
 * End-to-end flow control is mandatory on both sides.
 *
 * Data is sent to the other side as tunneled DATA packets. All the data
 * is owned by the user and passed as-is from the writer to the reader.
 *
 * Once the stream device is closed, a CLOSE packet is sent to the peer
 * so it can take the necessary action. On Linux this typically results
 * in EOF being returned to the reader.
 *
 * Tunneled packet types:
 *
 * +-------+---------+------------------+
 * |  PDF  |  Type   | Payload size     |
 * +-------+---------+------------------+
 * |   2   | DATA    | up to 4 KiB      |
 * |   3   | CLOSE   | up to 256 bytes  |
 * +-------+---------+------------------+
 *
 * Each stream can optionally publish configuration values under its own
 * XDomain property directory. The name of the directory is the name of
 * the stream in question and the UUID is up to the stream. For example
 * if the stream exposes video output then the directory name could be
 * "video".
 *
 * Below values are reserved and can be used by the stream:
 *
 * +----------+-----------+-------------------------+
 * |   Key    |   Type    | Contents                |
 * +----------+-----------+-------------------------+
 * | inhopid  | IMMEDIATE | Configured input HopID  |
 * | outhopid | IMMEDIATE | Configured output HopID |
 * +----------+-----------+-------------------------+
 *
 * It is allowed to add more stream specific properties as well if the
 * above are not enough.
 */

#define TBSTREAM_DEV_RING_SIZE		256
#define TBSTREAM_DEV_MIN_RING_SIZE	32
#define TBSTREAM_DEV_MAX_RING_SIZE	4096
#define TBSTREAM_DEV_THROTTLING		8192
#define TBSTREAM_DEV_MAX_THROTTLING	16776960

/**
 * enum tbstream_frame_pdf - PDF numbers for tunneled frames
 * @TBSTREAM_FRAME_START: PDF of the start of the frame
 * @TBSTREAM_DATA: PDF of the DATA frame
 * @TBSTREAM_CLOSE: PDF of the CLOSE frame
 */
enum tbstream_frame_pdf {
	TBSTREAM_FRAME_START = 1,
	TBSTREAM_DATA,
	TBSTREAM_CLOSE,
};

/**
 * struct tbstream_frame - Frame submitted to/from the rings
 * @sdev: Pointer to the stream device
 * @page: Page holding the packet
 * @offset: Offset inside @page if partial read is done
 * @completed: %true if the RX frame is completed
 * @frame: Underlying frame structure
 */
struct tbstream_frame {
	struct tbstream_dev *sdev;
	struct page *page;
	unsigned int offset;
	bool completed;
	struct ring_frame frame;
};

/**
 * struct tbstream_ring - Stream RX/TX ring structure
 * @ring: Pointer to the API ring
 * @prod: Current value of producer
 * @cons: Current value of consumer
 * @frames: Holds the ring frames
 */
struct tbstream_ring {
	struct tb_ring *ring;
	unsigned long prod;
	unsigned long cons;
	struct tbstream_frame *frames;
};

/**
 * struct tbstream_dev - Stream character device
 * @group: ConfigFS group for this device
 * @stream: Pointer to the stream if it is attached (%NULL otherwise)
 * @misc: Character device used for tunneling
 * @kref: Reference count
 * @index: Unique identifier for the character device
 * @in_hopid: In HopID
 * @out_hopid: Out HopID
 * @ring_size: Size of the rings
 * @throttling: Interrupt throttling rate in ns
 * @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
 * @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;
	struct tbstream *stream;
	struct miscdevice misc;
	struct kref kref;
	int index;
	int in_hopid;
	int out_hopid;
	unsigned int ring_size;
	unsigned int throttling;
	bool busy_poll;
	bool rx_pending;
	int users;
	bool close_received;
	bool closed;
	bool removed;
	wait_queue_head_t wait;
	struct mutex lock;
	struct tbstream_ring tx_ring;
	struct tbstream_ring rx_ring;
	struct list_head list;
};

/**
 * struct tbstream_group - Config group for stream
 * @group: ConfigFS group for @stream
 * @stream: Stream the ConfigFS group is attached to. %NULL if there is
 *	    no stream attached.
 * @lock: Lock protecting this structure
 * @dev_list: List of stream devices
 *
 * This is the ConfigFS directory for one connection to another host.
 * There can be several &struct stream_dev linked through @dev_list of
 * this structure. Reference count managed through @group.
 */
struct tbstream_group {
	struct config_group group;
	struct tbstream *stream;
	struct mutex lock;
	struct list_head dev_list;
};

/**
 * struct tbstream - Stream service private data
 * @kref: Reference count
 * @svc: Pointer to the service device
 * @list: Streams are linked through this in @stream_list
 *
 * This represents the actual physical connection between two hosts.
 */
struct tbstream {
	struct kref kref;
	struct tb_service *svc;
	struct list_head list;
};

static DEFINE_IDA(tbstream_indices);

/* Protects tbstream_list */
static DEFINE_MUTEX(tbstream_lock);
static LIST_HEAD(tbstream_list);

/* Serializes tbstream_get()/put() */
static DEFINE_MUTEX(tbstream_kref_lock);

/* Serializes tbstream_dev_get()/put() */
static DEFINE_MUTEX(tbstream_dev_kref_lock);

/* Stream property directory UUID: 3a1cb984-c4d9-4469-a277-ce2fdfd11f0d */
static const uuid_t tbstream_dir_uuid =
	UUID_INIT(0x3a1cb984, 0xc4d9, 0x4469,
		  0xa2, 0x77, 0xce, 0x2f, 0xdf, 0xd1, 0x1f, 0x0d);

static struct tb_property_dir *tbstream_dir;

static void tbstream_release(struct kref *kref)
{
	struct tbstream *stream = container_of(kref, typeof(*stream), kref);

	tb_service_put(stream->svc);
	kfree(stream);
}

static void tbstream_put(struct tbstream *stream)
{
	if (stream) {
		guard(mutex)(&tbstream_kref_lock);
		kref_put(&stream->kref, tbstream_release);
	}
}

static struct tbstream *tbstream_get(struct tbstream *stream)
{
	if (stream) {
		guard(mutex)(&tbstream_kref_lock);
		kref_get(&stream->kref);
	}
	return stream;
}

static inline bool tbstream_valid(const struct tbstream *stream)
{
	if (stream)
		return !tb_service_parent(stream->svc)->is_unplugged;
	return false;
}

static void tbstream_ring_free(struct tbstream_ring *ring)
{
	struct device *dma_dev = tb_ring_dma_device(ring->ring);
	enum dma_data_direction dir;
	int i;

	if (ring->ring->is_tx)
		dir = DMA_TO_DEVICE;
	else
		dir = DMA_FROM_DEVICE;

	for (i = 0; i < tb_ring_size(ring->ring); i++) {
		struct tbstream_frame *sf = &ring->frames[i];

		if (sf->frame.buffer_phy)
			dma_unmap_page(dma_dev, sf->frame.buffer_phy,
				       TB_MAX_FRAME_SIZE, dir);
		sf->frame.buffer_phy = 0;
		if (sf->page)
			__free_page(sf->page);
		sf->page = NULL;
	}

	ring->prod = 0;
	ring->cons = 0;
	kfree(ring->frames);
}

static inline bool tbstream_ring_available(const struct tbstream_ring *ring)
{
	return ring->prod > ring->cons;
}

static void tbstream_ring_poll(struct tbstream_ring *ring)
{
	struct ring_frame *frame;

	while ((frame = tb_ring_poll(ring->ring)))
		frame->callback(ring->ring, frame, false);
}

static inline struct tb_xdomain *tbstream_dev_xdomain(struct tbstream_dev *sdev)
{
	if (sdev->stream)
		return tb_service_parent(sdev->stream->svc);
	return NULL;
}

static void tbstream_dev_release(struct kref *kref)
{
	struct tbstream_dev *sdev = container_of(kref, struct tbstream_dev, kref);

	if (sdev->stream) {
		struct tb_xdomain *xd = tbstream_dev_xdomain(sdev);

		if (sdev->out_hopid > 0)
			tb_xdomain_release_out_hopid(xd, sdev->out_hopid);
		if (sdev->in_hopid > 0)
			tb_xdomain_release_in_hopid(xd, sdev->in_hopid);

		tbstream_put(sdev->stream);
	}
	ida_free(&tbstream_indices, sdev->index);
	kfree(sdev->misc.name);
	kfree(sdev);
}

static inline void tbstream_dev_put(struct tbstream_dev *sdev)
{
	guard(mutex)(&tbstream_dev_kref_lock);
	kref_put(&sdev->kref, tbstream_dev_release);
}

static inline struct tbstream_dev *tbstream_dev_get(struct tbstream_dev *sdev)
{
	guard(mutex)(&tbstream_dev_kref_lock);
	kref_get(&sdev->kref);
	return sdev;
}

static inline struct tbstream_dev *to_tbstream_dev(struct miscdevice *misc)
{
	return container_of(misc, struct tbstream_dev, misc);
}

static inline int tbstream_dev_valid(const struct tbstream_dev *sdev)
{
	const struct tbstream *stream = sdev->stream;

	if (!tbstream_valid(stream))
		return -ENXIO;
	if (sdev->in_hopid <= 0 || sdev->out_hopid <= 0)
		return -EINVAL;
	return 0;
}

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 READ_ONCE(sdev->closed);
}

static inline bool tbstream_dev_rx_pending(const struct tbstream_dev *sdev)
{
	return READ_ONCE(sdev->rx_pending);
}

static void
tbstream_dev_rx_callback(struct tb_ring *ring, struct ring_frame *frame,
			 bool canceled)
{
	struct tbstream_frame *sf = container_of(frame, typeof(*sf), frame);
	struct tbstream_dev *sdev = sf->sdev;
	__poll_t mask;

	if (canceled)
		return;

	sf->completed = true;
	sdev->rx_ring.prod++;

	mask = EPOLLIN | EPOLLRDNORM;
	/*
	 * The CLOSE packet does not have a payload so the flags do not
	 * matter. The read_iter() deals with the flags.
	 */
	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 *
tbstream_dev_completed_rx(struct tbstream_dev *sdev)
{
	struct device *dma_dev = tb_ring_dma_device(sdev->rx_ring.ring);
	struct tbstream_frame *sf;
	int index;

	index = sdev->rx_ring.cons % tb_ring_size(sdev->rx_ring.ring);
	sf = &sdev->rx_ring.frames[index];
	if (!sf->completed)
		return NULL;

	dma_sync_single_for_cpu(dma_dev, sf->frame.buffer_phy,
				tb_ring_frame_size(&sf->frame),
				DMA_FROM_DEVICE);
	return sf;
}

static int tbstream_dev_consume_rx(struct tbstream_dev *sdev)
{
	struct device *dma_dev = tb_ring_dma_device(sdev->rx_ring.ring);
	struct tbstream_frame *sf;
	int index;

	index = sdev->rx_ring.cons % tb_ring_size(sdev->rx_ring.ring);
	sdev->rx_ring.cons++;

	sf = &sdev->rx_ring.frames[index];
	sf->completed = false;
	sf->offset = 0;
	sf->frame.size = 0;

	dma_sync_single_for_device(dma_dev, sf->frame.buffer_phy,
				   tb_ring_frame_size(&sf->frame),
				   DMA_FROM_DEVICE);

	return tb_ring_rx(sdev->rx_ring.ring, &sf->frame);
}

static int tbstream_dev_alloc_rx_buffers(struct tbstream_dev *sdev)
{
	size_t ring_size = tb_ring_size(sdev->rx_ring.ring);
	int i;

	sdev->rx_ring.frames = kzalloc_objs(struct tbstream_frame, ring_size);
	if (!sdev->rx_ring.frames)
		return -ENOMEM;

	for (i = 0; i < ring_size; i++) {
		struct device *dma_dev = tb_ring_dma_device(sdev->rx_ring.ring);
		struct tbstream_frame *sf = &sdev->rx_ring.frames[i];
		dma_addr_t dma_addr;

		sf->page = alloc_page(GFP_KERNEL);
		if (!sf->page)
			return -ENOMEM;

		dma_addr = dma_map_page(dma_dev, sf->page, 0, TB_MAX_FRAME_SIZE,
					DMA_FROM_DEVICE);
		if (dma_mapping_error(dma_dev, dma_addr)) {
			__free_page(sf->page);
			sf->page = NULL;
			return -ENOMEM;
		}

		sf->sdev = sdev;
		sf->frame.callback = tbstream_dev_rx_callback;
		sf->frame.buffer_phy = dma_addr;

		tb_ring_rx(sdev->rx_ring.ring, &sf->frame);
	}

	sdev->rx_ring.cons = 0;
	sdev->rx_ring.prod = 0;
	return 0;
}

static void
tbstream_dev_tx_callback(struct tb_ring *ring, struct ring_frame *frame,
			 bool canceled)
{
	struct tbstream_frame *sf = container_of(frame, typeof(*sf), frame);
	struct tbstream_dev *sdev = sf->sdev;

	if (canceled)
		return;

	sdev->tx_ring.prod++;
	if (sf->frame.eof == TBSTREAM_DATA)
		wake_up_interruptible_poll(&sdev->wait, EPOLLOUT | EPOLLWRNORM);
}

static int tbstream_dev_alloc_tx_buffers(struct tbstream_dev *sdev)
{
	struct device *dma_dev = tb_ring_dma_device(sdev->tx_ring.ring);
	size_t ring_size = tb_ring_size(sdev->tx_ring.ring);
	int i;

	sdev->tx_ring.frames = kzalloc_objs(struct tbstream_frame, ring_size);
	if (!sdev->tx_ring.frames)
		return -ENOMEM;

	for (i = 0; i < ring_size; i++) {
		struct tbstream_frame *sf = &sdev->tx_ring.frames[i];
		dma_addr_t dma_addr;

		sf->page = alloc_page(GFP_KERNEL);
		if (!sf->page)
			return -ENOMEM;

		dma_addr = dma_map_page(dma_dev, sf->page, 0, TB_MAX_FRAME_SIZE,
					DMA_TO_DEVICE);
		if (dma_mapping_error(dma_dev, dma_addr)) {
			__free_page(sf->page);
			sf->page = NULL;
			return -ENOMEM;
		}

		sf->sdev = sdev;
		sf->frame.callback = tbstream_dev_tx_callback;
		sf->frame.buffer_phy = dma_addr;
		sf->frame.sof = TBSTREAM_FRAME_START;
	}

	sdev->tx_ring.cons = 0;
	sdev->tx_ring.prod = ring_size - 1;
	return 0;
}

static struct tbstream_frame *
tbstream_dev_alloc_tx(struct tbstream_dev *sdev, enum tbstream_frame_pdf pdf,
		      struct iov_iter *from, size_t size)
{
	struct device *dma_dev = tb_ring_dma_device(sdev->tx_ring.ring);
	struct tbstream_frame *sf;
	int index;

	if (!tbstream_ring_available(&sdev->tx_ring))
		return ERR_PTR(-ENOBUFS);

	index = sdev->tx_ring.cons % tb_ring_size(sdev->tx_ring.ring);
	sdev->tx_ring.cons++;

	sf = &sdev->tx_ring.frames[index];
	sf->frame.size = size < TB_MAX_FRAME_SIZE ? size : 0;
	sf->frame.eof = pdf;

	dma_sync_single_for_cpu(dma_dev, sf->frame.buffer_phy, size,
				DMA_TO_DEVICE);
	if (pdf == TBSTREAM_DATA) {
		if (copy_page_from_iter(sf->page, 0, size, from) != size) {
			sdev->tx_ring.cons--;
			return ERR_PTR(-EFAULT);
		}
	} else {
		memset(page_address(sf->page), 0, size);
	}
	dma_sync_single_for_device(dma_dev, sf->frame.buffer_phy, size,
				   DMA_TO_DEVICE);
	return sf;
}

static int
tbstream_dev_send_data(struct tbstream_dev *sdev, struct iov_iter *from,
		       size_t size)
{
	struct tbstream_frame *sf;

	sf = tbstream_dev_alloc_tx(sdev, TBSTREAM_DATA, from, size);
	if (IS_ERR(sf))
		return PTR_ERR(sf);
	return tb_ring_tx(sdev->tx_ring.ring, &sf->frame);
}

static int tbstream_dev_send_close(struct tbstream_dev *sdev)
{
	struct tbstream_frame *sf;
	ktime_t timeout;

	/*
	 * 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;
		/*
		 * For busy polling we need to advance the ring here as
		 * well to make the slots available.
		 */
		if (sdev->busy_poll)
			tbstream_ring_poll(&sdev->tx_ring);
		fsleep(15);
	} while (ktime_before(ktime_get(), timeout));

	sf = tbstream_dev_alloc_tx(sdev, TBSTREAM_CLOSE, NULL, SZ_256);
	if (IS_ERR(sf))
		return PTR_ERR(sf);
	return tb_ring_tx(sdev->tx_ring.ring, &sf->frame);
}

static void tbstream_dev_start_poll(void *data)
{
	struct tbstream_dev *sdev = data;

	WRITE_ONCE(sdev->rx_pending, true);
	wake_up_interruptible_poll(&sdev->wait, EPOLLIN | EPOLLRDNORM);
}

/* sdev->lock must be held */
static void tbstream_dev_advance_rx(struct tbstream_dev *sdev)
{
	/*
	 * Clear before running the completions so that an interrupt
	 * that arrives while we are doing that is not missed.
	 */
	WRITE_ONCE(sdev->rx_pending, false);
	tbstream_ring_poll(&sdev->rx_ring);
}

/* sdev->lock must be held */
static void tbstream_dev_complete_rx(struct tbstream_dev *sdev)
{
	if (!sdev->busy_poll)
		tb_ring_poll_complete(sdev->rx_ring.ring);
}

static int tbstream_dev_start(struct tbstream_dev *sdev)
{
	struct tb_xdomain *xd = tbstream_dev_xdomain(sdev);
	unsigned int flags = RING_FLAG_FRAME | RING_FLAG_E2E;
	void (*start_poll)(void *) = NULL;
	u16 sof_mask, eof_mask;
	struct tb_ring *ring;
	int ret, e2e_tx_hop;

	if (sdev->busy_poll)
		flags |= RING_FLAG_NO_INTERRUPT;
	else
		start_poll = tbstream_dev_start_poll;

	ring = tb_ring_alloc_tx(xd->tb->nhi, -1, sdev->ring_size, flags);
	if (!ring)
		return -ENOMEM;
	sdev->tx_ring.ring = ring;

	ret = tbstream_dev_alloc_tx_buffers(sdev);
	if (ret)
		goto err_free_tx;

	e2e_tx_hop = ring->hop;
	sof_mask = BIT(TBSTREAM_FRAME_START);
	eof_mask = BIT(TBSTREAM_DATA) | BIT(TBSTREAM_CLOSE);

	ring = tb_ring_alloc_rx(xd->tb->nhi, -1, sdev->ring_size, flags,
				e2e_tx_hop, sof_mask, eof_mask, start_poll,
				sdev);
	if (!ring) {
		ret = -ENOMEM;
		goto err_free_tx_buffers;
	}
	sdev->rx_ring.ring = ring;

	ret = tb_xdomain_enable_paths(xd, sdev->out_hopid,
				     sdev->tx_ring.ring->hop,
				     sdev->in_hopid,
				     sdev->rx_ring.ring->hop);
	if (ret)
		goto err_free_rx;

	tb_ring_throttling(sdev->tx_ring.ring, sdev->throttling);
	tb_ring_throttling(sdev->rx_ring.ring, sdev->throttling);

	sdev->rx_pending = false;

	tb_ring_start(sdev->tx_ring.ring);
	tb_ring_start(sdev->rx_ring.ring);

	ret = tbstream_dev_alloc_rx_buffers(sdev);
	if (ret)
		goto err_stop;
	return 0;

err_stop:
	tb_ring_stop(sdev->rx_ring.ring);
	tb_ring_stop(sdev->tx_ring.ring);
err_free_rx:
	tb_ring_free(sdev->rx_ring.ring);
err_free_tx_buffers:
	tbstream_ring_free(&sdev->tx_ring);
err_free_tx:
	tb_ring_free(sdev->tx_ring.ring);

	return ret;
}

static bool tbstream_dev_tx_drained(const struct tbstream_dev *sdev)
{
	const struct tbstream_ring *ring = &sdev->tx_ring;

	/*
	 * Everything is completed when number of free TX slots is back
	 * to the maximum.
	 */
	return ring->prod - ring->cons == tb_ring_size(ring->ring) - 1;
}

static void tbstream_dev_stop(struct tbstream_dev *sdev)
{
	struct tb_xdomain *xd;

	if (sdev->busy_poll) {
		/*
		 * When busy polling we must advance the ring ourselves
		 * to push all outstanding frames on the wire.
		 */
		ktime_t timeout = ktime_add_ms(ktime_get(), 500);

		do {
			if (tbstream_dev_tx_drained(sdev))
				break;
			tbstream_ring_poll(&sdev->tx_ring);
			fsleep(15);
		} while (ktime_before(ktime_get(), timeout));
	} else {
		tb_ring_flush(sdev->tx_ring.ring, 500);
	}

	tb_ring_stop(sdev->tx_ring.ring);
	tb_ring_stop(sdev->rx_ring.ring);

	xd = tbstream_dev_xdomain(sdev);
	if (xd) {
		tb_xdomain_disable_paths(xd, sdev->out_hopid,
					 sdev->tx_ring.ring->hop,
					 sdev->in_hopid,
					 sdev->rx_ring.ring->hop);
	}

	tbstream_ring_free(&sdev->rx_ring);
	tb_ring_free(sdev->rx_ring.ring);
	sdev->rx_ring.ring = NULL;
	tbstream_ring_free(&sdev->tx_ring);
	tb_ring_free(sdev->tx_ring.ring);
	sdev->tx_ring.ring = NULL;
}

/* Use only with read_iter/write_iter() to handle nowait */
static int tbstream_dev_lock(struct tbstream_dev *sdev, bool nowait)
{
	if (nowait) {
		if (!mutex_trylock(&sdev->lock))
			return -EAGAIN;
	} else {
		if (mutex_lock_interruptible(&sdev->lock))
			return -ERESTARTSYS;
	}
	return 0;
}

/* Must not be called with @sdev->lock held */
static int tbstream_dev_busy_poll_wait(struct tbstream_dev *sdev,
				       struct tbstream_ring *ring)
{
	for (;;) {
		if (signal_pending(current))
			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;
		cond_resched();
	}
}

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)
{
	struct file *file = kiocb->ki_filp;
	bool nowait = file->f_flags & O_NONBLOCK || kiocb->ki_flags & IOCB_NOWAIT;
	struct tbstream_dev *sdev = to_tbstream_dev(file->private_data);
	size_t nbytes;
	int ret;

	ret = tbstream_dev_valid(sdev);
	if (ret)
		return ret;

	ret = tbstream_dev_lock(sdev, nowait);
	if (ret)
		return ret;

	for (;;) {
		/* Advance RX completions */
		tbstream_dev_advance_rx(sdev);

		ret = tbstream_dev_valid(sdev);
		if (ret) {
			mutex_unlock(&sdev->lock);
			return ret;
		}

		if (tbstream_dev_closed(sdev) || tbstream_dev_removed(sdev)) {
			mutex_unlock(&sdev->lock);
			return 0;
		}

		if (tbstream_ring_available(&sdev->rx_ring))
			break;

		/* Polled all we could. Re-enable the interrupt now. */
		tbstream_dev_complete_rx(sdev);
		mutex_unlock(&sdev->lock);

		if (nowait)
			return -EAGAIN;

		if (sdev->busy_poll) {
			ret = tbstream_dev_busy_poll_wait(sdev, &sdev->rx_ring);
			if (ret)
				return ret;
		} else {
			ret = wait_event_interruptible(sdev->wait,
				tbstream_dev_has_event(sdev, &sdev->rx_ring));
			if (ret)
				return ret;
		}

		ret = tbstream_dev_lock(sdev, nowait);
		if (ret)
			return ret;
	}

	nbytes = 0;
	while (iov_iter_count(to)) {
		struct tbstream_frame *sf;
		size_t size, sf_size;

		sf = tbstream_dev_completed_rx(sdev);
		if (!sf)
			break;
		/*
		 * CLOSE tunneled packet. If userspace already read
		 * something then we stop processing now and return
		 * those bytes. Next time the first frame will be CLOSE
		 * in which case we return EOF to the user.
		 */
		if (sf->frame.eof == TBSTREAM_CLOSE) {
			if (!nbytes) {
				tbstream_dev_consume_rx(sdev);
				WRITE_ONCE(sdev->closed, true);
			}
			break;
		} else if (sf->frame.flags &
			   (RING_DESC_CRC_ERROR | RING_DESC_BUFFER_OVERRUN)) {
			/*
			 * If something was already read return that now
			 * and next read will report the error.
			 */
			if (!nbytes) {
				pr_warn("corrupted frame received, flags %#x\n",
					sf->frame.flags);
				tbstream_dev_consume_rx(sdev);
				ret = -EIO;
			}
			break;
		}

		sf_size = tb_ring_frame_size(&sf->frame);
		size = min(iov_iter_count(to), sf_size);

		if (copy_page_to_iter(sf->page, sf->offset, size, to) != size) {
			ret = -EFAULT;
			break;
		}

		/*
		 * If not all data from the frame is read so leave it in
		 * place and update the offset accordingly so next read
		 * gets the rest.
		 */
		if (size < sf_size) {
			sf->offset += size;
			sf->frame.size = sf_size - size;
		} else {
			ret = tbstream_dev_consume_rx(sdev);
			if (ret)
				break;
		}

		nbytes += size;
	}

	mutex_unlock(&sdev->lock);
	if (ret)
		return ret;
	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)
{
	struct file *file = kiocb->ki_filp;
	bool nowait = file->f_flags & O_NONBLOCK || kiocb->ki_flags & IOCB_NOWAIT;
	struct tbstream_dev *sdev = to_tbstream_dev(file->private_data);
	size_t nbytes;
	int ret;

	ret = tbstream_dev_valid(sdev);
	if (ret)
		return ret;

	ret = tbstream_dev_lock(sdev, nowait);
	if (ret)
		return ret;

	for (;;) {
		/* Advance TX (and RX) completions */
		tbstream_dev_advance_both(sdev);

		ret = tbstream_dev_valid(sdev);
		if (ret) {
			mutex_unlock(&sdev->lock);
			return ret;
		}

		if (tbstream_dev_close_received(sdev)) {
			mutex_unlock(&sdev->lock);
			return -EPIPE;
		}

		if (tbstream_dev_removed(sdev)) {
			mutex_unlock(&sdev->lock);
			return -ENXIO;
		}

		if (tbstream_ring_available(&sdev->tx_ring))
			break;

		mutex_unlock(&sdev->lock);

		if (nowait)
			return -EAGAIN;

		if (sdev->busy_poll) {
			ret = tbstream_dev_busy_poll_wait(sdev, &sdev->tx_ring);
			if (ret)
				return ret;
		} else {
			ret = wait_event_interruptible(sdev->wait,
				tbstream_dev_has_event(sdev, &sdev->tx_ring) ||
				tbstream_dev_close_received(sdev));
			if (ret)
				return ret;
		}

		ret = tbstream_dev_lock(sdev, nowait);
		if (ret)
			return ret;
	}

	nbytes = 0;
	while (iov_iter_count(from)) {
		size_t size;

		size = min(iov_iter_count(from), TB_MAX_FRAME_SIZE);
		ret = tbstream_dev_send_data(sdev, from, size);
		if (ret) {
			/*
			 * If there are no more buffers we are done for
			 * this write.
			 */
			if (ret == -ENOBUFS)
				ret = 0;
			break;
		}

		nbytes += size;
	}

	mutex_unlock(&sdev->lock);
	if (ret)
		return ret;
	return nbytes;
}

static __poll_t
tbstream_dev_fops_poll(struct file *file, struct poll_table_struct *wait)
{
	struct tbstream_dev *sdev = to_tbstream_dev(file->private_data);
	__poll_t mask = 0;

	/*
	 * Without interrupts there is nothing that can wake us up so
	 * return failure instead.
	 */
	if (sdev->busy_poll)
		return EPOLLERR;

	poll_wait(file, &sdev->wait, wait);
	guard(mutex)(&sdev->lock);
	if (tbstream_dev_valid(sdev) != 0)
		return EPOLLHUP | EPOLLERR;

	/*
	 * The RX completions are only advanced from here and from
	 * read(2) so do that now, otherwise we would never report
	 * anything to be available.
	 */
	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)) {
		mask |= EPOLLIN | EPOLLRDNORM;
	} else {
		tbstream_dev_complete_rx(sdev);
		if (tb_ring_poll_pending(sdev->rx_ring.ring))
			mask |= EPOLLIN | EPOLLRDNORM;
	}

	return mask;
}

static int tbstream_dev_fops_open(struct inode *inode, struct file *file)
{
	struct tbstream_dev *sdev = to_tbstream_dev(file->private_data);
	int ret;

	tbstream_dev_get(sdev);

	if (mutex_lock_interruptible(&sdev->lock)) {
		tbstream_dev_put(sdev);
		return -ERESTARTSYS;
	}

	/*
	 * If there is no stream attached yet, block until it appears
	 * unless this is opened in non-blocking mode.
	 */
	while ((ret = tbstream_dev_valid(sdev))) {
		mutex_unlock(&sdev->lock);

		if (ret != -ENXIO || (file->f_flags & O_NONBLOCK))
			goto err_put;

		ret = wait_event_interruptible(sdev->wait,
				tbstream_dev_valid(sdev) == 0 ||
				tbstream_dev_removed(sdev));
		if (ret)
			goto err_put;

		if (tbstream_dev_removed(sdev)) {
			ret = -ENXIO;
			goto err_put;
		}

		if (mutex_lock_interruptible(&sdev->lock)) {
			ret = -ERESTARTSYS;
			goto err_put;
		}
	}

	/* Only on first open we allocate rings and enable paths */
	if (!sdev->users++) {
		ret = tbstream_dev_start(sdev);
		if (ret) {
			sdev->users--;
			goto err_unlock;
		}

		sdev->close_received = false;
		sdev->closed = false;
	}

	mutex_unlock(&sdev->lock);

	/* Stream handles IOCB_NOWAIT just fine */
	file->f_mode |= FMODE_NOWAIT;
	return 0;

err_unlock:
	mutex_unlock(&sdev->lock);
err_put:
	tbstream_dev_put(sdev);

	return ret;
}

static int tbstream_dev_fops_release(struct inode *inode, struct file *file)
{
	struct tbstream_dev *sdev = to_tbstream_dev(file->private_data);

	mutex_lock(&sdev->lock);
	if (--sdev->users == 0) {
		/*
		 * Advance now in case there is CLOSE waiting in the RX
		 * ring.
		 */
		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);

	tbstream_dev_put(sdev);
	return 0;
}

static const struct file_operations tbstream_dev_fops = {
	.owner = THIS_MODULE,
	.llseek = noop_llseek,
	.read_iter = tbstream_dev_fops_read_iter,
	.write_iter = tbstream_dev_fops_write_iter,
	.poll = tbstream_dev_fops_poll,
	.open = tbstream_dev_fops_open,
	.release = tbstream_dev_fops_release,
};

static inline struct tbstream_dev *
tbstream_dev_from_group(struct config_group *group)
{
	return container_of(group, struct tbstream_dev, group);
}

static ssize_t tbstream_dev_busy_poll_show(struct config_item *item, char *buf)
{
	struct config_group *group = to_config_group(item);
	struct tbstream_dev *sdev = tbstream_dev_from_group(group);

	return sysfs_emit(buf, "%u\n", sdev->busy_poll);
}

static ssize_t
tbstream_dev_busy_poll_store(struct config_item *item, const char *buf,
			     size_t count)
{
	struct config_group *group = to_config_group(item);
	struct tbstream_dev *sdev = tbstream_dev_from_group(group);
	bool busy_poll;
	int ret;

	ret = kstrtobool(buf, &busy_poll);
	if (ret)
		return ret;

	guard(mutex)(&sdev->lock);
	if (sdev->users)
		return -EBUSY;
	sdev->busy_poll = busy_poll;
	return count;
}
CONFIGFS_ATTR(tbstream_dev_, busy_poll);

static ssize_t tbstream_dev_index_show(struct config_item *item, char *buf)
{
	struct config_group *group = to_config_group(item);
	struct tbstream_dev *sdev = tbstream_dev_from_group(group);

	return sysfs_emit(buf, "%d\n", sdev->index);
}
CONFIGFS_ATTR_RO(tbstream_dev_, index);

static ssize_t tbstream_dev_in_hopid_show(struct config_item *item, char *buf)
{
	struct config_group *group = to_config_group(item);
	struct tbstream_dev *sdev = tbstream_dev_from_group(group);

	return sysfs_emit(buf, "%d\n", sdev->in_hopid);
}

/* svc->lock must be held */
static void service_remove_properties(struct tb_service *svc, const char *name)
{
	struct tb_property *p;

	if (!svc->local_properties)
		return;

	p = tb_property_find(svc->local_properties, name,
			     TB_PROPERTY_TYPE_DIRECTORY);
	if (p) {
		tb_property_free_dir(p->value.dir);
		tb_property_remove(p);

		dev_dbg(&svc->dev, "removed local directory %s\n", name);

		/*
		 * Is the service directory empty already? If it is then
		 * we can release it as well.
		 */
		tb_property_for_each(svc->local_properties, p) {
			if (p->type == TB_PROPERTY_TYPE_DIRECTORY)
				return;
		}

		tb_property_free_dir(svc->local_properties);
		svc->local_properties = NULL;
	}
}

static int service_update_properties(struct tb_service *svc, const char *name,
				     int in_hopid, int out_hopid)
{
	struct tb_property_dir *dir;
	struct tb_property *p;

	guard(mutex)(&svc->lock);

	if (in_hopid < 8 || out_hopid < 8) {
		service_remove_properties(svc, name);
		return 0;
	}

	if (!svc->local_properties) {
		/*
		 * Add the service directory first time we
		 * populate the entries.
		 */
		svc->local_properties = tb_property_copy_dir(tbstream_dir);
		if (!svc->local_properties)
			return -ENOMEM;
	}

	p = tb_property_find(svc->local_properties, name,
			     TB_PROPERTY_TYPE_DIRECTORY);
	if (p) {
		dir = p->value.dir;

		p = tb_property_find(dir, "inhopid", TB_PROPERTY_TYPE_VALUE);
		if (p && p->value.immediate != in_hopid)
			p->value.immediate = in_hopid;
		p = tb_property_find(dir, "outhopid", TB_PROPERTY_TYPE_VALUE);
		if (p && p->value.immediate != out_hopid)
			p->value.immediate = out_hopid;

		dev_dbg(&svc->dev,
			"updated local directory %s: in HopID %d, out HopID %d\n",
			name, in_hopid, out_hopid);
	} else {
		uuid_t uuid;
		int ret;

		uuid_gen(&uuid);
		dir = tb_property_create_dir(&uuid);
		if (!dir)
			return -ENOMEM;

		tb_property_add_immediate(dir, "inhopid", in_hopid);
		tb_property_add_immediate(dir, "outhopid", out_hopid);

		ret = tb_property_add_dir(svc->local_properties, name, dir);
		if (ret) {
			tb_property_free_dir(dir);
			return ret;
		}

		dev_dbg(&svc->dev,
			"added local directory %s: in HopID %d, out HopID %d\n",
			name, in_hopid, out_hopid);
	}

	return 0;
}

static int tbstream_dev_update_properties(struct tbstream_dev *sdev)
{
	struct tbstream *stream;
	int ret;

	stream = tbstream_get(sdev->stream);
	if (!stream)
		return 0;

	ret = service_update_properties(stream->svc,
					config_item_name(&sdev->group.cg_item),
					sdev->in_hopid, sdev->out_hopid);
	if (!ret)
		tb_service_properties_changed(stream->svc);

	tbstream_put(stream);
	return ret;
}

static int tbstream_dev_alloc_in_hopid(struct tbstream_dev *sdev, int hopid)
{
	struct tb_xdomain *xd = tbstream_dev_xdomain(sdev);
	int ret;

	if (sdev->in_hopid > 0 && sdev->in_hopid != hopid)
		tb_xdomain_release_in_hopid(xd, sdev->in_hopid);
	if (!hopid) {
		sdev->in_hopid = hopid;
		return 0;
	}
	ret = tb_xdomain_alloc_in_hopid(xd, hopid);
	if (ret < 0)
		return ret;
	/*
	 * If specific HopID was asked by the user and we did not get
	 * that one then release and return error instead.
	 */
	if (hopid > 0 && hopid != ret) {
		tb_xdomain_release_in_hopid(xd, ret);
		return -EBUSY;
	}
	sdev->in_hopid = ret;
	return 0;
}

static int tbstream_dev_alloc_out_hopid(struct tbstream_dev *sdev, int hopid)
{
	struct tb_xdomain *xd = tbstream_dev_xdomain(sdev);
	int ret;

	if (sdev->out_hopid > 0 && sdev->out_hopid != hopid)
		tb_xdomain_release_out_hopid(xd, sdev->out_hopid);
	if (!hopid) {
		sdev->out_hopid = hopid;
		return 0;
	}
	ret = tb_xdomain_alloc_out_hopid(xd, hopid);
	if (ret < 0)
		return ret;
	if (hopid > 0 && hopid != ret) {
		tb_xdomain_release_out_hopid(xd, ret);
		return -EBUSY;
	}
	sdev->out_hopid = ret;
	return 0;
}

static ssize_t
tbstream_dev_in_hopid_store(struct config_item *item, const char *buf,
			    size_t count)
{
	struct config_group *group = to_config_group(item);
	struct tbstream_dev *sdev = tbstream_dev_from_group(group);
	int ret, in_hopid;

	ret = kstrtoint(buf, 0, &in_hopid);
	if (ret)
		return ret;

	guard(mutex)(&sdev->lock);
	if (sdev->users)
		return -EBUSY;
	if (sdev->stream) {
		ret = tbstream_dev_alloc_in_hopid(sdev, in_hopid);
		if (ret)
			return ret;
		ret = tbstream_dev_update_properties(sdev);
	} else {
		sdev->in_hopid = in_hopid;
	}
	return ret ? ret : count;
}
CONFIGFS_ATTR(tbstream_dev_, in_hopid);

static ssize_t tbstream_dev_out_hopid_show(struct config_item *item, char *buf)
{
	struct config_group *group = to_config_group(item);
	struct tbstream_dev *sdev = tbstream_dev_from_group(group);

	return sysfs_emit(buf, "%d\n", sdev->out_hopid);
}

static ssize_t
tbstream_dev_out_hopid_store(struct config_item *item, const char *buf,
			     size_t count)
{
	struct config_group *group = to_config_group(item);
	struct tbstream_dev *sdev = tbstream_dev_from_group(group);
	int ret, out_hopid;

	ret = kstrtoint(buf, 0, &out_hopid);
	if (ret)
		return ret;

	guard(mutex)(&sdev->lock);
	if (sdev->users)
		return -EBUSY;
	if (sdev->stream) {
		ret = tbstream_dev_alloc_out_hopid(sdev, out_hopid);
		if (ret)
			return ret;
		ret = tbstream_dev_update_properties(sdev);
	} else {
		sdev->out_hopid = out_hopid;
	}
	return ret ? ret : count;
}
CONFIGFS_ATTR(tbstream_dev_, out_hopid);

static ssize_t tbstream_dev_ring_size_show(struct config_item *item, char *buf)
{
	struct config_group *group = to_config_group(item);
	struct tbstream_dev *sdev = tbstream_dev_from_group(group);

	return sysfs_emit(buf, "%u\n", sdev->ring_size);
}

static ssize_t
tbstream_dev_ring_size_store(struct config_item *item, const char *buf,
			     size_t count)
{
	struct config_group *group = to_config_group(item);
	struct tbstream_dev *sdev = tbstream_dev_from_group(group);
	unsigned int ring_size;
	int ret;

	ret = kstrtouint(buf, 0, &ring_size);
	if (ret)
		return ret;

	if (ring_size < TBSTREAM_DEV_MIN_RING_SIZE ||
	    ring_size > TBSTREAM_DEV_MAX_RING_SIZE)
		return -EINVAL;

	guard(mutex)(&sdev->lock);
	if (sdev->users)
		return -EBUSY;
	sdev->ring_size = ring_size;
	return count;
}
CONFIGFS_ATTR(tbstream_dev_, ring_size);

static ssize_t tbstream_dev_throttling_show(struct config_item *item, char *buf)
{
	struct config_group *group = to_config_group(item);
	struct tbstream_dev *sdev = tbstream_dev_from_group(group);

	return sysfs_emit(buf, "%u\n", sdev->throttling);
}

static ssize_t
tbstream_dev_throttling_store(struct config_item *item, const char *buf,
			      size_t count)
{
	struct config_group *group = to_config_group(item);
	struct tbstream_dev *sdev = tbstream_dev_from_group(group);
	unsigned int throttling;
	int ret;

	ret = kstrtouint(buf, 0, &throttling);
	if (ret)
		return ret;

	if (throttling > TBSTREAM_DEV_MAX_THROTTLING)
		return -EINVAL;

	guard(mutex)(&sdev->lock);
	if (sdev->users)
		return -EBUSY;
	sdev->throttling = throttling;
	return count;
}
CONFIGFS_ATTR(tbstream_dev_, throttling);

static struct configfs_attribute *tbstream_dev_attrs[] = {
	&tbstream_dev_attr_busy_poll,
	&tbstream_dev_attr_index,
	&tbstream_dev_attr_in_hopid,
	&tbstream_dev_attr_out_hopid,
	&tbstream_dev_attr_ring_size,
	&tbstream_dev_attr_throttling,
	NULL,
};

static void tbstream_dev_item_release(struct config_item *item)
{
	struct config_group *group = to_config_group(item);
	struct tbstream_dev *sdev = tbstream_dev_from_group(group);

	misc_deregister(&sdev->misc);
	tbstream_dev_put(sdev);
}

static struct configfs_item_operations tbstream_dev_item_ops = {
	.release = tbstream_dev_item_release,
};

static const struct config_item_type tbstream_dev_type = {
	.ct_owner = THIS_MODULE,
	.ct_item_ops = &tbstream_dev_item_ops,
	.ct_attrs = tbstream_dev_attrs,
};

static void service_get_hopids(struct tb_service *svc, const char *name,
			       int *in_hopid, int *out_hopid)
{
	struct tb_property_dir *dir;
	struct tb_property *p;

	guard(mutex)(&svc->lock);

	/* See if we have directory entry with the matching name */
	p = tb_property_find(svc->remote_properties, name,
			     TB_PROPERTY_TYPE_DIRECTORY);
	if (!p)
		return;

	dir = p->value.dir;

	/*
	 * We need to reverse the HopIDs on our end so that in becomes
	 * out and vice versa.
	 */
	p = tb_property_find(dir, "inhopid", TB_PROPERTY_TYPE_VALUE);
	if (p && p->value.immediate >= 8)
		*out_hopid = p->value.immediate;
	p = tb_property_find(dir, "outhopid", TB_PROPERTY_TYPE_VALUE);
	if (p && p->value.immediate >= 8)
		*in_hopid = p->value.immediate;
}

static void
tbstream_dev_attach_stream(struct tbstream_dev *sdev, struct tbstream_group *sg)
{
	const char *name = config_item_name(&sdev->group.cg_item);
	struct tbstream *stream;

	stream = tbstream_get(sg->stream);
	if (!stream)
		return;

	scoped_guard(mutex, &sdev->lock) {
		sdev->stream = stream;
		/*
		 * If there is no existing configuration (or automatic
		 * configuration is being used) check if the other side
		 * has configuration for this and use it.
		 */
		if (sdev->in_hopid <= 0 && sdev->out_hopid <= 0)
			service_get_hopids(stream->svc, name, &sdev->in_hopid,
					   &sdev->out_hopid);
		if (sdev->in_hopid)
			tbstream_dev_alloc_in_hopid(sdev, sdev->in_hopid);
		if (sdev->out_hopid)
			tbstream_dev_alloc_out_hopid(sdev, sdev->out_hopid);
	}

	service_update_properties(stream->svc, name, sdev->in_hopid,
				  sdev->out_hopid);
	tb_service_properties_changed(stream->svc);

	/* Notify any openerers that the stream is now attached */
	wake_up_interruptible(&sdev->wait);
}

static void tbstream_dev_detach_stream(struct tbstream_dev *sdev)
{
	const char *name = config_item_name(&sdev->group.cg_item);
	struct tbstream *stream;
	struct tb_xdomain *xd;

	scoped_guard(mutex, &sdev->lock) {
		stream = sdev->stream;
		if (!stream)
			return;
		sdev->stream = NULL;
		xd = tb_service_parent(stream->svc);
		if (sdev->out_hopid > 0)
			tb_xdomain_release_out_hopid(xd, sdev->out_hopid);
		if (sdev->in_hopid > 0)
			tb_xdomain_release_in_hopid(xd, sdev->in_hopid);
	}

	service_update_properties(stream->svc, name, 0, 0);
	tb_service_properties_changed(stream->svc);

	tbstream_put(stream);

	/* Notify any task that the stream is not valid anymore */
	wake_up_interruptible_poll(&sdev->wait, EPOLLHUP | EPOLLERR);
}

static inline struct tbstream_group *
to_tbstream_group(struct config_group *group)
{
	return container_of(group, struct tbstream_group, group);
}

static struct config_group *
tbstream_dev_make_group(struct config_group *group, const char *name)
{
	struct tbstream_group *sg = to_tbstream_group(group);
	struct tbstream_dev *sdev;
	int ret, index;

	/*
	 * We want the names to be suitable for passing as property
	 * directory names.
	 */
	if (strlen(name) > TB_PROPERTY_KEY_SIZE)
		return ERR_PTR(-ENAMETOOLONG);

	sdev = kzalloc_obj(*sdev);
	if (!sdev)
		return ERR_PTR(-ENOMEM);

	index = ida_alloc(&tbstream_indices, GFP_KERNEL);
	if (index < 0) {
		kfree(sdev);
		return ERR_PTR(index);
	}

	sdev->index = index;
	sdev->ring_size = TBSTREAM_DEV_RING_SIZE;
	sdev->throttling = TBSTREAM_DEV_THROTTLING;
	mutex_init(&sdev->lock);
	init_waitqueue_head(&sdev->wait);
	INIT_LIST_HEAD(&sdev->list);
	/* This point forward tbstream_dev_put() must be used to release sdev */
	kref_init(&sdev->kref);

	config_group_init_type_name(&sdev->group, name, &tbstream_dev_type);

	scoped_guard(mutex, &sg->lock)
		list_add_tail(&sdev->list, &sg->dev_list);

	tbstream_dev_attach_stream(sdev, sg);

	sdev->misc.name = kasprintf(GFP_KERNEL, "tbstream%d", index);
	sdev->misc.minor = MISC_DYNAMIC_MINOR;
	sdev->misc.fops = &tbstream_dev_fops;

	ret = misc_register(&sdev->misc);
	if (ret) {
		tbstream_dev_detach_stream(sdev);
		scoped_guard(mutex, &sg->lock)
			list_del(&sdev->list);
		/* Calls tbstream_dev_put() */
		config_group_put(&sdev->group);
		return ERR_PTR(ret);
	}

	return &sdev->group;
}

static void
tbstream_dev_drop_item(struct config_group *group, struct config_item *item)
{
	struct config_group *sdev_group = to_config_group(item);
	struct tbstream_dev *sdev = tbstream_dev_from_group(sdev_group);
	struct tbstream_group *sg = to_tbstream_group(group);

	scoped_guard(mutex, &sg->lock)
		list_del(&sdev->list);
	/* Notify any task that the underlying group was removed  */
	sdev->removed = true;
	wake_up_interruptible_poll(&sdev->wait, EPOLLHUP | EPOLLERR);
	config_item_put(item);
}

static struct configfs_group_operations tbstream_dev_group_ops = {
	.make_group = tbstream_dev_make_group,
	.drop_item = tbstream_dev_drop_item,
};

static void tbstream_item_release(struct config_item *item)
{
	struct config_group *group = to_config_group(item);
	struct tbstream_group *sg = to_tbstream_group(group);

	tbstream_put(sg->stream);
	kfree(sg);
}

static struct configfs_item_operations tbstream_item_ops = {
	.release = tbstream_item_release,
};

static const struct config_item_type tbstream_dev_group_type = {
	.ct_owner = THIS_MODULE,
	.ct_group_ops = &tbstream_dev_group_ops,
	.ct_item_ops = &tbstream_item_ops,
};

static struct config_group *
tbstream_make_group(struct config_group *group, const char *name)
{
	struct tbstream_group *sg;
	struct tbstream *stream;
	int domain, index;
	u64 route;

	/* Make sure the format is correct */
	if (sscanf(name, "%u-%llx.%u", &domain, &route, &index) != 3)
		return ERR_PTR(-EINVAL);

	sg = kzalloc_obj(*sg);
	if (!sg)
		return ERR_PTR(-ENOMEM);

	mutex_init(&sg->lock);
	INIT_LIST_HEAD(&sg->dev_list);

	guard(mutex)(&tbstream_lock);
	list_for_each_entry(stream, &tbstream_list, list) {
		tbstream_get(stream);
		if (sysfs_streq(name, dev_name(&stream->svc->dev))) {
			sg->stream = stream;
			break;
		}
		tbstream_put(stream);
	}

	config_group_init_type_name(&sg->group, name, &tbstream_dev_group_type);
	return &sg->group;
}

static struct configfs_group_operations tbstream_group_ops = {
	.make_group = tbstream_make_group,
};

static const struct config_item_type tbstream_group_type = {
	.ct_owner = THIS_MODULE,
	.ct_group_ops = &tbstream_group_ops,
};

static struct config_group tbstream_group = {
	.cg_item = {
		.ci_namebuf = "stream",
		.ci_type = &tbstream_group_type,
	},
};

/* Returns reference count increased */
static struct tbstream_group *tbstream_group_find(struct tbstream *stream)
{
	const char *name = dev_name(&stream->svc->dev);
	struct config_item *item;

	guard(mutex)(&tbstream_group.cg_subsys->su_mutex);
	item = config_group_find_item(&tbstream_group, name);
	if (!item)
		return NULL;
	return to_tbstream_group(to_config_group(item));
}

static void tbstream_group_attach_stream(struct tbstream *stream)
{
	struct tbstream_group *sg;
	struct tbstream_dev *sdev;

	sg = tbstream_group_find(stream);
	if (!sg)
		return;

	guard(mutex)(&sg->lock);
	if (WARN_ON(sg->stream)) {
		config_group_put(&sg->group);
		return;
	}
	sg->stream = tbstream_get(stream);
	/*
	 * If there are existing stream devices, attach the stream to
	 * them now.
	 */
	list_for_each_entry(sdev, &sg->dev_list, list) {
		tbstream_dev_get(sdev);
		tbstream_dev_attach_stream(sdev, sg);
		tbstream_dev_put(sdev);
	}

	config_group_put(&sg->group);
}

static void tbstream_group_detach_stream(struct tbstream *stream)
{
	struct tbstream_group *sg;
	struct tbstream_dev *sdev;

	sg = tbstream_group_find(stream);
	if (!sg)
		return;

	guard(mutex)(&sg->lock);
	if (sg->stream) {
		/* Detach this stream from the stream devices */
		list_for_each_entry_reverse(sdev, &sg->dev_list, list) {
			tbstream_dev_get(sdev);
			tbstream_dev_detach_stream(sdev);
			tbstream_dev_put(sdev);
		}
		tbstream_put(sg->stream);
		sg->stream = NULL;
	}

	config_group_put(&sg->group);
}

static int tbstream_probe(struct tb_service *svc)
{
	struct tbstream *stream;

	stream = kzalloc_obj(*stream);
	if (!stream)
		return -ENOMEM;

	/* After this point, release stream by calling tbstream_put() */
	kref_init(&stream->kref);
	stream->svc = tb_service_get(svc);
	INIT_LIST_HEAD(&stream->list);

	scoped_guard(mutex, &tbstream_lock)
		list_add_tail(&stream->list, &tbstream_list);

	tbstream_group_attach_stream(stream);
	tb_service_set_drvdata(svc, stream);
	return 0;
}

static void tbstream_remove(struct tb_service *svc)
{
	struct tbstream *stream = tb_service_get_drvdata(svc);

	tbstream_group_detach_stream(stream);
	scoped_guard(mutex, &tbstream_lock)
		list_del(&stream->list);
	tbstream_put(stream);
}

static int __maybe_unused tbstream_suspend(struct device *dev)
{
	struct tb_service *svc = tb_to_service(dev);
	struct tbstream *stream = tb_service_get_drvdata(svc);
	struct tbstream_group *sg;
	struct tbstream_dev *sdev;

	sg = tbstream_group_find(stream);
	if (!sg)
		return 0;

	list_for_each_entry_reverse(sdev, &sg->dev_list, list) {
		tbstream_dev_get(sdev);
		/* Stop the stream (if it was open) */
		if (sdev->users)
			tbstream_dev_stop(sdev);
		tbstream_dev_put(sdev);
	}

	config_group_put(&sg->group);
	return 0;
}

static int __maybe_unused tbstream_resume(struct device *dev)
{
	struct tb_service *svc = tb_to_service(dev);
	struct tbstream *stream = tb_service_get_drvdata(svc);
	struct tbstream_group *sg;
	struct tbstream_dev *sdev;

	sg = tbstream_group_find(stream);
	if (!sg)
		return 0;

	list_for_each_entry(sdev, &sg->dev_list, list) {
		tbstream_dev_get(sdev);
		if (sdev->users) {
			int ret;

			ret = tbstream_dev_start(sdev);
			if (ret) {
				tbstream_dev_put(sdev);
				config_group_put(&sg->group);
				return ret;
			}
		}
		tbstream_dev_put(sdev);
	}

	config_group_put(&sg->group);
	return 0;
}

static const struct dev_pm_ops tbstream_pm_ops = {
	SET_SYSTEM_SLEEP_PM_OPS(tbstream_suspend, tbstream_resume)
};

static const struct tb_service_id tbstream_ids[] = {
	{ TB_SERVICE("stream", 1) },
	{ }
};
MODULE_DEVICE_TABLE(tbsvc, tbstream_ids);

static struct tb_service_driver tbstream_driver = {
	.driver = {
		.owner = THIS_MODULE,
		.name = "thunderbolt_stream",
		.pm = &tbstream_pm_ops,
	},
	.probe = tbstream_probe,
	.remove = tbstream_remove,
	.id_table = tbstream_ids,
};

static int __init tbstream_init(void)
{
	int ret;

	tbstream_dir = tb_property_create_dir(&tbstream_dir_uuid);
	if (!tbstream_dir)
		return -ENOMEM;

	tb_property_add_immediate(tbstream_dir, "prtcid", 1);
	tb_property_add_immediate(tbstream_dir, "prtcvers", 1);
	tb_property_add_immediate(tbstream_dir, "prtcrevs", 0);
	tb_property_add_immediate(tbstream_dir, "prtcstns", 0);

	ret = tb_register_property_dir("stream", tbstream_dir);
	if (ret)
		goto err_free_dir;

	config_group_init(&tbstream_group);
	ret = tb_configfs_register_group(&tbstream_group);
	if (ret)
		goto err_unregister_dir;

	ret = tb_register_service_driver(&tbstream_driver);
	if (ret)
		goto err_unregister_group;
	return 0;

err_unregister_group:
	tb_configfs_unregister_group(&tbstream_group);
err_unregister_dir:
	tb_unregister_property_dir("stream", tbstream_dir);
err_free_dir:
	tb_property_free_dir(tbstream_dir);
	return ret;
}
module_init(tbstream_init);

static void __exit tbstream_exit(void)
{
	tb_unregister_service_driver(&tbstream_driver);
	tb_configfs_unregister_group(&tbstream_group);
	tb_unregister_property_dir("stream", tbstream_dir);
	tb_property_free_dir(tbstream_dir);
	ida_destroy(&tbstream_indices);
}
module_exit(tbstream_exit);

MODULE_AUTHOR("Alan Borzeszkowski <alan.borzeszkowski@linux.intel.com>");
MODULE_AUTHOR("Mika Westerberg <mika.westerberg@linux.intel.com>");
MODULE_DESCRIPTION("Stream data over Thunderbolt/USB4 cable");
MODULE_LICENSE("GPL");