ptr_ring.h 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674
  1. /* SPDX-License-Identifier: GPL-2.0-or-later */
  2. /*
  3. * Definitions for the 'struct ptr_ring' datastructure.
  4. *
  5. * Author:
  6. * Michael S. Tsirkin <mst@redhat.com>
  7. *
  8. * Copyright (C) 2016 Red Hat, Inc.
  9. *
  10. * This is a limited-size FIFO maintaining pointers in FIFO order, with
  11. * one CPU producing entries and another consuming entries from a FIFO.
  12. *
  13. * This implementation tries to minimize cache-contention when there is a
  14. * single producer and a single consumer CPU.
  15. */
  16. #ifndef _LINUX_PTR_RING_H
  17. #define _LINUX_PTR_RING_H 1
  18. #ifdef __KERNEL__
  19. #include <linux/spinlock.h>
  20. #include <linux/cache.h>
  21. #include <linux/types.h>
  22. #include <linux/compiler.h>
  23. #include <linux/slab.h>
  24. #include <linux/mm.h>
  25. #include <asm/errno.h>
  26. #endif
  27. struct ptr_ring {
  28. int producer ____cacheline_aligned_in_smp;
  29. spinlock_t producer_lock;
  30. int consumer_head ____cacheline_aligned_in_smp; /* next valid entry */
  31. int consumer_tail; /* next entry to invalidate */
  32. spinlock_t consumer_lock;
  33. /* Shared consumer/producer data */
  34. /* Read-only by both the producer and the consumer */
  35. int size ____cacheline_aligned_in_smp; /* max entries in queue */
  36. int batch; /* number of entries to consume in a batch */
  37. void **queue;
  38. };
  39. /* Note: callers invoking this in a loop must use a compiler barrier,
  40. * for example cpu_relax().
  41. *
  42. * NB: this is unlike __ptr_ring_empty in that callers must hold producer_lock:
  43. * see e.g. ptr_ring_full.
  44. */
  45. static inline bool __ptr_ring_full(struct ptr_ring *r)
  46. {
  47. return r->queue[r->producer];
  48. }
  49. static inline bool ptr_ring_full(struct ptr_ring *r)
  50. {
  51. bool ret;
  52. spin_lock(&r->producer_lock);
  53. ret = __ptr_ring_full(r);
  54. spin_unlock(&r->producer_lock);
  55. return ret;
  56. }
  57. static inline bool ptr_ring_full_irq(struct ptr_ring *r)
  58. {
  59. bool ret;
  60. spin_lock_irq(&r->producer_lock);
  61. ret = __ptr_ring_full(r);
  62. spin_unlock_irq(&r->producer_lock);
  63. return ret;
  64. }
  65. static inline bool ptr_ring_full_any(struct ptr_ring *r)
  66. {
  67. unsigned long flags;
  68. bool ret;
  69. spin_lock_irqsave(&r->producer_lock, flags);
  70. ret = __ptr_ring_full(r);
  71. spin_unlock_irqrestore(&r->producer_lock, flags);
  72. return ret;
  73. }
  74. static inline bool ptr_ring_full_bh(struct ptr_ring *r)
  75. {
  76. bool ret;
  77. spin_lock_bh(&r->producer_lock);
  78. ret = __ptr_ring_full(r);
  79. spin_unlock_bh(&r->producer_lock);
  80. return ret;
  81. }
  82. /* Note: callers invoking this in a loop must use a compiler barrier,
  83. * for example cpu_relax(). Callers must hold producer_lock.
  84. * Callers are responsible for making sure pointer that is being queued
  85. * points to a valid data.
  86. */
  87. static inline int __ptr_ring_produce(struct ptr_ring *r, void *ptr)
  88. {
  89. if (unlikely(!r->size) || r->queue[r->producer])
  90. return -ENOSPC;
  91. /* Make sure the pointer we are storing points to a valid data. */
  92. /* Pairs with the dependency ordering in __ptr_ring_consume. */
  93. smp_wmb();
  94. WRITE_ONCE(r->queue[r->producer++], ptr);
  95. if (unlikely(r->producer >= r->size))
  96. r->producer = 0;
  97. return 0;
  98. }
  99. /*
  100. * Note: resize (below) nests producer lock within consumer lock, so if you
  101. * consume in interrupt or BH context, you must disable interrupts/BH when
  102. * calling this.
  103. */
  104. static inline int ptr_ring_produce(struct ptr_ring *r, void *ptr)
  105. {
  106. int ret;
  107. spin_lock(&r->producer_lock);
  108. ret = __ptr_ring_produce(r, ptr);
  109. spin_unlock(&r->producer_lock);
  110. return ret;
  111. }
  112. static inline int ptr_ring_produce_irq(struct ptr_ring *r, void *ptr)
  113. {
  114. int ret;
  115. spin_lock_irq(&r->producer_lock);
  116. ret = __ptr_ring_produce(r, ptr);
  117. spin_unlock_irq(&r->producer_lock);
  118. return ret;
  119. }
  120. static inline int ptr_ring_produce_any(struct ptr_ring *r, void *ptr)
  121. {
  122. unsigned long flags;
  123. int ret;
  124. spin_lock_irqsave(&r->producer_lock, flags);
  125. ret = __ptr_ring_produce(r, ptr);
  126. spin_unlock_irqrestore(&r->producer_lock, flags);
  127. return ret;
  128. }
  129. static inline int ptr_ring_produce_bh(struct ptr_ring *r, void *ptr)
  130. {
  131. int ret;
  132. spin_lock_bh(&r->producer_lock);
  133. ret = __ptr_ring_produce(r, ptr);
  134. spin_unlock_bh(&r->producer_lock);
  135. return ret;
  136. }
  137. static inline void *__ptr_ring_peek(struct ptr_ring *r)
  138. {
  139. if (likely(r->size))
  140. return READ_ONCE(r->queue[r->consumer_head]);
  141. return NULL;
  142. }
  143. /*
  144. * Test ring empty status without taking any locks.
  145. *
  146. * NB: This is only safe to call if ring is never resized.
  147. *
  148. * However, if some other CPU consumes ring entries at the same time, the value
  149. * returned is not guaranteed to be correct.
  150. *
  151. * In this case - to avoid incorrectly detecting the ring
  152. * as empty - the CPU consuming the ring entries is responsible
  153. * for either consuming all ring entries until the ring is empty,
  154. * or synchronizing with some other CPU and causing it to
  155. * re-test __ptr_ring_empty and/or consume the ring enteries
  156. * after the synchronization point.
  157. *
  158. * Note: callers invoking this in a loop must use a compiler barrier,
  159. * for example cpu_relax().
  160. */
  161. static inline bool __ptr_ring_empty(struct ptr_ring *r)
  162. {
  163. if (likely(r->size))
  164. return !r->queue[READ_ONCE(r->consumer_head)];
  165. return true;
  166. }
  167. static inline bool ptr_ring_empty(struct ptr_ring *r)
  168. {
  169. bool ret;
  170. spin_lock(&r->consumer_lock);
  171. ret = __ptr_ring_empty(r);
  172. spin_unlock(&r->consumer_lock);
  173. return ret;
  174. }
  175. static inline bool ptr_ring_empty_irq(struct ptr_ring *r)
  176. {
  177. bool ret;
  178. spin_lock_irq(&r->consumer_lock);
  179. ret = __ptr_ring_empty(r);
  180. spin_unlock_irq(&r->consumer_lock);
  181. return ret;
  182. }
  183. static inline bool ptr_ring_empty_any(struct ptr_ring *r)
  184. {
  185. unsigned long flags;
  186. bool ret;
  187. spin_lock_irqsave(&r->consumer_lock, flags);
  188. ret = __ptr_ring_empty(r);
  189. spin_unlock_irqrestore(&r->consumer_lock, flags);
  190. return ret;
  191. }
  192. static inline bool ptr_ring_empty_bh(struct ptr_ring *r)
  193. {
  194. bool ret;
  195. spin_lock_bh(&r->consumer_lock);
  196. ret = __ptr_ring_empty(r);
  197. spin_unlock_bh(&r->consumer_lock);
  198. return ret;
  199. }
  200. /* Must only be called after __ptr_ring_peek returned !NULL */
  201. static inline void __ptr_ring_discard_one(struct ptr_ring *r)
  202. {
  203. /* Fundamentally, what we want to do is update consumer
  204. * index and zero out the entry so producer can reuse it.
  205. * Doing it naively at each consume would be as simple as:
  206. * consumer = r->consumer;
  207. * r->queue[consumer++] = NULL;
  208. * if (unlikely(consumer >= r->size))
  209. * consumer = 0;
  210. * r->consumer = consumer;
  211. * but that is suboptimal when the ring is full as producer is writing
  212. * out new entries in the same cache line. Defer these updates until a
  213. * batch of entries has been consumed.
  214. */
  215. /* Note: we must keep consumer_head valid at all times for __ptr_ring_empty
  216. * to work correctly.
  217. */
  218. int consumer_head = r->consumer_head;
  219. int head = consumer_head++;
  220. /* Once we have processed enough entries invalidate them in
  221. * the ring all at once so producer can reuse their space in the ring.
  222. * We also do this when we reach end of the ring - not mandatory
  223. * but helps keep the implementation simple.
  224. */
  225. if (unlikely(consumer_head - r->consumer_tail >= r->batch ||
  226. consumer_head >= r->size)) {
  227. /* Zero out entries in the reverse order: this way we touch the
  228. * cache line that producer might currently be reading the last;
  229. * producer won't make progress and touch other cache lines
  230. * besides the first one until we write out all entries.
  231. */
  232. while (likely(head >= r->consumer_tail))
  233. r->queue[head--] = NULL;
  234. r->consumer_tail = consumer_head;
  235. }
  236. if (unlikely(consumer_head >= r->size)) {
  237. consumer_head = 0;
  238. r->consumer_tail = 0;
  239. }
  240. /* matching READ_ONCE in __ptr_ring_empty for lockless tests */
  241. WRITE_ONCE(r->consumer_head, consumer_head);
  242. }
  243. static inline void *__ptr_ring_consume(struct ptr_ring *r)
  244. {
  245. void *ptr;
  246. /* The READ_ONCE in __ptr_ring_peek guarantees that anyone
  247. * accessing data through the pointer is up to date. Pairs
  248. * with smp_wmb in __ptr_ring_produce.
  249. */
  250. ptr = __ptr_ring_peek(r);
  251. if (ptr)
  252. __ptr_ring_discard_one(r);
  253. return ptr;
  254. }
  255. static inline int __ptr_ring_consume_batched(struct ptr_ring *r,
  256. void **array, int n)
  257. {
  258. void *ptr;
  259. int i;
  260. for (i = 0; i < n; i++) {
  261. ptr = __ptr_ring_consume(r);
  262. if (!ptr)
  263. break;
  264. array[i] = ptr;
  265. }
  266. return i;
  267. }
  268. /*
  269. * Note: resize (below) nests producer lock within consumer lock, so if you
  270. * call this in interrupt or BH context, you must disable interrupts/BH when
  271. * producing.
  272. */
  273. static inline void *ptr_ring_consume(struct ptr_ring *r)
  274. {
  275. void *ptr;
  276. spin_lock(&r->consumer_lock);
  277. ptr = __ptr_ring_consume(r);
  278. spin_unlock(&r->consumer_lock);
  279. return ptr;
  280. }
  281. static inline void *ptr_ring_consume_irq(struct ptr_ring *r)
  282. {
  283. void *ptr;
  284. spin_lock_irq(&r->consumer_lock);
  285. ptr = __ptr_ring_consume(r);
  286. spin_unlock_irq(&r->consumer_lock);
  287. return ptr;
  288. }
  289. static inline void *ptr_ring_consume_any(struct ptr_ring *r)
  290. {
  291. unsigned long flags;
  292. void *ptr;
  293. spin_lock_irqsave(&r->consumer_lock, flags);
  294. ptr = __ptr_ring_consume(r);
  295. spin_unlock_irqrestore(&r->consumer_lock, flags);
  296. return ptr;
  297. }
  298. static inline void *ptr_ring_consume_bh(struct ptr_ring *r)
  299. {
  300. void *ptr;
  301. spin_lock_bh(&r->consumer_lock);
  302. ptr = __ptr_ring_consume(r);
  303. spin_unlock_bh(&r->consumer_lock);
  304. return ptr;
  305. }
  306. static inline int ptr_ring_consume_batched(struct ptr_ring *r,
  307. void **array, int n)
  308. {
  309. int ret;
  310. spin_lock(&r->consumer_lock);
  311. ret = __ptr_ring_consume_batched(r, array, n);
  312. spin_unlock(&r->consumer_lock);
  313. return ret;
  314. }
  315. static inline int ptr_ring_consume_batched_irq(struct ptr_ring *r,
  316. void **array, int n)
  317. {
  318. int ret;
  319. spin_lock_irq(&r->consumer_lock);
  320. ret = __ptr_ring_consume_batched(r, array, n);
  321. spin_unlock_irq(&r->consumer_lock);
  322. return ret;
  323. }
  324. static inline int ptr_ring_consume_batched_any(struct ptr_ring *r,
  325. void **array, int n)
  326. {
  327. unsigned long flags;
  328. int ret;
  329. spin_lock_irqsave(&r->consumer_lock, flags);
  330. ret = __ptr_ring_consume_batched(r, array, n);
  331. spin_unlock_irqrestore(&r->consumer_lock, flags);
  332. return ret;
  333. }
  334. static inline int ptr_ring_consume_batched_bh(struct ptr_ring *r,
  335. void **array, int n)
  336. {
  337. int ret;
  338. spin_lock_bh(&r->consumer_lock);
  339. ret = __ptr_ring_consume_batched(r, array, n);
  340. spin_unlock_bh(&r->consumer_lock);
  341. return ret;
  342. }
  343. /* Cast to structure type and call a function without discarding from FIFO.
  344. * Function must return a value.
  345. * Callers must take consumer_lock.
  346. */
  347. #define __PTR_RING_PEEK_CALL(r, f) ((f)(__ptr_ring_peek(r)))
  348. #define PTR_RING_PEEK_CALL(r, f) ({ \
  349. typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
  350. \
  351. spin_lock(&(r)->consumer_lock); \
  352. __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
  353. spin_unlock(&(r)->consumer_lock); \
  354. __PTR_RING_PEEK_CALL_v; \
  355. })
  356. #define PTR_RING_PEEK_CALL_IRQ(r, f) ({ \
  357. typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
  358. \
  359. spin_lock_irq(&(r)->consumer_lock); \
  360. __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
  361. spin_unlock_irq(&(r)->consumer_lock); \
  362. __PTR_RING_PEEK_CALL_v; \
  363. })
  364. #define PTR_RING_PEEK_CALL_BH(r, f) ({ \
  365. typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
  366. \
  367. spin_lock_bh(&(r)->consumer_lock); \
  368. __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
  369. spin_unlock_bh(&(r)->consumer_lock); \
  370. __PTR_RING_PEEK_CALL_v; \
  371. })
  372. #define PTR_RING_PEEK_CALL_ANY(r, f) ({ \
  373. typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
  374. unsigned long __PTR_RING_PEEK_CALL_f;\
  375. \
  376. spin_lock_irqsave(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
  377. __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
  378. spin_unlock_irqrestore(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
  379. __PTR_RING_PEEK_CALL_v; \
  380. })
  381. /* Not all gfp_t flags (besides GFP_KERNEL) are allowed. See
  382. * documentation for vmalloc for which of them are legal.
  383. */
  384. static inline void **__ptr_ring_init_queue_alloc(unsigned int size, gfp_t gfp)
  385. {
  386. if (size > KMALLOC_MAX_SIZE / sizeof(void *))
  387. return NULL;
  388. return kvmalloc_array(size, sizeof(void *), gfp | __GFP_ZERO);
  389. }
  390. static inline void __ptr_ring_set_size(struct ptr_ring *r, int size)
  391. {
  392. r->size = size;
  393. r->batch = SMP_CACHE_BYTES * 2 / sizeof(*(r->queue));
  394. /* We need to set batch at least to 1 to make logic
  395. * in __ptr_ring_discard_one work correctly.
  396. * Batching too much (because ring is small) would cause a lot of
  397. * burstiness. Needs tuning, for now disable batching.
  398. */
  399. if (r->batch > r->size / 2 || !r->batch)
  400. r->batch = 1;
  401. }
  402. static inline int ptr_ring_init(struct ptr_ring *r, int size, gfp_t gfp)
  403. {
  404. r->queue = __ptr_ring_init_queue_alloc(size, gfp);
  405. if (!r->queue)
  406. return -ENOMEM;
  407. __ptr_ring_set_size(r, size);
  408. r->producer = r->consumer_head = r->consumer_tail = 0;
  409. spin_lock_init(&r->producer_lock);
  410. spin_lock_init(&r->consumer_lock);
  411. return 0;
  412. }
  413. /*
  414. * Return entries into ring. Destroy entries that don't fit.
  415. *
  416. * Note: this is expected to be a rare slow path operation.
  417. *
  418. * Note: producer lock is nested within consumer lock, so if you
  419. * resize you must make sure all uses nest correctly.
  420. * In particular if you consume ring in interrupt or BH context, you must
  421. * disable interrupts/BH when doing so.
  422. */
  423. static inline void ptr_ring_unconsume(struct ptr_ring *r, void **batch, int n,
  424. void (*destroy)(void *))
  425. {
  426. unsigned long flags;
  427. int head;
  428. spin_lock_irqsave(&r->consumer_lock, flags);
  429. spin_lock(&r->producer_lock);
  430. if (!r->size)
  431. goto done;
  432. /*
  433. * Clean out buffered entries (for simplicity). This way following code
  434. * can test entries for NULL and if not assume they are valid.
  435. */
  436. head = r->consumer_head - 1;
  437. while (likely(head >= r->consumer_tail))
  438. r->queue[head--] = NULL;
  439. r->consumer_tail = r->consumer_head;
  440. /*
  441. * Go over entries in batch, start moving head back and copy entries.
  442. * Stop when we run into previously unconsumed entries.
  443. */
  444. while (n) {
  445. head = r->consumer_head - 1;
  446. if (head < 0)
  447. head = r->size - 1;
  448. if (r->queue[head]) {
  449. /* This batch entry will have to be destroyed. */
  450. goto done;
  451. }
  452. r->queue[head] = batch[--n];
  453. r->consumer_tail = head;
  454. /* matching READ_ONCE in __ptr_ring_empty for lockless tests */
  455. WRITE_ONCE(r->consumer_head, head);
  456. }
  457. done:
  458. /* Destroy all entries left in the batch. */
  459. while (n)
  460. destroy(batch[--n]);
  461. spin_unlock(&r->producer_lock);
  462. spin_unlock_irqrestore(&r->consumer_lock, flags);
  463. }
  464. static inline void **__ptr_ring_swap_queue(struct ptr_ring *r, void **queue,
  465. int size, gfp_t gfp,
  466. void (*destroy)(void *))
  467. {
  468. int producer = 0;
  469. void **old;
  470. void *ptr;
  471. while ((ptr = __ptr_ring_consume(r)))
  472. if (producer < size)
  473. queue[producer++] = ptr;
  474. else if (destroy)
  475. destroy(ptr);
  476. if (producer >= size)
  477. producer = 0;
  478. __ptr_ring_set_size(r, size);
  479. r->producer = producer;
  480. r->consumer_head = 0;
  481. r->consumer_tail = 0;
  482. old = r->queue;
  483. r->queue = queue;
  484. return old;
  485. }
  486. /*
  487. * Note: producer lock is nested within consumer lock, so if you
  488. * resize you must make sure all uses nest correctly.
  489. * In particular if you consume ring in interrupt or BH context, you must
  490. * disable interrupts/BH when doing so.
  491. */
  492. static inline int ptr_ring_resize(struct ptr_ring *r, int size, gfp_t gfp,
  493. void (*destroy)(void *))
  494. {
  495. unsigned long flags;
  496. void **queue = __ptr_ring_init_queue_alloc(size, gfp);
  497. void **old;
  498. if (!queue)
  499. return -ENOMEM;
  500. spin_lock_irqsave(&(r)->consumer_lock, flags);
  501. spin_lock(&(r)->producer_lock);
  502. old = __ptr_ring_swap_queue(r, queue, size, gfp, destroy);
  503. spin_unlock(&(r)->producer_lock);
  504. spin_unlock_irqrestore(&(r)->consumer_lock, flags);
  505. kvfree(old);
  506. return 0;
  507. }
  508. /*
  509. * Note: producer lock is nested within consumer lock, so if you
  510. * resize you must make sure all uses nest correctly.
  511. * In particular if you consume ring in interrupt or BH context, you must
  512. * disable interrupts/BH when doing so.
  513. */
  514. static inline int ptr_ring_resize_multiple(struct ptr_ring **rings,
  515. unsigned int nrings,
  516. int size,
  517. gfp_t gfp, void (*destroy)(void *))
  518. {
  519. unsigned long flags;
  520. void ***queues;
  521. int i;
  522. queues = kmalloc_array(nrings, sizeof(*queues), gfp);
  523. if (!queues)
  524. goto noqueues;
  525. for (i = 0; i < nrings; ++i) {
  526. queues[i] = __ptr_ring_init_queue_alloc(size, gfp);
  527. if (!queues[i])
  528. goto nomem;
  529. }
  530. for (i = 0; i < nrings; ++i) {
  531. spin_lock_irqsave(&(rings[i])->consumer_lock, flags);
  532. spin_lock(&(rings[i])->producer_lock);
  533. queues[i] = __ptr_ring_swap_queue(rings[i], queues[i],
  534. size, gfp, destroy);
  535. spin_unlock(&(rings[i])->producer_lock);
  536. spin_unlock_irqrestore(&(rings[i])->consumer_lock, flags);
  537. }
  538. for (i = 0; i < nrings; ++i)
  539. kvfree(queues[i]);
  540. kfree(queues);
  541. return 0;
  542. nomem:
  543. while (--i >= 0)
  544. kvfree(queues[i]);
  545. kfree(queues);
  546. noqueues:
  547. return -ENOMEM;
  548. }
  549. static inline void ptr_ring_cleanup(struct ptr_ring *r, void (*destroy)(void *))
  550. {
  551. void *ptr;
  552. if (destroy)
  553. while ((ptr = ptr_ring_consume(r)))
  554. destroy(ptr);
  555. kvfree(r->queue);
  556. }
  557. #endif /* _LINUX_PTR_RING_H */