drm/amd/display: Try RGB before YCbCr 4:4:4 in stream validationgrafted

amdgpu_dm_create_validate_stream_for_sink() walks encoding_order[] and
uses the first encoding that validates. YCbCr 4:4:4 is listed before
RGB, so an HDMI sink that advertises 4:4:4 gets YCbCr 4:4:4 whenever the
"color format" property is left at AUTO, even though RGB fits the same
link.

That contradicts the documented AUTO behaviour for HDMI in enum
drm_connector_color_format (RGB, falling back to YCbCr 4:2:0 only when
the bandwidth is not available or the mode is 4:2:0-only), which the
amdgpu implementation of the property also describes. It also leaves
the "Broadcast RGB" property without effect on such sinks, since the
quantization range it selects only applies to RGB output.

Try RGB first. The mask still holds every encoding the sink supports,
so a mode that cannot carry RGB falls back exactly as before.

For reference, v7.2 picked RGB here unless YCbCr 4:4:4 was forced
through debugfs, while earlier kernels picked YCbCr 4:4:4 for any HDMI
sink that advertised it.

Fixes: 0b0ff65d3ca1 ("drm/amd/display: Refactor stream validation")
Suggested-by: Adolfo Rodrigues <adolfotregosa@gmail.com>
Assisted-by: Claude Code:claude-fable-5-1
Signed-off-by: Adrian Betschart <adrian.betschart@cinemaone.ch>
Reviewed-by: Fangzhi Zuo <jerry.zuo@amd.com>
Tested-by: Adolfo Rodrigues <adolfotregosa@gmail.com>
Tested-by: Daniel Wheeler <daniel.wheeler@amd.com>
Signed-off-by: Alex Deucher <alexander.deucher@amd.com>
(cherry picked from commit 7de432bc753133764dc40d37f9388da56de251e7)

1 files changed, 1850 insertions, 0 deletions

diff --git a/drivers/thunderbolt/stream.c b/drivers/thunderbolt/stream.c
new file mode 100644
index 000000000..25c259dd0
--- /dev/null
+++ b/drivers/thunderbolt/stream.c
@@ -0,0 +1,1850 @@
+// 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
+ * @users: Number of times @cdev has been opened
+ * @closed: CLOSE packet was received
+ * @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
+ */
+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;
+	int users;
+	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 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_closed(const struct tbstream_dev *sdev)
+{
+	return sdev->closed;
+}
+
+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;
+
+	if (canceled)
+		return;
+
+	sf->completed = true;
+	sdev->rx_ring.prod++;
+
+	if (sf->frame.flags & RING_DESC_CRC_ERROR)
+		pr_warn("RX CRC error\n");
+	else if (sf->frame.flags & RING_DESC_BUFFER_OVERRUN)
+		pr_warn("RX buffer overrun\n");
+	else
+		wake_up_interruptible_poll(&sdev->wait, EPOLLIN | EPOLLRDNORM);
+}
+
+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 void
+tbstream_dev_poll_ring(struct tbstream_dev *sdev, struct tbstream_ring *ring)
+{
+	struct ring_frame *frame;
+
+	if (!sdev->busy_poll)
+		return;
+
+	while ((frame = tb_ring_poll(ring->ring)))
+		frame->callback(ring->ring, frame, false);
+}
+
+static int tbstream_dev_send_close(struct tbstream_dev *sdev)
+{
+	struct tbstream_frame *sf;
+
+	if (sdev->busy_poll) {
+		/*
+		 * 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.
+		 */
+		ktime_t timeout = ktime_add_ms(ktime_get(), 500);
+
+		do {
+			if (tbstream_ring_available(&sdev->tx_ring))
+				break;
+			tbstream_dev_poll_ring(sdev, &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 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;
+	u16 sof_mask, eof_mask;
+	struct tb_ring *ring;
+	int ret, e2e_tx_hop;
+
+	if (sdev->busy_poll)
+		flags |= RING_FLAG_NO_INTERRUPT;
+
+	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, NULL, NULL);
+	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);
+
+	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_dev_poll_ring(sdev, &sdev->tx_ring);
+			fsleep(15);
+		} while (ktime_before(ktime_get(), timeout));
+
+		tb_ring_stop(sdev->tx_ring.ring);
+		tb_ring_stop(sdev->rx_ring.ring);
+	} else {
+		tb_ring_flush(sdev->tx_ring.ring, 500);
+		tb_ring_stop(sdev->tx_ring.ring);
+		tb_ring_flush(sdev->rx_ring.ring, 500);
+		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;
+}
+
+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 (;;) {
+		/* When busy polling, advance any completions manually */
+		tbstream_dev_poll_ring(sdev, &sdev->rx_ring);
+
+		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;
+
+		mutex_unlock(&sdev->lock);
+
+		if (nowait)
+			return -EAGAIN;
+
+		if (sdev->busy_poll) {
+			if (signal_pending(current))
+				return -ERESTARTSYS;
+			cond_resched();
+		} else {
+			ret = wait_event_interruptible(sdev->wait,
+					tbstream_ring_available(&sdev->rx_ring) ||
+					tbstream_dev_valid(sdev) != 0 ||
+					tbstream_dev_closed(sdev) ||
+					tbstream_dev_removed(sdev));
+			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);
+				sdev->closed = true;
+			}
+			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 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 (;;) {
+		tbstream_dev_poll_ring(sdev, &sdev->tx_ring);
+
+		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 -ENXIO;
+		}
+
+		if (tbstream_ring_available(&sdev->tx_ring))
+			break;
+
+		mutex_unlock(&sdev->lock);
+
+		if (nowait)
+			return -EAGAIN;
+
+		if (sdev->busy_poll) {
+			if (signal_pending(current))
+				return -ERESTARTSYS;
+			cond_resched();
+		} 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));
+			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) {
+		mask |= EPOLLHUP | EPOLLERR;
+	} else {
+		if (tbstream_ring_available(&sdev->tx_ring))
+			mask |= EPOLLOUT | EPOLLWRNORM;
+		if (tbstream_ring_available(&sdev->rx_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->closed = false;
+	}
+
+	mutex_unlock(&sdev->lock);
+	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) {
+		/*
+		 * Send CLOSE tunneled packet to notify the other end
+		 * that we are closing the file. We do this twice if the
+		 * first one fails.
+		 */
+		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");