ck/include/ck_ring.h

37  * Concurrent ring buffer.
57 ck_ring_size(const struct ck_ring *ring)
61 	c = ck_pr_load_uint(&ring->c_head);
62 	p = ck_pr_load_uint(&ring->p_tail);
63 	return (p - c) & ring->mask;
67 ck_ring_capacity(const struct ck_ring *ring)
70 	return ring->size;
75  * on the ring. This is primarily meant for persistent ck_ring use-cases. The
76  * function returns true if any mutations were performed on the ring.
79 ck_ring_repair(struct ck_ring *ring)
83 	if (ring->p_tail != ring->p_head) {
84 		ring->p_tail = ring->p_head;
92  * This can be called when no concurrent updates are occurring on the ring
94  * persistent storage of the ring. If this functions returns false, the ring
98 ck_ring_valid(const struct ck_ring *ring)
100 	unsigned int size = ring->size;
101 	unsigned int c_head = ring->c_head;
102 	unsigned int p_head = ring->p_head;
104 	/* The ring must be a power of 2. */
120 ck_ring_init(struct ck_ring *ring, unsigned int size)
123 	ring->size = size;
124 	ring->mask = size - 1;
125 	ring->p_tail = 0;
126 	ring->p_head = 0;
127 	ring->c_head = 0;
140 _ck_ring_enqueue_reserve_sp(struct ck_ring *ring,
145 	const unsigned int mask = ring->mask;
148 	consumer = ck_pr_load_uint(&ring->c_head);
149 	producer = ring->p_tail;
165 _ck_ring_enqueue_commit_sp(struct ck_ring *ring)
169 	ck_pr_store_uint(&ring->p_tail, ring->p_tail + 1);
174 _ck_ring_enqueue_sp(struct ck_ring *ring,
180 	const unsigned int mask = ring->mask;
183 	consumer = ck_pr_load_uint(&ring->c_head);
184 	producer = ring->p_tail;
200 	ck_pr_store_uint(&ring->p_tail, delta);
205 _ck_ring_enqueue_sp_size(struct ck_ring *ring,
214 	r = _ck_ring_enqueue_sp(ring, buffer, entry, ts, &sz);
220 _ck_ring_dequeue_sc(struct ck_ring *ring,
225 	const unsigned int mask = ring->mask;
228 	consumer = ring->c_head;
229 	producer = ck_pr_load_uint(&ring->p_tail);
248 	ck_pr_store_uint(&ring->c_head, consumer + 1);
253 _ck_ring_enqueue_reserve_mp(struct ck_ring *ring,
259 	const unsigned int mask = ring->mask;
262 	producer = ck_pr_load_uint(&ring->p_head);
266 		consumer = ck_pr_load_uint(&ring->c_head);
271 			if (ck_pr_cas_uint_value(&ring->p_head,
279 			new_producer = ck_pr_load_uint(&ring->p_head);
300 _ck_ring_enqueue_commit_mp(struct ck_ring *ring, unsigned int producer)
303 	while (ck_pr_load_uint(&ring->p_tail) != producer)
307 	ck_pr_store_uint(&ring->p_tail, producer + 1);
312 _ck_ring_enqueue_mp(struct ck_ring *ring,
318 	const unsigned int mask = ring->mask;
322 	producer = ck_pr_load_uint(&ring->p_head);
330 		consumer = ck_pr_load_uint(&ring->c_head);
339 			if (ck_pr_cas_uint_value(&ring->p_head,
353 			new_producer = ck_pr_load_uint(&ring->p_head);
378 	 * their data into the ring buffer.
380 	while (ck_pr_load_uint(&ring->p_tail) != producer)
388 	ck_pr_store_uint(&ring->p_tail, delta);
398 _ck_ring_enqueue_mp_size(struct ck_ring *ring,
407 	r = _ck_ring_enqueue_mp(ring, buffer, entry, ts, &sz);
413 _ck_ring_trydequeue_mc(struct ck_ring *ring,
418 	const unsigned int mask = ring->mask;
421 	consumer = ck_pr_load_uint(&ring->c_head);
423 	producer = ck_pr_load_uint(&ring->p_tail);
434 	return ck_pr_cas_uint(&ring->c_head, consumer, consumer + 1);
438 _ck_ring_dequeue_mc(struct ck_ring *ring,
443 	const unsigned int mask = ring->mask;
446 	consumer = ck_pr_load_uint(&ring->c_head);
456 		producer = ck_pr_load_uint(&ring->p_tail);
468 	} while (ck_pr_cas_uint_value(&ring->c_head,
478  * ring buffer containing pointers. Correctness is only provided if there is up
482 ck_ring_enqueue_spsc_size(struct ck_ring *ring,
488 	return _ck_ring_enqueue_sp_size(ring, buffer, &entry,
493 ck_ring_enqueue_spsc(struct ck_ring *ring,
498 	return _ck_ring_enqueue_sp(ring, buffer,
503 ck_ring_enqueue_reserve_spsc_size(struct ck_ring *ring,
508 	return _ck_ring_enqueue_reserve_sp(ring, buffer, sizeof(void *),
513 ck_ring_enqueue_reserve_spsc(struct ck_ring *ring,
517 	return _ck_ring_enqueue_reserve_sp(ring, buffer, sizeof(void *),
522 ck_ring_enqueue_commit_spsc(struct ck_ring *ring)
525 	_ck_ring_enqueue_commit_sp(ring);
530 ck_ring_dequeue_spsc(struct ck_ring *ring,
535 	return _ck_ring_dequeue_sc(ring, buffer,
541  * ring buffer containing pointers. Correctness is provided for any number of
545 ck_ring_enqueue_mpmc(struct ck_ring *ring,
550 	return _ck_ring_enqueue_mp(ring, buffer, &entry, sizeof(entry), NULL);
554 ck_ring_enqueue_mpmc_size(struct ck_ring *ring,
560 	return _ck_ring_enqueue_mp_size(ring, buffer, &entry, sizeof(entry),
565 ck_ring_enqueue_reserve_mpmc(struct ck_ring *ring,
570 	return _ck_ring_enqueue_reserve_mp(ring, buffer, sizeof(void *),
575 ck_ring_enqueue_reserve_mpmc_size(struct ck_ring *ring,
581 	return _ck_ring_enqueue_reserve_mp(ring, buffer, sizeof(void *),
586 ck_ring_enqueue_commit_mpmc(struct ck_ring *ring, unsigned int ticket)
589 	_ck_ring_enqueue_commit_mp(ring, ticket);
594 ck_ring_trydequeue_mpmc(struct ck_ring *ring,
599 	return _ck_ring_trydequeue_mc(ring,
604 ck_ring_dequeue_mpmc(struct ck_ring *ring,
609 	return _ck_ring_dequeue_mc(ring, buffer, (void **)data,
615  * ring buffer containing pointers. Correctness is provided for any number of
619 ck_ring_enqueue_reserve_spmc_size(struct ck_ring *ring,
624 	return _ck_ring_enqueue_reserve_sp(ring, buffer, sizeof(void *), size);
628 ck_ring_enqueue_reserve_spmc(struct ck_ring *ring,
632 	return _ck_ring_enqueue_reserve_sp(ring, buffer, sizeof(void *), NULL);
636 ck_ring_enqueue_commit_spmc(struct ck_ring *ring)
639 	_ck_ring_enqueue_commit_sp(ring);
644 ck_ring_enqueue_spmc_size(struct ck_ring *ring,
650 	return _ck_ring_enqueue_sp_size(ring, buffer, &entry,
655 ck_ring_enqueue_spmc(struct ck_ring *ring,
660 	return _ck_ring_enqueue_sp(ring, buffer, &entry,
665 ck_ring_trydequeue_spmc(struct ck_ring *ring,
670 	return _ck_ring_trydequeue_mc(ring, buffer, (void **)data, sizeof(void *));
674 ck_ring_dequeue_spmc(struct ck_ring *ring,
679 	return _ck_ring_dequeue_mc(ring, buffer, (void **)data, sizeof(void *));
684  * ring buffer containing pointers. Correctness is provided for any number of
688 ck_ring_enqueue_reserve_mpsc(struct ck_ring *ring,
693 	return _ck_ring_enqueue_reserve_mp(ring, buffer, sizeof(void *),
698 ck_ring_enqueue_reserve_mpsc_size(struct ck_ring *ring,
704 	return _ck_ring_enqueue_reserve_mp(ring, buffer, sizeof(void *),
709 ck_ring_enqueue_commit_mpsc(struct ck_ring *ring, unsigned int ticket)
712 	_ck_ring_enqueue_commit_mp(ring, ticket);
717 ck_ring_enqueue_mpsc(struct ck_ring *ring,
722 	return _ck_ring_enqueue_mp(ring, buffer, &entry,
727 ck_ring_enqueue_mpsc_size(struct ck_ring *ring,
733 	return _ck_ring_enqueue_mp_size(ring, buffer, &entry,
738 ck_ring_dequeue_mpsc(struct ck_ring *ring,
743 	return _ck_ring_dequeue_sc(ring, buffer, (void **)data,
749  * values of a particular type in the ring the buffer.