Something went wrong. Try again.
(I'm so sorry!) Linux kernel mirror (for testing) git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git
kernel os linux
Something went wrong. Try again.
50 kB · 2000 lines
C
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001// 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 voidtbstream_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 voidtbstream_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 inttbstream_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 booltbstream_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_ttbstream_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_ttbstream_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_ttbstream_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_ttbstream_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_ttbstream_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_ttbstream_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_ttbstream_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_ttbstream_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 voidtbstream_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 voidtbstream_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");