1 // SPDX-License-Identifier: GPL-2.0
3 #include <linux/kernel.h>
4 #include <linux/irqflags.h>
5 #include <linux/string.h>
6 #include <linux/errno.h>
8 #include "printk_ringbuffer.h"
11 * DOC: printk_ringbuffer overview
15 * The printk_ringbuffer is made up of 3 internal ringbuffers:
18 * A ring of descriptors and their meta data (such as sequence number,
19 * timestamp, loglevel, etc.) as well as internal state information about
20 * the record and logical positions specifying where in the other
21 * ringbuffer the text strings are located.
24 * A ring of data blocks. A data block consists of an unsigned long
25 * integer (ID) that maps to a desc_ring index followed by the text
26 * string of the record.
28 * The internal state information of a descriptor is the key element to allow
29 * readers and writers to locklessly synchronize access to the data.
36 * The descriptor ring is an array of descriptors. A descriptor contains
37 * essential meta data to track the data of a printk record using
38 * blk_lpos structs pointing to associated text data blocks (see
39 * "Data Rings" below). Each descriptor is assigned an ID that maps
40 * directly to index values of the descriptor array and has a state. The ID
41 * and the state are bitwise combined into a single descriptor field named
42 * @state_var, allowing ID and state to be synchronously and atomically
45 * Descriptors have four states:
48 * A writer is modifying the record.
51 * The record and all its data are written. A writer can reopen the
52 * descriptor (transitioning it back to reserved), but in the committed
53 * state the data is consistent.
56 * The record and all its data are complete and available for reading. A
57 * writer cannot reopen the descriptor.
60 * The record exists, but its text and/or meta data may no longer be
63 * Querying the @state_var of a record requires providing the ID of the
64 * descriptor to query. This can yield a possible fifth (pseudo) state:
67 * The descriptor being queried has an unexpected ID.
69 * The descriptor ring has a @tail_id that contains the ID of the oldest
70 * descriptor and @head_id that contains the ID of the newest descriptor.
72 * When a new descriptor should be created (and the ring is full), the tail
73 * descriptor is invalidated by first transitioning to the reusable state and
74 * then invalidating all tail data blocks up to and including the data blocks
75 * associated with the tail descriptor (for the text ring). Then
76 * @tail_id is advanced, followed by advancing @head_id. And finally the
77 * @state_var of the new descriptor is initialized to the new ID and reserved
80 * The @tail_id can only be advanced if the new @tail_id would be in the
81 * committed or reusable queried state. This makes it possible that a valid
82 * sequence number of the tail is always available.
84 * Descriptor Finalization
85 * ~~~~~~~~~~~~~~~~~~~~~~~
86 * When a writer calls the commit function prb_commit(), record data is
87 * fully stored and is consistent within the ringbuffer. However, a writer can
88 * reopen that record, claiming exclusive access (as with prb_reserve()), and
89 * modify that record. When finished, the writer must again commit the record.
91 * In order for a record to be made available to readers (and also become
92 * recyclable for writers), it must be finalized. A finalized record cannot be
93 * reopened and can never become "unfinalized". Record finalization can occur
94 * in three different scenarios:
96 * 1) A writer can simultaneously commit and finalize its record by calling
97 * prb_final_commit() instead of prb_commit().
99 * 2) When a new record is reserved and the previous record has been
100 * committed via prb_commit(), that previous record is automatically
103 * 3) When a record is committed via prb_commit() and a newer record
104 * already exists, the record being committed is automatically finalized.
108 * The text data ring is a byte array composed of data blocks. Data blocks are
109 * referenced by blk_lpos structs that point to the logical position of the
110 * beginning of a data block and the beginning of the next adjacent data
111 * block. Logical positions are mapped directly to index values of the byte
114 * Each data block consists of an ID followed by the writer data. The ID is
115 * the identifier of a descriptor that is associated with the data block. A
116 * given data block is considered valid if all of the following conditions
119 * 1) The descriptor associated with the data block is in the committed
120 * or finalized queried state.
122 * 2) The blk_lpos struct within the descriptor associated with the data
123 * block references back to the same data block.
125 * 3) The data block is within the head/tail logical position range.
127 * If the writer data of a data block would extend beyond the end of the
128 * byte array, only the ID of the data block is stored at the logical
129 * position and the full data block (ID and writer data) is stored at the
130 * beginning of the byte array. The referencing blk_lpos will point to the
131 * ID before the wrap and the next data block will be at the logical
132 * position adjacent the full data block after the wrap.
134 * Data rings have a @tail_lpos that points to the beginning of the oldest
135 * data block and a @head_lpos that points to the logical position of the
136 * next (not yet existing) data block.
138 * When a new data block should be created (and the ring is full), tail data
139 * blocks will first be invalidated by putting their associated descriptors
140 * into the reusable state and then pushing the @tail_lpos forward beyond
141 * them. Then the @head_lpos is pushed forward and is associated with a new
142 * descriptor. If a data block is not valid, the @tail_lpos cannot be
143 * advanced beyond it.
147 * The general meta data of printk records are stored in printk_info structs,
148 * stored in an array with the same number of elements as the descriptor ring.
149 * Each info corresponds to the descriptor of the same index in the
150 * descriptor ring. Info validity is confirmed by evaluating the corresponding
151 * descriptor before and after loading the info.
155 * Here are some simple examples demonstrating writers and readers. For the
156 * examples a global ringbuffer (test_rb) is available (which is not the
157 * actual ringbuffer used by printk)::
159 * DEFINE_PRINTKRB(test_rb, 15, 5);
161 * This ringbuffer allows up to 32768 records (2 ^ 15) and has a size of
162 * 1 MiB (2 ^ (15 + 5)) for text data.
164 * Sample writer code::
166 * const char *textstr = "message text";
167 * struct prb_reserved_entry e;
168 * struct printk_record r;
170 * // specify how much to allocate
171 * prb_rec_init_wr(&r, strlen(textstr) + 1);
173 * if (prb_reserve(&e, &test_rb, &r)) {
174 * snprintf(r.text_buf, r.text_buf_size, "%s", textstr);
176 * r.info->text_len = strlen(textstr);
177 * r.info->ts_nsec = local_clock();
178 * r.info->caller_id = printk_caller_id();
180 * // commit and finalize the record
181 * prb_final_commit(&e);
184 * Note that additional writer functions are available to extend a record
185 * after it has been committed but not yet finalized. This can be done as
186 * long as no new records have been reserved and the caller is the same.
188 * Sample writer code (record extending)::
190 * // alternate rest of previous example
192 * r.info->text_len = strlen(textstr);
193 * r.info->ts_nsec = local_clock();
194 * r.info->caller_id = printk_caller_id();
196 * // commit the record (but do not finalize yet)
202 * // specify additional 5 bytes text space to extend
203 * prb_rec_init_wr(&r, 5);
205 * // try to extend, but only if it does not exceed 32 bytes
206 * if (prb_reserve_in_last(&e, &test_rb, &r, printk_caller_id()), 32) {
207 * snprintf(&r.text_buf[r.info->text_len],
208 * r.text_buf_size - r.info->text_len, "hello");
210 * r.info->text_len += 5;
212 * // commit and finalize the record
213 * prb_final_commit(&e);
216 * Sample reader code::
218 * struct printk_info info;
219 * struct printk_record r;
223 * prb_rec_init_rd(&r, &info, &text_buf[0], sizeof(text_buf));
225 * prb_for_each_record(0, &test_rb, &seq, &r) {
226 * if (info.seq != seq)
227 * pr_warn("lost %llu records\n", info.seq - seq);
229 * if (info.text_len > r.text_buf_size) {
230 * pr_warn("record %llu text truncated\n", info.seq);
231 * text_buf[r.text_buf_size - 1] = 0;
234 * pr_info("%llu: %llu: %s\n", info.seq, info.ts_nsec,
238 * Note that additional less convenient reader functions are available to
239 * allow complex record access.
243 * To help avoid ABA issues, descriptors are referenced by IDs (array index
244 * values combined with tagged bits counting array wraps) and data blocks are
245 * referenced by logical positions (array index values combined with tagged
246 * bits counting array wraps). However, on 32-bit systems the number of
247 * tagged bits is relatively small such that an ABA incident is (at least
248 * theoretically) possible. For example, if 4 million maximally sized (1KiB)
249 * printk messages were to occur in NMI context on a 32-bit system, the
250 * interrupted context would not be able to recognize that the 32-bit integer
251 * completely wrapped and thus represents a different data block than the one
252 * the interrupted context expects.
254 * To help combat this possibility, additional state checking is performed
255 * (such as using cmpxchg() even though set() would suffice). These extra
256 * checks are commented as such and will hopefully catch any ABA issue that
257 * a 32-bit system might experience.
261 * Multiple memory barriers are used. To simplify proving correctness and
262 * generating litmus tests, lines of code related to memory barriers
263 * (loads, stores, and the associated memory barriers) are labeled::
265 * LMM(function:letter)
267 * Comments reference the labels using only the "function:letter" part.
269 * The memory barrier pairs and their ordering are:
271 * desc_reserve:D / desc_reserve:B
272 * push descriptor tail (id), then push descriptor head (id)
274 * desc_reserve:D / data_push_tail:B
275 * push data tail (lpos), then set new descriptor reserved (state)
277 * desc_reserve:D / desc_push_tail:C
278 * push descriptor tail (id), then set new descriptor reserved (state)
280 * desc_reserve:D / prb_first_seq:C
281 * push descriptor tail (id), then set new descriptor reserved (state)
283 * desc_reserve:F / desc_read:D
284 * set new descriptor id and reserved (state), then allow writer changes
286 * data_alloc:A (or data_realloc:A) / desc_read:D
287 * set old descriptor reusable (state), then modify new data block area
289 * data_alloc:A (or data_realloc:A) / data_push_tail:B
290 * push data tail (lpos), then modify new data block area
292 * _prb_commit:B / desc_read:B
293 * store writer changes, then set new descriptor committed (state)
295 * desc_reopen_last:A / _prb_commit:B
296 * set descriptor reserved (state), then read descriptor data
298 * _prb_commit:B / desc_reserve:D
299 * set new descriptor committed (state), then check descriptor head (id)
301 * data_push_tail:D / data_push_tail:A
302 * set descriptor reusable (state), then push data tail (lpos)
304 * desc_push_tail:B / desc_reserve:D
305 * set descriptor reusable (state), then push descriptor tail (id)
308 #define DATA_SIZE(data_ring) _DATA_SIZE((data_ring)->size_bits)
309 #define DATA_SIZE_MASK(data_ring) (DATA_SIZE(data_ring) - 1)
311 #define DESCS_COUNT(desc_ring) _DESCS_COUNT((desc_ring)->count_bits)
312 #define DESCS_COUNT_MASK(desc_ring) (DESCS_COUNT(desc_ring) - 1)
314 /* Determine the data array index from a logical position. */
315 #define DATA_INDEX(data_ring, lpos) ((lpos) & DATA_SIZE_MASK(data_ring))
317 /* Determine the desc array index from an ID or sequence number. */
318 #define DESC_INDEX(desc_ring, n) ((n) & DESCS_COUNT_MASK(desc_ring))
320 /* Determine how many times the data array has wrapped. */
321 #define DATA_WRAPS(data_ring, lpos) ((lpos) >> (data_ring)->size_bits)
323 /* Determine if a logical position refers to a data-less block. */
324 #define LPOS_DATALESS(lpos) ((lpos) & 1UL)
325 #define BLK_DATALESS(blk) (LPOS_DATALESS((blk)->begin) && \
326 LPOS_DATALESS((blk)->next))
328 /* Get the logical position at index 0 of the current wrap. */
329 #define DATA_THIS_WRAP_START_LPOS(data_ring, lpos) \
330 ((lpos) & ~DATA_SIZE_MASK(data_ring))
332 /* Get the ID for the same index of the previous wrap as the given ID. */
333 #define DESC_ID_PREV_WRAP(desc_ring, id) \
334 DESC_ID((id) - DESCS_COUNT(desc_ring))
337 * A data block: mapped directly to the beginning of the data block area
338 * specified as a logical position within the data ring.
340 * @id: the ID of the associated descriptor
341 * @data: the writer data
343 * Note that the size of a data block is only known by its associated
346 struct prb_data_block
{
352 * Return the descriptor associated with @n. @n can be either a
353 * descriptor ID or a sequence number.
355 static struct prb_desc
*to_desc(struct prb_desc_ring
*desc_ring
, u64 n
)
357 return &desc_ring
->descs
[DESC_INDEX(desc_ring
, n
)];
361 * Return the printk_info associated with @n. @n can be either a
362 * descriptor ID or a sequence number.
364 static struct printk_info
*to_info(struct prb_desc_ring
*desc_ring
, u64 n
)
366 return &desc_ring
->infos
[DESC_INDEX(desc_ring
, n
)];
369 static struct prb_data_block
*to_block(struct prb_data_ring
*data_ring
,
370 unsigned long begin_lpos
)
372 return (void *)&data_ring
->data
[DATA_INDEX(data_ring
, begin_lpos
)];
376 * Increase the data size to account for data block meta data plus any
377 * padding so that the adjacent data block is aligned on the ID size.
379 static unsigned int to_blk_size(unsigned int size
)
381 struct prb_data_block
*db
= NULL
;
384 size
= ALIGN(size
, sizeof(db
->id
));
389 * Sanity checker for reserve size. The ringbuffer code assumes that a data
390 * block does not exceed the maximum possible size that could fit within the
391 * ringbuffer. This function provides that basic size check so that the
392 * assumption is safe.
394 static bool data_check_size(struct prb_data_ring
*data_ring
, unsigned int size
)
396 struct prb_data_block
*db
= NULL
;
402 * Ensure the alignment padded size could possibly fit in the data
403 * array. The largest possible data block must still leave room for
404 * at least the ID of the next block.
406 size
= to_blk_size(size
);
407 if (size
> DATA_SIZE(data_ring
) - sizeof(db
->id
))
413 /* Query the state of a descriptor. */
414 static enum desc_state
get_desc_state(unsigned long id
,
415 unsigned long state_val
)
417 if (id
!= DESC_ID(state_val
))
420 return DESC_STATE(state_val
);
424 * Get a copy of a specified descriptor and return its queried state. If the
425 * descriptor is in an inconsistent state (miss or reserved), the caller can
426 * only expect the descriptor's @state_var field to be valid.
428 * The sequence number and caller_id can be optionally retrieved. Like all
429 * non-state_var data, they are only valid if the descriptor is in a
432 static enum desc_state
desc_read(struct prb_desc_ring
*desc_ring
,
433 unsigned long id
, struct prb_desc
*desc_out
,
434 u64
*seq_out
, u32
*caller_id_out
)
436 struct printk_info
*info
= to_info(desc_ring
, id
);
437 struct prb_desc
*desc
= to_desc(desc_ring
, id
);
438 atomic_long_t
*state_var
= &desc
->state_var
;
439 enum desc_state d_state
;
440 unsigned long state_val
;
442 /* Check the descriptor state. */
443 state_val
= atomic_long_read(state_var
); /* LMM(desc_read:A) */
444 d_state
= get_desc_state(id
, state_val
);
445 if (d_state
== desc_miss
|| d_state
== desc_reserved
) {
447 * The descriptor is in an inconsistent state. Set at least
448 * @state_var so that the caller can see the details of
449 * the inconsistent state.
455 * Guarantee the state is loaded before copying the descriptor
456 * content. This avoids copying obsolete descriptor content that might
457 * not apply to the descriptor state. This pairs with _prb_commit:B.
459 * Memory barrier involvement:
461 * If desc_read:A reads from _prb_commit:B, then desc_read:C reads
462 * from _prb_commit:A.
466 * WMB from _prb_commit:A to _prb_commit:B
468 * RMB from desc_read:A to desc_read:C
470 smp_rmb(); /* LMM(desc_read:B) */
473 * Copy the descriptor data. The data is not valid until the
474 * state has been re-checked. A memcpy() for all of @desc
475 * cannot be used because of the atomic_t @state_var field.
477 memcpy(&desc_out
->text_blk_lpos
, &desc
->text_blk_lpos
,
478 sizeof(desc_out
->text_blk_lpos
)); /* LMM(desc_read:C) */
480 *seq_out
= info
->seq
; /* also part of desc_read:C */
482 *caller_id_out
= info
->caller_id
; /* also part of desc_read:C */
485 * 1. Guarantee the descriptor content is loaded before re-checking
486 * the state. This avoids reading an obsolete descriptor state
487 * that may not apply to the copied content. This pairs with
490 * Memory barrier involvement:
492 * If desc_read:C reads from desc_reserve:G, then desc_read:E
493 * reads from desc_reserve:F.
497 * WMB from desc_reserve:F to desc_reserve:G
499 * RMB from desc_read:C to desc_read:E
501 * 2. Guarantee the record data is loaded before re-checking the
502 * state. This avoids reading an obsolete descriptor state that may
503 * not apply to the copied data. This pairs with data_alloc:A and
506 * Memory barrier involvement:
508 * If copy_data:A reads from data_alloc:B, then desc_read:E
509 * reads from desc_make_reusable:A.
513 * MB from desc_make_reusable:A to data_alloc:B
515 * RMB from desc_read:C to desc_read:E
517 * Note: desc_make_reusable:A and data_alloc:B can be different
518 * CPUs. However, the data_alloc:B CPU (which performs the
519 * full memory barrier) must have previously seen
520 * desc_make_reusable:A.
522 smp_rmb(); /* LMM(desc_read:D) */
525 * The data has been copied. Return the current descriptor state,
526 * which may have changed since the load above.
528 state_val
= atomic_long_read(state_var
); /* LMM(desc_read:E) */
529 d_state
= get_desc_state(id
, state_val
);
531 atomic_long_set(&desc_out
->state_var
, state_val
);
536 * Take a specified descriptor out of the finalized state by attempting
537 * the transition from finalized to reusable. Either this context or some
538 * other context will have been successful.
540 static void desc_make_reusable(struct prb_desc_ring
*desc_ring
,
543 unsigned long val_finalized
= DESC_SV(id
, desc_finalized
);
544 unsigned long val_reusable
= DESC_SV(id
, desc_reusable
);
545 struct prb_desc
*desc
= to_desc(desc_ring
, id
);
546 atomic_long_t
*state_var
= &desc
->state_var
;
548 atomic_long_cmpxchg_relaxed(state_var
, val_finalized
,
549 val_reusable
); /* LMM(desc_make_reusable:A) */
553 * Given the text data ring, put the associated descriptor of each
554 * data block from @lpos_begin until @lpos_end into the reusable state.
556 * If there is any problem making the associated descriptor reusable, either
557 * the descriptor has not yet been finalized or another writer context has
558 * already pushed the tail lpos past the problematic data block. Regardless,
559 * on error the caller can re-load the tail lpos to determine the situation.
561 static bool data_make_reusable(struct printk_ringbuffer
*rb
,
562 unsigned long lpos_begin
,
563 unsigned long lpos_end
,
564 unsigned long *lpos_out
)
567 struct prb_data_ring
*data_ring
= &rb
->text_data_ring
;
568 struct prb_desc_ring
*desc_ring
= &rb
->desc_ring
;
569 struct prb_data_block
*blk
;
570 enum desc_state d_state
;
571 struct prb_desc desc
;
572 struct prb_data_blk_lpos
*blk_lpos
= &desc
.text_blk_lpos
;
575 /* Loop until @lpos_begin has advanced to or beyond @lpos_end. */
576 while ((lpos_end
- lpos_begin
) - 1 < DATA_SIZE(data_ring
)) {
577 blk
= to_block(data_ring
, lpos_begin
);
580 * Load the block ID from the data block. This is a data race
581 * against a writer that may have newly reserved this data
582 * area. If the loaded value matches a valid descriptor ID,
583 * the blk_lpos of that descriptor will be checked to make
584 * sure it points back to this data block. If the check fails,
585 * the data area has been recycled by another writer.
587 id
= blk
->id
; /* LMM(data_make_reusable:A) */
589 d_state
= desc_read(desc_ring
, id
, &desc
,
590 NULL
, NULL
); /* LMM(data_make_reusable:B) */
599 * This data block is invalid if the descriptor
600 * does not point back to it.
602 if (blk_lpos
->begin
!= lpos_begin
)
604 desc_make_reusable(desc_ring
, id
);
608 * This data block is invalid if the descriptor
609 * does not point back to it.
611 if (blk_lpos
->begin
!= lpos_begin
)
616 /* Advance @lpos_begin to the next data block. */
617 lpos_begin
= blk_lpos
->next
;
620 *lpos_out
= lpos_begin
;
625 * Advance the data ring tail to at least @lpos. This function puts
626 * descriptors into the reusable state if the tail is pushed beyond
627 * their associated data block.
629 static bool data_push_tail(struct printk_ringbuffer
*rb
, unsigned long lpos
)
631 struct prb_data_ring
*data_ring
= &rb
->text_data_ring
;
632 unsigned long tail_lpos_new
;
633 unsigned long tail_lpos
;
634 unsigned long next_lpos
;
636 /* If @lpos is from a data-less block, there is nothing to do. */
637 if (LPOS_DATALESS(lpos
))
641 * Any descriptor states that have transitioned to reusable due to the
642 * data tail being pushed to this loaded value will be visible to this
643 * CPU. This pairs with data_push_tail:D.
645 * Memory barrier involvement:
647 * If data_push_tail:A reads from data_push_tail:D, then this CPU can
648 * see desc_make_reusable:A.
652 * MB from desc_make_reusable:A to data_push_tail:D
654 * READFROM from data_push_tail:D to data_push_tail:A
656 * READFROM from desc_make_reusable:A to this CPU
658 tail_lpos
= atomic_long_read(&data_ring
->tail_lpos
); /* LMM(data_push_tail:A) */
661 * Loop until the tail lpos is at or beyond @lpos. This condition
662 * may already be satisfied, resulting in no full memory barrier
663 * from data_push_tail:D being performed. However, since this CPU
664 * sees the new tail lpos, any descriptor states that transitioned to
665 * the reusable state must already be visible.
667 while ((lpos
- tail_lpos
) - 1 < DATA_SIZE(data_ring
)) {
669 * Make all descriptors reusable that are associated with
670 * data blocks before @lpos.
672 if (!data_make_reusable(rb
, tail_lpos
, lpos
, &next_lpos
)) {
674 * 1. Guarantee the block ID loaded in
675 * data_make_reusable() is performed before
676 * reloading the tail lpos. The failed
677 * data_make_reusable() may be due to a newly
678 * recycled data area causing the tail lpos to
679 * have been previously pushed. This pairs with
680 * data_alloc:A and data_realloc:A.
682 * Memory barrier involvement:
684 * If data_make_reusable:A reads from data_alloc:B,
685 * then data_push_tail:C reads from
690 * MB from data_push_tail:D to data_alloc:B
692 * RMB from data_make_reusable:A to
695 * Note: data_push_tail:D and data_alloc:B can be
696 * different CPUs. However, the data_alloc:B
697 * CPU (which performs the full memory
698 * barrier) must have previously seen
701 * 2. Guarantee the descriptor state loaded in
702 * data_make_reusable() is performed before
703 * reloading the tail lpos. The failed
704 * data_make_reusable() may be due to a newly
705 * recycled descriptor causing the tail lpos to
706 * have been previously pushed. This pairs with
709 * Memory barrier involvement:
711 * If data_make_reusable:B reads from
712 * desc_reserve:F, then data_push_tail:C reads
713 * from data_push_tail:D.
717 * MB from data_push_tail:D to desc_reserve:F
719 * RMB from data_make_reusable:B to
722 * Note: data_push_tail:D and desc_reserve:F can
723 * be different CPUs. However, the
724 * desc_reserve:F CPU (which performs the
725 * full memory barrier) must have previously
726 * seen data_push_tail:D.
728 smp_rmb(); /* LMM(data_push_tail:B) */
730 tail_lpos_new
= atomic_long_read(&data_ring
->tail_lpos
731 ); /* LMM(data_push_tail:C) */
732 if (tail_lpos_new
== tail_lpos
)
735 /* Another CPU pushed the tail. Try again. */
736 tail_lpos
= tail_lpos_new
;
741 * Guarantee any descriptor states that have transitioned to
742 * reusable are stored before pushing the tail lpos. A full
743 * memory barrier is needed since other CPUs may have made
744 * the descriptor states reusable. This pairs with
747 if (atomic_long_try_cmpxchg(&data_ring
->tail_lpos
, &tail_lpos
,
748 next_lpos
)) { /* LMM(data_push_tail:D) */
757 * Advance the desc ring tail. This function advances the tail by one
758 * descriptor, thus invalidating the oldest descriptor. Before advancing
759 * the tail, the tail descriptor is made reusable and all data blocks up to
760 * and including the descriptor's data block are invalidated (i.e. the data
761 * ring tail is pushed past the data block of the descriptor being made
764 static bool desc_push_tail(struct printk_ringbuffer
*rb
,
765 unsigned long tail_id
)
767 struct prb_desc_ring
*desc_ring
= &rb
->desc_ring
;
768 enum desc_state d_state
;
769 struct prb_desc desc
;
771 d_state
= desc_read(desc_ring
, tail_id
, &desc
, NULL
, NULL
);
776 * If the ID is exactly 1 wrap behind the expected, it is
777 * in the process of being reserved by another writer and
778 * must be considered reserved.
780 if (DESC_ID(atomic_long_read(&desc
.state_var
)) ==
781 DESC_ID_PREV_WRAP(desc_ring
, tail_id
)) {
786 * The ID has changed. Another writer must have pushed the
787 * tail and recycled the descriptor already. Success is
788 * returned because the caller is only interested in the
789 * specified tail being pushed, which it was.
796 desc_make_reusable(desc_ring
, tail_id
);
803 * Data blocks must be invalidated before their associated
804 * descriptor can be made available for recycling. Invalidating
805 * them later is not possible because there is no way to trust
806 * data blocks once their associated descriptor is gone.
809 if (!data_push_tail(rb
, desc
.text_blk_lpos
.next
))
813 * Check the next descriptor after @tail_id before pushing the tail
814 * to it because the tail must always be in a finalized or reusable
815 * state. The implementation of prb_first_seq() relies on this.
817 * A successful read implies that the next descriptor is less than or
818 * equal to @head_id so there is no risk of pushing the tail past the
821 d_state
= desc_read(desc_ring
, DESC_ID(tail_id
+ 1), &desc
,
822 NULL
, NULL
); /* LMM(desc_push_tail:A) */
824 if (d_state
== desc_finalized
|| d_state
== desc_reusable
) {
826 * Guarantee any descriptor states that have transitioned to
827 * reusable are stored before pushing the tail ID. This allows
828 * verifying the recycled descriptor state. A full memory
829 * barrier is needed since other CPUs may have made the
830 * descriptor states reusable. This pairs with desc_reserve:D.
832 atomic_long_cmpxchg(&desc_ring
->tail_id
, tail_id
,
833 DESC_ID(tail_id
+ 1)); /* LMM(desc_push_tail:B) */
836 * Guarantee the last state load from desc_read() is before
837 * reloading @tail_id in order to see a new tail ID in the
838 * case that the descriptor has been recycled. This pairs
839 * with desc_reserve:D.
841 * Memory barrier involvement:
843 * If desc_push_tail:A reads from desc_reserve:F, then
844 * desc_push_tail:D reads from desc_push_tail:B.
848 * MB from desc_push_tail:B to desc_reserve:F
850 * RMB from desc_push_tail:A to desc_push_tail:D
852 * Note: desc_push_tail:B and desc_reserve:F can be different
853 * CPUs. However, the desc_reserve:F CPU (which performs
854 * the full memory barrier) must have previously seen
857 smp_rmb(); /* LMM(desc_push_tail:C) */
860 * Re-check the tail ID. The descriptor following @tail_id is
861 * not in an allowed tail state. But if the tail has since
862 * been moved by another CPU, then it does not matter.
864 if (atomic_long_read(&desc_ring
->tail_id
) == tail_id
) /* LMM(desc_push_tail:D) */
871 /* Reserve a new descriptor, invalidating the oldest if necessary. */
872 static bool desc_reserve(struct printk_ringbuffer
*rb
, unsigned long *id_out
)
874 struct prb_desc_ring
*desc_ring
= &rb
->desc_ring
;
875 unsigned long prev_state_val
;
876 unsigned long id_prev_wrap
;
877 struct prb_desc
*desc
;
878 unsigned long head_id
;
881 head_id
= atomic_long_read(&desc_ring
->head_id
); /* LMM(desc_reserve:A) */
884 id
= DESC_ID(head_id
+ 1);
885 id_prev_wrap
= DESC_ID_PREV_WRAP(desc_ring
, id
);
888 * Guarantee the head ID is read before reading the tail ID.
889 * Since the tail ID is updated before the head ID, this
890 * guarantees that @id_prev_wrap is never ahead of the tail
891 * ID. This pairs with desc_reserve:D.
893 * Memory barrier involvement:
895 * If desc_reserve:A reads from desc_reserve:D, then
896 * desc_reserve:C reads from desc_push_tail:B.
900 * MB from desc_push_tail:B to desc_reserve:D
902 * RMB from desc_reserve:A to desc_reserve:C
904 * Note: desc_push_tail:B and desc_reserve:D can be different
905 * CPUs. However, the desc_reserve:D CPU (which performs
906 * the full memory barrier) must have previously seen
909 smp_rmb(); /* LMM(desc_reserve:B) */
911 if (id_prev_wrap
== atomic_long_read(&desc_ring
->tail_id
912 )) { /* LMM(desc_reserve:C) */
914 * Make space for the new descriptor by
915 * advancing the tail.
917 if (!desc_push_tail(rb
, id_prev_wrap
))
922 * 1. Guarantee the tail ID is read before validating the
923 * recycled descriptor state. A read memory barrier is
924 * sufficient for this. This pairs with desc_push_tail:B.
926 * Memory barrier involvement:
928 * If desc_reserve:C reads from desc_push_tail:B, then
929 * desc_reserve:E reads from desc_make_reusable:A.
933 * MB from desc_make_reusable:A to desc_push_tail:B
935 * RMB from desc_reserve:C to desc_reserve:E
937 * Note: desc_make_reusable:A and desc_push_tail:B can be
938 * different CPUs. However, the desc_push_tail:B CPU
939 * (which performs the full memory barrier) must have
940 * previously seen desc_make_reusable:A.
942 * 2. Guarantee the tail ID is stored before storing the head
943 * ID. This pairs with desc_reserve:B.
945 * 3. Guarantee any data ring tail changes are stored before
946 * recycling the descriptor. Data ring tail changes can
947 * happen via desc_push_tail()->data_push_tail(). A full
948 * memory barrier is needed since another CPU may have
949 * pushed the data ring tails. This pairs with
952 * 4. Guarantee a new tail ID is stored before recycling the
953 * descriptor. A full memory barrier is needed since
954 * another CPU may have pushed the tail ID. This pairs
955 * with desc_push_tail:C and this also pairs with
958 * 5. Guarantee the head ID is stored before trying to
959 * finalize the previous descriptor. This pairs with
962 } while (!atomic_long_try_cmpxchg(&desc_ring
->head_id
, &head_id
,
963 id
)); /* LMM(desc_reserve:D) */
965 desc
= to_desc(desc_ring
, id
);
968 * If the descriptor has been recycled, verify the old state val.
969 * See "ABA Issues" about why this verification is performed.
971 prev_state_val
= atomic_long_read(&desc
->state_var
); /* LMM(desc_reserve:E) */
972 if (prev_state_val
&&
973 get_desc_state(id_prev_wrap
, prev_state_val
) != desc_reusable
) {
979 * Assign the descriptor a new ID and set its state to reserved.
980 * See "ABA Issues" about why cmpxchg() instead of set() is used.
982 * Guarantee the new descriptor ID and state is stored before making
983 * any other changes. A write memory barrier is sufficient for this.
984 * This pairs with desc_read:D.
986 if (!atomic_long_try_cmpxchg(&desc
->state_var
, &prev_state_val
,
987 DESC_SV(id
, desc_reserved
))) { /* LMM(desc_reserve:F) */
992 /* Now data in @desc can be modified: LMM(desc_reserve:G) */
998 /* Determine the end of a data block. */
999 static unsigned long get_next_lpos(struct prb_data_ring
*data_ring
,
1000 unsigned long lpos
, unsigned int size
)
1002 unsigned long begin_lpos
;
1003 unsigned long next_lpos
;
1006 next_lpos
= lpos
+ size
;
1008 /* First check if the data block does not wrap. */
1009 if (DATA_WRAPS(data_ring
, begin_lpos
) == DATA_WRAPS(data_ring
, next_lpos
))
1012 /* Wrapping data blocks store their data at the beginning. */
1013 return (DATA_THIS_WRAP_START_LPOS(data_ring
, next_lpos
) + size
);
1017 * Allocate a new data block, invalidating the oldest data block(s)
1018 * if necessary. This function also associates the data block with
1019 * a specified descriptor.
1021 static char *data_alloc(struct printk_ringbuffer
*rb
, unsigned int size
,
1022 struct prb_data_blk_lpos
*blk_lpos
, unsigned long id
)
1024 struct prb_data_ring
*data_ring
= &rb
->text_data_ring
;
1025 struct prb_data_block
*blk
;
1026 unsigned long begin_lpos
;
1027 unsigned long next_lpos
;
1030 /* Specify a data-less block. */
1031 blk_lpos
->begin
= NO_LPOS
;
1032 blk_lpos
->next
= NO_LPOS
;
1036 size
= to_blk_size(size
);
1038 begin_lpos
= atomic_long_read(&data_ring
->head_lpos
);
1041 next_lpos
= get_next_lpos(data_ring
, begin_lpos
, size
);
1043 if (!data_push_tail(rb
, next_lpos
- DATA_SIZE(data_ring
))) {
1044 /* Failed to allocate, specify a data-less block. */
1045 blk_lpos
->begin
= FAILED_LPOS
;
1046 blk_lpos
->next
= FAILED_LPOS
;
1051 * 1. Guarantee any descriptor states that have transitioned
1052 * to reusable are stored before modifying the newly
1053 * allocated data area. A full memory barrier is needed
1054 * since other CPUs may have made the descriptor states
1055 * reusable. See data_push_tail:A about why the reusable
1056 * states are visible. This pairs with desc_read:D.
1058 * 2. Guarantee any updated tail lpos is stored before
1059 * modifying the newly allocated data area. Another CPU may
1060 * be in data_make_reusable() and is reading a block ID
1061 * from this area. data_make_reusable() can handle reading
1062 * a garbage block ID value, but then it must be able to
1063 * load a new tail lpos. A full memory barrier is needed
1064 * since other CPUs may have updated the tail lpos. This
1065 * pairs with data_push_tail:B.
1067 } while (!atomic_long_try_cmpxchg(&data_ring
->head_lpos
, &begin_lpos
,
1068 next_lpos
)); /* LMM(data_alloc:A) */
1070 blk
= to_block(data_ring
, begin_lpos
);
1071 blk
->id
= id
; /* LMM(data_alloc:B) */
1073 if (DATA_WRAPS(data_ring
, begin_lpos
) != DATA_WRAPS(data_ring
, next_lpos
)) {
1074 /* Wrapping data blocks store their data at the beginning. */
1075 blk
= to_block(data_ring
, 0);
1078 * Store the ID on the wrapped block for consistency.
1079 * The printk_ringbuffer does not actually use it.
1084 blk_lpos
->begin
= begin_lpos
;
1085 blk_lpos
->next
= next_lpos
;
1087 return &blk
->data
[0];
1091 * Try to resize an existing data block associated with the descriptor
1092 * specified by @id. If the resized data block should become wrapped, it
1093 * copies the old data to the new data block. If @size yields a data block
1094 * with the same or less size, the data block is left as is.
1096 * Fail if this is not the last allocated data block or if there is not
1097 * enough space or it is not possible make enough space.
1099 * Return a pointer to the beginning of the entire data buffer or NULL on
1102 static char *data_realloc(struct printk_ringbuffer
*rb
, unsigned int size
,
1103 struct prb_data_blk_lpos
*blk_lpos
, unsigned long id
)
1105 struct prb_data_ring
*data_ring
= &rb
->text_data_ring
;
1106 struct prb_data_block
*blk
;
1107 unsigned long head_lpos
;
1108 unsigned long next_lpos
;
1111 /* Reallocation only works if @blk_lpos is the newest data block. */
1112 head_lpos
= atomic_long_read(&data_ring
->head_lpos
);
1113 if (head_lpos
!= blk_lpos
->next
)
1116 /* Keep track if @blk_lpos was a wrapping data block. */
1117 wrapped
= (DATA_WRAPS(data_ring
, blk_lpos
->begin
) != DATA_WRAPS(data_ring
, blk_lpos
->next
));
1119 size
= to_blk_size(size
);
1121 next_lpos
= get_next_lpos(data_ring
, blk_lpos
->begin
, size
);
1123 /* If the data block does not increase, there is nothing to do. */
1124 if (head_lpos
- next_lpos
< DATA_SIZE(data_ring
)) {
1126 blk
= to_block(data_ring
, 0);
1128 blk
= to_block(data_ring
, blk_lpos
->begin
);
1129 return &blk
->data
[0];
1132 if (!data_push_tail(rb
, next_lpos
- DATA_SIZE(data_ring
)))
1135 /* The memory barrier involvement is the same as data_alloc:A. */
1136 if (!atomic_long_try_cmpxchg(&data_ring
->head_lpos
, &head_lpos
,
1137 next_lpos
)) { /* LMM(data_realloc:A) */
1141 blk
= to_block(data_ring
, blk_lpos
->begin
);
1143 if (DATA_WRAPS(data_ring
, blk_lpos
->begin
) != DATA_WRAPS(data_ring
, next_lpos
)) {
1144 struct prb_data_block
*old_blk
= blk
;
1146 /* Wrapping data blocks store their data at the beginning. */
1147 blk
= to_block(data_ring
, 0);
1150 * Store the ID on the wrapped block for consistency.
1151 * The printk_ringbuffer does not actually use it.
1157 * Since the allocated space is now in the newly
1158 * created wrapping data block, copy the content
1159 * from the old data block.
1161 memcpy(&blk
->data
[0], &old_blk
->data
[0],
1162 (blk_lpos
->next
- blk_lpos
->begin
) - sizeof(blk
->id
));
1166 blk_lpos
->next
= next_lpos
;
1168 return &blk
->data
[0];
1171 /* Return the number of bytes used by a data block. */
1172 static unsigned int space_used(struct prb_data_ring
*data_ring
,
1173 struct prb_data_blk_lpos
*blk_lpos
)
1175 /* Data-less blocks take no space. */
1176 if (BLK_DATALESS(blk_lpos
))
1179 if (DATA_WRAPS(data_ring
, blk_lpos
->begin
) == DATA_WRAPS(data_ring
, blk_lpos
->next
)) {
1180 /* Data block does not wrap. */
1181 return (DATA_INDEX(data_ring
, blk_lpos
->next
) -
1182 DATA_INDEX(data_ring
, blk_lpos
->begin
));
1186 * For wrapping data blocks, the trailing (wasted) space is
1189 return (DATA_INDEX(data_ring
, blk_lpos
->next
) +
1190 DATA_SIZE(data_ring
) - DATA_INDEX(data_ring
, blk_lpos
->begin
));
1194 * Given @blk_lpos, return a pointer to the writer data from the data block
1195 * and calculate the size of the data part. A NULL pointer is returned if
1196 * @blk_lpos specifies values that could never be legal.
1198 * This function (used by readers) performs strict validation on the lpos
1199 * values to possibly detect bugs in the writer code. A WARN_ON_ONCE() is
1200 * triggered if an internal error is detected.
1202 static const char *get_data(struct prb_data_ring
*data_ring
,
1203 struct prb_data_blk_lpos
*blk_lpos
,
1204 unsigned int *data_size
)
1206 struct prb_data_block
*db
;
1208 /* Data-less data block description. */
1209 if (BLK_DATALESS(blk_lpos
)) {
1210 if (blk_lpos
->begin
== NO_LPOS
&& blk_lpos
->next
== NO_LPOS
) {
1217 /* Regular data block: @begin less than @next and in same wrap. */
1218 if (DATA_WRAPS(data_ring
, blk_lpos
->begin
) == DATA_WRAPS(data_ring
, blk_lpos
->next
) &&
1219 blk_lpos
->begin
< blk_lpos
->next
) {
1220 db
= to_block(data_ring
, blk_lpos
->begin
);
1221 *data_size
= blk_lpos
->next
- blk_lpos
->begin
;
1223 /* Wrapping data block: @begin is one wrap behind @next. */
1224 } else if (DATA_WRAPS(data_ring
, blk_lpos
->begin
+ DATA_SIZE(data_ring
)) ==
1225 DATA_WRAPS(data_ring
, blk_lpos
->next
)) {
1226 db
= to_block(data_ring
, 0);
1227 *data_size
= DATA_INDEX(data_ring
, blk_lpos
->next
);
1229 /* Illegal block description. */
1235 /* A valid data block will always be aligned to the ID size. */
1236 if (WARN_ON_ONCE(blk_lpos
->begin
!= ALIGN(blk_lpos
->begin
, sizeof(db
->id
))) ||
1237 WARN_ON_ONCE(blk_lpos
->next
!= ALIGN(blk_lpos
->next
, sizeof(db
->id
)))) {
1241 /* A valid data block will always have at least an ID. */
1242 if (WARN_ON_ONCE(*data_size
< sizeof(db
->id
)))
1245 /* Subtract block ID space from size to reflect data size. */
1246 *data_size
-= sizeof(db
->id
);
1248 return &db
->data
[0];
1252 * Attempt to transition the newest descriptor from committed back to reserved
1253 * so that the record can be modified by a writer again. This is only possible
1254 * if the descriptor is not yet finalized and the provided @caller_id matches.
1256 static struct prb_desc
*desc_reopen_last(struct prb_desc_ring
*desc_ring
,
1257 u32 caller_id
, unsigned long *id_out
)
1259 unsigned long prev_state_val
;
1260 enum desc_state d_state
;
1261 struct prb_desc desc
;
1266 id
= atomic_long_read(&desc_ring
->head_id
);
1269 * To reduce unnecessarily reopening, first check if the descriptor
1270 * state and caller ID are correct.
1272 d_state
= desc_read(desc_ring
, id
, &desc
, NULL
, &cid
);
1273 if (d_state
!= desc_committed
|| cid
!= caller_id
)
1276 d
= to_desc(desc_ring
, id
);
1278 prev_state_val
= DESC_SV(id
, desc_committed
);
1281 * Guarantee the reserved state is stored before reading any
1282 * record data. A full memory barrier is needed because @state_var
1283 * modification is followed by reading. This pairs with _prb_commit:B.
1285 * Memory barrier involvement:
1287 * If desc_reopen_last:A reads from _prb_commit:B, then
1288 * prb_reserve_in_last:A reads from _prb_commit:A.
1292 * WMB from _prb_commit:A to _prb_commit:B
1294 * MB If desc_reopen_last:A to prb_reserve_in_last:A
1296 if (!atomic_long_try_cmpxchg(&d
->state_var
, &prev_state_val
,
1297 DESC_SV(id
, desc_reserved
))) { /* LMM(desc_reopen_last:A) */
1306 * prb_reserve_in_last() - Re-reserve and extend the space in the ringbuffer
1307 * used by the newest record.
1309 * @e: The entry structure to setup.
1310 * @rb: The ringbuffer to re-reserve and extend data in.
1311 * @r: The record structure to allocate buffers for.
1312 * @caller_id: The caller ID of the caller (reserving writer).
1313 * @max_size: Fail if the extended size would be greater than this.
1315 * This is the public function available to writers to re-reserve and extend
1318 * The writer specifies the text size to extend (not the new total size) by
1319 * setting the @text_buf_size field of @r. To ensure proper initialization
1320 * of @r, prb_rec_init_wr() should be used.
1322 * This function will fail if @caller_id does not match the caller ID of the
1323 * newest record. In that case the caller must reserve new data using
1326 * Context: Any context. Disables local interrupts on success.
1327 * Return: true if text data could be extended, otherwise false.
1331 * - @r->text_buf points to the beginning of the entire text buffer.
1333 * - @r->text_buf_size is set to the new total size of the buffer.
1335 * - @r->info is not touched so that @r->info->text_len could be used
1336 * to append the text.
1338 * - prb_record_text_space() can be used on @e to query the new
1339 * actually used space.
1341 * Important: All @r->info fields will already be set with the current values
1342 * for the record. I.e. @r->info->text_len will be less than
1343 * @text_buf_size. Writers can use @r->info->text_len to know
1344 * where concatenation begins and writers should update
1345 * @r->info->text_len after concatenating.
1347 bool prb_reserve_in_last(struct prb_reserved_entry
*e
, struct printk_ringbuffer
*rb
,
1348 struct printk_record
*r
, u32 caller_id
, unsigned int max_size
)
1350 struct prb_desc_ring
*desc_ring
= &rb
->desc_ring
;
1351 struct printk_info
*info
;
1352 unsigned int data_size
;
1356 local_irq_save(e
->irqflags
);
1358 /* Transition the newest descriptor back to the reserved state. */
1359 d
= desc_reopen_last(desc_ring
, caller_id
, &id
);
1361 local_irq_restore(e
->irqflags
);
1365 /* Now the writer has exclusive access: LMM(prb_reserve_in_last:A) */
1367 info
= to_info(desc_ring
, id
);
1370 * Set the @e fields here so that prb_commit() can be used if
1371 * anything fails from now on.
1377 * desc_reopen_last() checked the caller_id, but there was no
1378 * exclusive access at that point. The descriptor may have
1379 * changed since then.
1381 if (caller_id
!= info
->caller_id
)
1384 if (BLK_DATALESS(&d
->text_blk_lpos
)) {
1385 if (WARN_ON_ONCE(info
->text_len
!= 0)) {
1386 pr_warn_once("wrong text_len value (%hu, expecting 0)\n",
1391 if (!data_check_size(&rb
->text_data_ring
, r
->text_buf_size
))
1394 if (r
->text_buf_size
> max_size
)
1397 r
->text_buf
= data_alloc(rb
, r
->text_buf_size
,
1398 &d
->text_blk_lpos
, id
);
1400 if (!get_data(&rb
->text_data_ring
, &d
->text_blk_lpos
, &data_size
))
1404 * Increase the buffer size to include the original size. If
1405 * the meta data (@text_len) is not sane, use the full data
1408 if (WARN_ON_ONCE(info
->text_len
> data_size
)) {
1409 pr_warn_once("wrong text_len value (%hu, expecting <=%u)\n",
1410 info
->text_len
, data_size
);
1411 info
->text_len
= data_size
;
1413 r
->text_buf_size
+= info
->text_len
;
1415 if (!data_check_size(&rb
->text_data_ring
, r
->text_buf_size
))
1418 if (r
->text_buf_size
> max_size
)
1421 r
->text_buf
= data_realloc(rb
, r
->text_buf_size
,
1422 &d
->text_blk_lpos
, id
);
1424 if (r
->text_buf_size
&& !r
->text_buf
)
1429 e
->text_space
= space_used(&rb
->text_data_ring
, &d
->text_blk_lpos
);
1434 /* prb_commit() re-enabled interrupts. */
1436 /* Make it clear to the caller that the re-reserve failed. */
1437 memset(r
, 0, sizeof(*r
));
1442 * Attempt to finalize a specified descriptor. If this fails, the descriptor
1443 * is either already final or it will finalize itself when the writer commits.
1445 static void desc_make_final(struct prb_desc_ring
*desc_ring
, unsigned long id
)
1447 unsigned long prev_state_val
= DESC_SV(id
, desc_committed
);
1448 struct prb_desc
*d
= to_desc(desc_ring
, id
);
1450 atomic_long_cmpxchg_relaxed(&d
->state_var
, prev_state_val
,
1451 DESC_SV(id
, desc_finalized
)); /* LMM(desc_make_final:A) */
1455 * prb_reserve() - Reserve space in the ringbuffer.
1457 * @e: The entry structure to setup.
1458 * @rb: The ringbuffer to reserve data in.
1459 * @r: The record structure to allocate buffers for.
1461 * This is the public function available to writers to reserve data.
1463 * The writer specifies the text size to reserve by setting the
1464 * @text_buf_size field of @r. To ensure proper initialization of @r,
1465 * prb_rec_init_wr() should be used.
1467 * Context: Any context. Disables local interrupts on success.
1468 * Return: true if at least text data could be allocated, otherwise false.
1470 * On success, the fields @info and @text_buf of @r will be set by this
1471 * function and should be filled in by the writer before committing. Also
1472 * on success, prb_record_text_space() can be used on @e to query the actual
1473 * space used for the text data block.
1475 * Important: @info->text_len needs to be set correctly by the writer in
1476 * order for data to be readable and/or extended. Its value
1477 * is initialized to 0.
1479 bool prb_reserve(struct prb_reserved_entry
*e
, struct printk_ringbuffer
*rb
,
1480 struct printk_record
*r
)
1482 struct prb_desc_ring
*desc_ring
= &rb
->desc_ring
;
1483 struct printk_info
*info
;
1488 if (!data_check_size(&rb
->text_data_ring
, r
->text_buf_size
))
1492 * Descriptors in the reserved state act as blockers to all further
1493 * reservations once the desc_ring has fully wrapped. Disable
1494 * interrupts during the reserve/commit window in order to minimize
1495 * the likelihood of this happening.
1497 local_irq_save(e
->irqflags
);
1499 if (!desc_reserve(rb
, &id
)) {
1500 /* Descriptor reservation failures are tracked. */
1501 atomic_long_inc(&rb
->fail
);
1502 local_irq_restore(e
->irqflags
);
1506 d
= to_desc(desc_ring
, id
);
1507 info
= to_info(desc_ring
, id
);
1510 * All @info fields (except @seq) are cleared and must be filled in
1511 * by the writer. Save @seq before clearing because it is used to
1512 * determine the new sequence number.
1515 memset(info
, 0, sizeof(*info
));
1518 * Set the @e fields here so that prb_commit() can be used if
1519 * text data allocation fails.
1525 * Initialize the sequence number if it has "never been set".
1526 * Otherwise just increment it by a full wrap.
1528 * @seq is considered "never been set" if it has a value of 0,
1529 * _except_ for @infos[0], which was specially setup by the ringbuffer
1530 * initializer and therefore is always considered as set.
1532 * See the "Bootstrap" comment block in printk_ringbuffer.h for
1533 * details about how the initializer bootstraps the descriptors.
1535 if (seq
== 0 && DESC_INDEX(desc_ring
, id
) != 0)
1536 info
->seq
= DESC_INDEX(desc_ring
, id
);
1538 info
->seq
= seq
+ DESCS_COUNT(desc_ring
);
1541 * New data is about to be reserved. Once that happens, previous
1542 * descriptors are no longer able to be extended. Finalize the
1543 * previous descriptor now so that it can be made available to
1544 * readers. (For seq==0 there is no previous descriptor.)
1547 desc_make_final(desc_ring
, DESC_ID(id
- 1));
1549 r
->text_buf
= data_alloc(rb
, r
->text_buf_size
, &d
->text_blk_lpos
, id
);
1550 /* If text data allocation fails, a data-less record is committed. */
1551 if (r
->text_buf_size
&& !r
->text_buf
) {
1553 /* prb_commit() re-enabled interrupts. */
1559 /* Record full text space used by record. */
1560 e
->text_space
= space_used(&rb
->text_data_ring
, &d
->text_blk_lpos
);
1564 /* Make it clear to the caller that the reserve failed. */
1565 memset(r
, 0, sizeof(*r
));
1569 /* Commit the data (possibly finalizing it) and restore interrupts. */
1570 static void _prb_commit(struct prb_reserved_entry
*e
, unsigned long state_val
)
1572 struct prb_desc_ring
*desc_ring
= &e
->rb
->desc_ring
;
1573 struct prb_desc
*d
= to_desc(desc_ring
, e
->id
);
1574 unsigned long prev_state_val
= DESC_SV(e
->id
, desc_reserved
);
1576 /* Now the writer has finished all writing: LMM(_prb_commit:A) */
1579 * Set the descriptor as committed. See "ABA Issues" about why
1580 * cmpxchg() instead of set() is used.
1582 * 1 Guarantee all record data is stored before the descriptor state
1583 * is stored as committed. A write memory barrier is sufficient
1584 * for this. This pairs with desc_read:B and desc_reopen_last:A.
1586 * 2. Guarantee the descriptor state is stored as committed before
1587 * re-checking the head ID in order to possibly finalize this
1588 * descriptor. This pairs with desc_reserve:D.
1590 * Memory barrier involvement:
1592 * If prb_commit:A reads from desc_reserve:D, then
1593 * desc_make_final:A reads from _prb_commit:B.
1597 * MB _prb_commit:B to prb_commit:A
1599 * MB desc_reserve:D to desc_make_final:A
1601 if (!atomic_long_try_cmpxchg(&d
->state_var
, &prev_state_val
,
1602 DESC_SV(e
->id
, state_val
))) { /* LMM(_prb_commit:B) */
1606 /* Restore interrupts, the reserve/commit window is finished. */
1607 local_irq_restore(e
->irqflags
);
1611 * prb_commit() - Commit (previously reserved) data to the ringbuffer.
1613 * @e: The entry containing the reserved data information.
1615 * This is the public function available to writers to commit data.
1617 * Note that the data is not yet available to readers until it is finalized.
1618 * Finalizing happens automatically when space for the next record is
1621 * See prb_final_commit() for a version of this function that finalizes
1624 * Context: Any context. Enables local interrupts.
1626 void prb_commit(struct prb_reserved_entry
*e
)
1628 struct prb_desc_ring
*desc_ring
= &e
->rb
->desc_ring
;
1629 unsigned long head_id
;
1631 _prb_commit(e
, desc_committed
);
1634 * If this descriptor is no longer the head (i.e. a new record has
1635 * been allocated), extending the data for this record is no longer
1636 * allowed and therefore it must be finalized.
1638 head_id
= atomic_long_read(&desc_ring
->head_id
); /* LMM(prb_commit:A) */
1639 if (head_id
!= e
->id
)
1640 desc_make_final(desc_ring
, e
->id
);
1644 * prb_final_commit() - Commit and finalize (previously reserved) data to
1647 * @e: The entry containing the reserved data information.
1649 * This is the public function available to writers to commit+finalize data.
1651 * By finalizing, the data is made immediately available to readers.
1653 * This function should only be used if there are no intentions of extending
1654 * this data using prb_reserve_in_last().
1656 * Context: Any context. Enables local interrupts.
1658 void prb_final_commit(struct prb_reserved_entry
*e
)
1660 _prb_commit(e
, desc_finalized
);
1664 * Count the number of lines in provided text. All text has at least 1 line
1665 * (even if @text_size is 0). Each '\n' processed is counted as an additional
1668 static unsigned int count_lines(const char *text
, unsigned int text_size
)
1670 unsigned int next_size
= text_size
;
1671 unsigned int line_count
= 1;
1672 const char *next
= text
;
1675 next
= memchr(next
, '\n', next_size
);
1680 next_size
= text_size
- (next
- text
);
1687 * Given @blk_lpos, copy an expected @len of data into the provided buffer.
1688 * If @line_count is provided, count the number of lines in the data.
1690 * This function (used by readers) performs strict validation on the data
1691 * size to possibly detect bugs in the writer code. A WARN_ON_ONCE() is
1692 * triggered if an internal error is detected.
1694 static bool copy_data(struct prb_data_ring
*data_ring
,
1695 struct prb_data_blk_lpos
*blk_lpos
, u16 len
, char *buf
,
1696 unsigned int buf_size
, unsigned int *line_count
)
1698 unsigned int data_size
;
1701 /* Caller might not want any data. */
1702 if ((!buf
|| !buf_size
) && !line_count
)
1705 data
= get_data(data_ring
, blk_lpos
, &data_size
);
1710 * Actual cannot be less than expected. It can be more than expected
1711 * because of the trailing alignment padding.
1713 * Note that invalid @len values can occur because the caller loads
1714 * the value during an allowed data race.
1716 if (data_size
< (unsigned int)len
)
1719 /* Caller interested in the line count? */
1721 *line_count
= count_lines(data
, data_size
);
1723 /* Caller interested in the data content? */
1724 if (!buf
|| !buf_size
)
1727 data_size
= min_t(u16
, buf_size
, len
);
1729 memcpy(&buf
[0], data
, data_size
); /* LMM(copy_data:A) */
1734 * This is an extended version of desc_read(). It gets a copy of a specified
1735 * descriptor. However, it also verifies that the record is finalized and has
1736 * the sequence number @seq. On success, 0 is returned.
1738 * Error return values:
1739 * -EINVAL: A finalized record with sequence number @seq does not exist.
1740 * -ENOENT: A finalized record with sequence number @seq exists, but its data
1741 * is not available. This is a valid record, so readers should
1742 * continue with the next record.
1744 static int desc_read_finalized_seq(struct prb_desc_ring
*desc_ring
,
1745 unsigned long id
, u64 seq
,
1746 struct prb_desc
*desc_out
)
1748 struct prb_data_blk_lpos
*blk_lpos
= &desc_out
->text_blk_lpos
;
1749 enum desc_state d_state
;
1752 d_state
= desc_read(desc_ring
, id
, desc_out
, &s
, NULL
);
1755 * An unexpected @id (desc_miss) or @seq mismatch means the record
1756 * does not exist. A descriptor in the reserved or committed state
1757 * means the record does not yet exist for the reader.
1759 if (d_state
== desc_miss
||
1760 d_state
== desc_reserved
||
1761 d_state
== desc_committed
||
1767 * A descriptor in the reusable state may no longer have its data
1768 * available; report it as existing but with lost data. Or the record
1769 * may actually be a record with lost data.
1771 if (d_state
== desc_reusable
||
1772 (blk_lpos
->begin
== FAILED_LPOS
&& blk_lpos
->next
== FAILED_LPOS
)) {
1780 * Copy the ringbuffer data from the record with @seq to the provided
1781 * @r buffer. On success, 0 is returned.
1783 * See desc_read_finalized_seq() for error return values.
1785 static int prb_read(struct printk_ringbuffer
*rb
, u64 seq
,
1786 struct printk_record
*r
, unsigned int *line_count
)
1788 struct prb_desc_ring
*desc_ring
= &rb
->desc_ring
;
1789 struct printk_info
*info
= to_info(desc_ring
, seq
);
1790 struct prb_desc
*rdesc
= to_desc(desc_ring
, seq
);
1791 atomic_long_t
*state_var
= &rdesc
->state_var
;
1792 struct prb_desc desc
;
1796 /* Extract the ID, used to specify the descriptor to read. */
1797 id
= DESC_ID(atomic_long_read(state_var
));
1799 /* Get a local copy of the correct descriptor (if available). */
1800 err
= desc_read_finalized_seq(desc_ring
, id
, seq
, &desc
);
1803 * If @r is NULL, the caller is only interested in the availability
1809 /* If requested, copy meta data. */
1811 memcpy(r
->info
, info
, sizeof(*(r
->info
)));
1813 /* Copy text data. If it fails, this is a data-less record. */
1814 if (!copy_data(&rb
->text_data_ring
, &desc
.text_blk_lpos
, info
->text_len
,
1815 r
->text_buf
, r
->text_buf_size
, line_count
)) {
1819 /* Ensure the record is still finalized and has the same @seq. */
1820 return desc_read_finalized_seq(desc_ring
, id
, seq
, &desc
);
1823 /* Get the sequence number of the tail descriptor. */
1824 static u64
prb_first_seq(struct printk_ringbuffer
*rb
)
1826 struct prb_desc_ring
*desc_ring
= &rb
->desc_ring
;
1827 enum desc_state d_state
;
1828 struct prb_desc desc
;
1833 id
= atomic_long_read(&rb
->desc_ring
.tail_id
); /* LMM(prb_first_seq:A) */
1835 d_state
= desc_read(desc_ring
, id
, &desc
, &seq
, NULL
); /* LMM(prb_first_seq:B) */
1838 * This loop will not be infinite because the tail is
1839 * _always_ in the finalized or reusable state.
1841 if (d_state
== desc_finalized
|| d_state
== desc_reusable
)
1845 * Guarantee the last state load from desc_read() is before
1846 * reloading @tail_id in order to see a new tail in the case
1847 * that the descriptor has been recycled. This pairs with
1850 * Memory barrier involvement:
1852 * If prb_first_seq:B reads from desc_reserve:F, then
1853 * prb_first_seq:A reads from desc_push_tail:B.
1857 * MB from desc_push_tail:B to desc_reserve:F
1859 * RMB prb_first_seq:B to prb_first_seq:A
1861 smp_rmb(); /* LMM(prb_first_seq:C) */
1868 * Non-blocking read of a record. Updates @seq to the last finalized record
1869 * (which may have no data available).
1871 * See the description of prb_read_valid() and prb_read_valid_info()
1874 static bool _prb_read_valid(struct printk_ringbuffer
*rb
, u64
*seq
,
1875 struct printk_record
*r
, unsigned int *line_count
)
1880 while ((err
= prb_read(rb
, *seq
, r
, line_count
))) {
1881 tail_seq
= prb_first_seq(rb
);
1883 if (*seq
< tail_seq
) {
1885 * Behind the tail. Catch up and try again. This
1886 * can happen for -ENOENT and -EINVAL cases.
1890 } else if (err
== -ENOENT
) {
1891 /* Record exists, but no data available. Skip. */
1895 /* Non-existent/non-finalized record. Must stop. */
1904 * prb_read_valid() - Non-blocking read of a requested record or (if gone)
1905 * the next available record.
1907 * @rb: The ringbuffer to read from.
1908 * @seq: The sequence number of the record to read.
1909 * @r: A record data buffer to store the read record to.
1911 * This is the public function available to readers to read a record.
1913 * The reader provides the @info and @text_buf buffers of @r to be
1914 * filled in. Any of the buffer pointers can be set to NULL if the reader
1915 * is not interested in that data. To ensure proper initialization of @r,
1916 * prb_rec_init_rd() should be used.
1918 * Context: Any context.
1919 * Return: true if a record was read, otherwise false.
1921 * On success, the reader must check r->info.seq to see which record was
1922 * actually read. This allows the reader to detect dropped records.
1924 * Failure means @seq refers to a not yet written record.
1926 bool prb_read_valid(struct printk_ringbuffer
*rb
, u64 seq
,
1927 struct printk_record
*r
)
1929 return _prb_read_valid(rb
, &seq
, r
, NULL
);
1933 * prb_read_valid_info() - Non-blocking read of meta data for a requested
1934 * record or (if gone) the next available record.
1936 * @rb: The ringbuffer to read from.
1937 * @seq: The sequence number of the record to read.
1938 * @info: A buffer to store the read record meta data to.
1939 * @line_count: A buffer to store the number of lines in the record text.
1941 * This is the public function available to readers to read only the
1942 * meta data of a record.
1944 * The reader provides the @info, @line_count buffers to be filled in.
1945 * Either of the buffer pointers can be set to NULL if the reader is not
1946 * interested in that data.
1948 * Context: Any context.
1949 * Return: true if a record's meta data was read, otherwise false.
1951 * On success, the reader must check info->seq to see which record meta data
1952 * was actually read. This allows the reader to detect dropped records.
1954 * Failure means @seq refers to a not yet written record.
1956 bool prb_read_valid_info(struct printk_ringbuffer
*rb
, u64 seq
,
1957 struct printk_info
*info
, unsigned int *line_count
)
1959 struct printk_record r
;
1961 prb_rec_init_rd(&r
, info
, NULL
, 0);
1963 return _prb_read_valid(rb
, &seq
, &r
, line_count
);
1967 * prb_first_valid_seq() - Get the sequence number of the oldest available
1970 * @rb: The ringbuffer to get the sequence number from.
1972 * This is the public function available to readers to see what the
1973 * first/oldest valid sequence number is.
1975 * This provides readers a starting point to begin iterating the ringbuffer.
1977 * Context: Any context.
1978 * Return: The sequence number of the first/oldest record or, if the
1979 * ringbuffer is empty, 0 is returned.
1981 u64
prb_first_valid_seq(struct printk_ringbuffer
*rb
)
1985 if (!_prb_read_valid(rb
, &seq
, NULL
, NULL
))
1992 * prb_next_seq() - Get the sequence number after the last available record.
1994 * @rb: The ringbuffer to get the sequence number from.
1996 * This is the public function available to readers to see what the next
1997 * newest sequence number available to readers will be.
1999 * This provides readers a sequence number to jump to if all currently
2000 * available records should be skipped.
2002 * Context: Any context.
2003 * Return: The sequence number of the next newest (not yet available) record
2006 u64
prb_next_seq(struct printk_ringbuffer
*rb
)
2010 /* Search forward from the oldest descriptor. */
2011 while (_prb_read_valid(rb
, &seq
, NULL
, NULL
))
2018 * prb_init() - Initialize a ringbuffer to use provided external buffers.
2020 * @rb: The ringbuffer to initialize.
2021 * @text_buf: The data buffer for text data.
2022 * @textbits: The size of @text_buf as a power-of-2 value.
2023 * @descs: The descriptor buffer for ringbuffer records.
2024 * @descbits: The count of @descs items as a power-of-2 value.
2025 * @infos: The printk_info buffer for ringbuffer records.
2027 * This is the public function available to writers to setup a ringbuffer
2028 * during runtime using provided buffers.
2030 * This must match the initialization of DEFINE_PRINTKRB().
2032 * Context: Any context.
2034 void prb_init(struct printk_ringbuffer
*rb
,
2035 char *text_buf
, unsigned int textbits
,
2036 struct prb_desc
*descs
, unsigned int descbits
,
2037 struct printk_info
*infos
)
2039 memset(descs
, 0, _DESCS_COUNT(descbits
) * sizeof(descs
[0]));
2040 memset(infos
, 0, _DESCS_COUNT(descbits
) * sizeof(infos
[0]));
2042 rb
->desc_ring
.count_bits
= descbits
;
2043 rb
->desc_ring
.descs
= descs
;
2044 rb
->desc_ring
.infos
= infos
;
2045 atomic_long_set(&rb
->desc_ring
.head_id
, DESC0_ID(descbits
));
2046 atomic_long_set(&rb
->desc_ring
.tail_id
, DESC0_ID(descbits
));
2048 rb
->text_data_ring
.size_bits
= textbits
;
2049 rb
->text_data_ring
.data
= text_buf
;
2050 atomic_long_set(&rb
->text_data_ring
.head_lpos
, BLK0_LPOS(textbits
));
2051 atomic_long_set(&rb
->text_data_ring
.tail_lpos
, BLK0_LPOS(textbits
));
2053 atomic_long_set(&rb
->fail
, 0);
2055 atomic_long_set(&(descs
[_DESCS_COUNT(descbits
) - 1].state_var
), DESC0_SV(descbits
));
2056 descs
[_DESCS_COUNT(descbits
) - 1].text_blk_lpos
.begin
= FAILED_LPOS
;
2057 descs
[_DESCS_COUNT(descbits
) - 1].text_blk_lpos
.next
= FAILED_LPOS
;
2059 infos
[0].seq
= -(u64
)_DESCS_COUNT(descbits
);
2060 infos
[_DESCS_COUNT(descbits
) - 1].seq
= 0;
2064 * prb_record_text_space() - Query the full actual used ringbuffer space for
2065 * the text data of a reserved entry.
2067 * @e: The successfully reserved entry to query.
2069 * This is the public function available to writers to see how much actual
2070 * space is used in the ringbuffer to store the text data of the specified
2073 * This function is only valid if @e has been successfully reserved using
2076 * Context: Any context.
2077 * Return: The size in bytes used by the text data of the associated record.
2079 unsigned int prb_record_text_space(struct prb_reserved_entry
*e
)
2081 return e
->text_space
;