1 // SPDX-License-Identifier: GPL-2.0
2 
3 #include <linux/kernel.h>
4 #include <linux/irqflags.h>
5 #include <linux/string.h>
6 #include <linux/errno.h>
7 #include <linux/bug.h>
8 #include "printk_ringbuffer.h"
9 
10 /**
11  * DOC: printk_ringbuffer overview
12  *
13  * Data Structure
14  * --------------
15  * The printk_ringbuffer is made up of 3 internal ringbuffers:
16  *
17  *   desc_ring
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.
22  *
23  *   text_data_ring
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.
27  *
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.
30  *
31  * Implementation
32  * --------------
33  *
34  * Descriptor Ring
35  * ~~~~~~~~~~~~~~~
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
43  * updated.
44  *
45  * Descriptors have four states:
46  *
47  *   reserved
48  *     A writer is modifying the record.
49  *
50  *   committed
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.
54  *
55  *   finalized
56  *     The record and all its data are complete and available for reading. A
57  *     writer cannot reopen the descriptor.
58  *
59  *   reusable
60  *     The record exists, but its text and/or meta data may no longer be
61  *     available.
62  *
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:
65  *
66  *   miss
67  *     The descriptor being queried has an unexpected ID.
68  *
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.
71  *
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
78  * state.
79  *
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.
83  *
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.
90  *
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:
95  *
96  *   1) A writer can simultaneously commit and finalize its record by calling
97  *      prb_final_commit() instead of prb_commit().
98  *
99  *   2) When a new record is reserved and the previous record has been
100  *      committed via prb_commit(), that previous record is automatically
101  *      finalized.
102  *
103  *   3) When a record is committed via prb_commit() and a newer record
104  *      already exists, the record being committed is automatically finalized.
105  *
106  * Data Ring
107  * ~~~~~~~~~
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
112  * array ringbuffer.
113  *
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
117  * are met:
118  *
119  *   1) The descriptor associated with the data block is in the committed
120  *      or finalized queried state.
121  *
122  *   2) The blk_lpos struct within the descriptor associated with the data
123  *      block references back to the same data block.
124  *
125  *   3) The data block is within the head/tail logical position range.
126  *
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.
133  *
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.
137  *
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.
144  *
145  * Info Array
146  * ~~~~~~~~~~
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.
152  *
153  * Usage
154  * -----
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)::
158  *
159  *	DEFINE_PRINTKRB(test_rb, 15, 5);
160  *
161  * This ringbuffer allows up to 32768 records (2 ^ 15) and has a size of
162  * 1 MiB (2 ^ (15 + 5)) for text data.
163  *
164  * Sample writer code::
165  *
166  *	const char *textstr = "message text";
167  *	struct prb_reserved_entry e;
168  *	struct printk_record r;
169  *
170  *	// specify how much to allocate
171  *	prb_rec_init_wr(&r, strlen(textstr) + 1);
172  *
173  *	if (prb_reserve(&e, &test_rb, &r)) {
174  *		snprintf(r.text_buf, r.text_buf_size, "%s", textstr);
175  *
176  *		r.info->text_len = strlen(textstr);
177  *		r.info->ts_nsec = local_clock();
178  *		r.info->caller_id = printk_caller_id();
179  *
180  *		// commit and finalize the record
181  *		prb_final_commit(&e);
182  *	}
183  *
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.
187  *
188  * Sample writer code (record extending)::
189  *
190  *		// alternate rest of previous example
191  *
192  *		r.info->text_len = strlen(textstr);
193  *		r.info->ts_nsec = local_clock();
194  *		r.info->caller_id = printk_caller_id();
195  *
196  *		// commit the record (but do not finalize yet)
197  *		prb_commit(&e);
198  *	}
199  *
200  *	...
201  *
202  *	// specify additional 5 bytes text space to extend
203  *	prb_rec_init_wr(&r, 5);
204  *
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");
209  *
210  *		r.info->text_len += 5;
211  *
212  *		// commit and finalize the record
213  *		prb_final_commit(&e);
214  *	}
215  *
216  * Sample reader code::
217  *
218  *	struct printk_info info;
219  *	struct printk_record r;
220  *	char text_buf[32];
221  *	u64 seq;
222  *
223  *	prb_rec_init_rd(&r, &info, &text_buf[0], sizeof(text_buf));
224  *
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);
228  *
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;
232  *		}
233  *
234  *		pr_info("%llu: %llu: %s\n", info.seq, info.ts_nsec,
235  *			&text_buf[0]);
236  *	}
237  *
238  * Note that additional less convenient reader functions are available to
239  * allow complex record access.
240  *
241  * ABA Issues
242  * ~~~~~~~~~~
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.
253  *
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.
258  *
259  * Memory Barriers
260  * ~~~~~~~~~~~~~~~
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::
264  *
265  *	LMM(function:letter)
266  *
267  * Comments reference the labels using only the "function:letter" part.
268  *
269  * The memory barrier pairs and their ordering are:
270  *
271  *   desc_reserve:D / desc_reserve:B
272  *     push descriptor tail (id), then push descriptor head (id)
273  *
274  *   desc_reserve:D / data_push_tail:B
275  *     push data tail (lpos), then set new descriptor reserved (state)
276  *
277  *   desc_reserve:D / desc_push_tail:C
278  *     push descriptor tail (id), then set new descriptor reserved (state)
279  *
280  *   desc_reserve:D / prb_first_seq:C
281  *     push descriptor tail (id), then set new descriptor reserved (state)
282  *
283  *   desc_reserve:F / desc_read:D
284  *     set new descriptor id and reserved (state), then allow writer changes
285  *
286  *   data_alloc:A (or data_realloc:A) / desc_read:D
287  *     set old descriptor reusable (state), then modify new data block area
288  *
289  *   data_alloc:A (or data_realloc:A) / data_push_tail:B
290  *     push data tail (lpos), then modify new data block area
291  *
292  *   _prb_commit:B / desc_read:B
293  *     store writer changes, then set new descriptor committed (state)
294  *
295  *   desc_reopen_last:A / _prb_commit:B
296  *     set descriptor reserved (state), then read descriptor data
297  *
298  *   _prb_commit:B / desc_reserve:D
299  *     set new descriptor committed (state), then check descriptor head (id)
300  *
301  *   data_push_tail:D / data_push_tail:A
302  *     set descriptor reusable (state), then push data tail (lpos)
303  *
304  *   desc_push_tail:B / desc_reserve:D
305  *     set descriptor reusable (state), then push descriptor tail (id)
306  */
307 
308 #define DATA_SIZE(data_ring)		_DATA_SIZE((data_ring)->size_bits)
309 #define DATA_SIZE_MASK(data_ring)	(DATA_SIZE(data_ring) - 1)
310 
311 #define DESCS_COUNT(desc_ring)		_DESCS_COUNT((desc_ring)->count_bits)
312 #define DESCS_COUNT_MASK(desc_ring)	(DESCS_COUNT(desc_ring) - 1)
313 
314 /* Determine the data array index from a logical position. */
315 #define DATA_INDEX(data_ring, lpos)	((lpos) & DATA_SIZE_MASK(data_ring))
316 
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))
319 
320 /* Determine how many times the data array has wrapped. */
321 #define DATA_WRAPS(data_ring, lpos)	((lpos) >> (data_ring)->size_bits)
322 
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))
327 
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))
331 
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))
335 
336 /*
337  * A data block: mapped directly to the beginning of the data block area
338  * specified as a logical position within the data ring.
339  *
340  * @id:   the ID of the associated descriptor
341  * @data: the writer data
342  *
343  * Note that the size of a data block is only known by its associated
344  * descriptor.
345  */
346 struct prb_data_block {
347 	unsigned long	id;
348 	char		data[];
349 };
350 
351 /*
352  * Return the descriptor associated with @n. @n can be either a
353  * descriptor ID or a sequence number.
354  */
355 static struct prb_desc *to_desc(struct prb_desc_ring *desc_ring, u64 n)
356 {
357 	return &desc_ring->descs[DESC_INDEX(desc_ring, n)];
358 }
359 
360 /*
361  * Return the printk_info associated with @n. @n can be either a
362  * descriptor ID or a sequence number.
363  */
364 static struct printk_info *to_info(struct prb_desc_ring *desc_ring, u64 n)
365 {
366 	return &desc_ring->infos[DESC_INDEX(desc_ring, n)];
367 }
368 
369 static struct prb_data_block *to_block(struct prb_data_ring *data_ring,
370 				       unsigned long begin_lpos)
371 {
372 	return (void *)&data_ring->data[DATA_INDEX(data_ring, begin_lpos)];
373 }
374 
375 /*
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.
378  */
379 static unsigned int to_blk_size(unsigned int size)
380 {
381 	struct prb_data_block *db = NULL;
382 
383 	size += sizeof(*db);
384 	size = ALIGN(size, sizeof(db->id));
385 	return size;
386 }
387 
388 /*
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.
393  */
394 static bool data_check_size(struct prb_data_ring *data_ring, unsigned int size)
395 {
396 	struct prb_data_block *db = NULL;
397 
398 	if (size == 0)
399 		return true;
400 
401 	/*
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.
405 	 */
406 	size = to_blk_size(size);
407 	if (size > DATA_SIZE(data_ring) - sizeof(db->id))
408 		return false;
409 
410 	return true;
411 }
412 
413 /* Query the state of a descriptor. */
414 static enum desc_state get_desc_state(unsigned long id,
415 				      unsigned long state_val)
416 {
417 	if (id != DESC_ID(state_val))
418 		return desc_miss;
419 
420 	return DESC_STATE(state_val);
421 }
422 
423 /*
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.
427  *
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
430  * consistent state.
431  */
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)
435 {
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;
441 
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) {
446 		/*
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.
450 		 */
451 		goto out;
452 	}
453 
454 	/*
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.
458 	 *
459 	 * Memory barrier involvement:
460 	 *
461 	 * If desc_read:A reads from _prb_commit:B, then desc_read:C reads
462 	 * from _prb_commit:A.
463 	 *
464 	 * Relies on:
465 	 *
466 	 * WMB from _prb_commit:A to _prb_commit:B
467 	 *    matching
468 	 * RMB from desc_read:A to desc_read:C
469 	 */
470 	smp_rmb(); /* LMM(desc_read:B) */
471 
472 	/*
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.
476 	 */
477 	if (desc_out) {
478 		memcpy(&desc_out->text_blk_lpos, &desc->text_blk_lpos,
479 		       sizeof(desc_out->text_blk_lpos)); /* LMM(desc_read:C) */
480 	}
481 	if (seq_out)
482 		*seq_out = info->seq; /* also part of desc_read:C */
483 	if (caller_id_out)
484 		*caller_id_out = info->caller_id; /* also part of desc_read:C */
485 
486 	/*
487 	 * 1. Guarantee the descriptor content is loaded before re-checking
488 	 *    the state. This avoids reading an obsolete descriptor state
489 	 *    that may not apply to the copied content. This pairs with
490 	 *    desc_reserve:F.
491 	 *
492 	 *    Memory barrier involvement:
493 	 *
494 	 *    If desc_read:C reads from desc_reserve:G, then desc_read:E
495 	 *    reads from desc_reserve:F.
496 	 *
497 	 *    Relies on:
498 	 *
499 	 *    WMB from desc_reserve:F to desc_reserve:G
500 	 *       matching
501 	 *    RMB from desc_read:C to desc_read:E
502 	 *
503 	 * 2. Guarantee the record data is loaded before re-checking the
504 	 *    state. This avoids reading an obsolete descriptor state that may
505 	 *    not apply to the copied data. This pairs with data_alloc:A and
506 	 *    data_realloc:A.
507 	 *
508 	 *    Memory barrier involvement:
509 	 *
510 	 *    If copy_data:A reads from data_alloc:B, then desc_read:E
511 	 *    reads from desc_make_reusable:A.
512 	 *
513 	 *    Relies on:
514 	 *
515 	 *    MB from desc_make_reusable:A to data_alloc:B
516 	 *       matching
517 	 *    RMB from desc_read:C to desc_read:E
518 	 *
519 	 *    Note: desc_make_reusable:A and data_alloc:B can be different
520 	 *          CPUs. However, the data_alloc:B CPU (which performs the
521 	 *          full memory barrier) must have previously seen
522 	 *          desc_make_reusable:A.
523 	 */
524 	smp_rmb(); /* LMM(desc_read:D) */
525 
526 	/*
527 	 * The data has been copied. Return the current descriptor state,
528 	 * which may have changed since the load above.
529 	 */
530 	state_val = atomic_long_read(state_var); /* LMM(desc_read:E) */
531 	d_state = get_desc_state(id, state_val);
532 out:
533 	if (desc_out)
534 		atomic_long_set(&desc_out->state_var, state_val);
535 	return d_state;
536 }
537 
538 /*
539  * Take a specified descriptor out of the finalized state by attempting
540  * the transition from finalized to reusable. Either this context or some
541  * other context will have been successful.
542  */
543 static void desc_make_reusable(struct prb_desc_ring *desc_ring,
544 			       unsigned long id)
545 {
546 	unsigned long val_finalized = DESC_SV(id, desc_finalized);
547 	unsigned long val_reusable = DESC_SV(id, desc_reusable);
548 	struct prb_desc *desc = to_desc(desc_ring, id);
549 	atomic_long_t *state_var = &desc->state_var;
550 
551 	atomic_long_cmpxchg_relaxed(state_var, val_finalized,
552 				    val_reusable); /* LMM(desc_make_reusable:A) */
553 }
554 
555 /*
556  * Given the text data ring, put the associated descriptor of each
557  * data block from @lpos_begin until @lpos_end into the reusable state.
558  *
559  * If there is any problem making the associated descriptor reusable, either
560  * the descriptor has not yet been finalized or another writer context has
561  * already pushed the tail lpos past the problematic data block. Regardless,
562  * on error the caller can re-load the tail lpos to determine the situation.
563  */
564 static bool data_make_reusable(struct printk_ringbuffer *rb,
565 			       unsigned long lpos_begin,
566 			       unsigned long lpos_end,
567 			       unsigned long *lpos_out)
568 {
569 
570 	struct prb_data_ring *data_ring = &rb->text_data_ring;
571 	struct prb_desc_ring *desc_ring = &rb->desc_ring;
572 	struct prb_data_block *blk;
573 	enum desc_state d_state;
574 	struct prb_desc desc;
575 	struct prb_data_blk_lpos *blk_lpos = &desc.text_blk_lpos;
576 	unsigned long id;
577 
578 	/* Loop until @lpos_begin has advanced to or beyond @lpos_end. */
579 	while ((lpos_end - lpos_begin) - 1 < DATA_SIZE(data_ring)) {
580 		blk = to_block(data_ring, lpos_begin);
581 
582 		/*
583 		 * Load the block ID from the data block. This is a data race
584 		 * against a writer that may have newly reserved this data
585 		 * area. If the loaded value matches a valid descriptor ID,
586 		 * the blk_lpos of that descriptor will be checked to make
587 		 * sure it points back to this data block. If the check fails,
588 		 * the data area has been recycled by another writer.
589 		 */
590 		id = blk->id; /* LMM(data_make_reusable:A) */
591 
592 		d_state = desc_read(desc_ring, id, &desc,
593 				    NULL, NULL); /* LMM(data_make_reusable:B) */
594 
595 		switch (d_state) {
596 		case desc_miss:
597 		case desc_reserved:
598 		case desc_committed:
599 			return false;
600 		case desc_finalized:
601 			/*
602 			 * This data block is invalid if the descriptor
603 			 * does not point back to it.
604 			 */
605 			if (blk_lpos->begin != lpos_begin)
606 				return false;
607 			desc_make_reusable(desc_ring, id);
608 			break;
609 		case desc_reusable:
610 			/*
611 			 * This data block is invalid if the descriptor
612 			 * does not point back to it.
613 			 */
614 			if (blk_lpos->begin != lpos_begin)
615 				return false;
616 			break;
617 		}
618 
619 		/* Advance @lpos_begin to the next data block. */
620 		lpos_begin = blk_lpos->next;
621 	}
622 
623 	*lpos_out = lpos_begin;
624 	return true;
625 }
626 
627 /*
628  * Advance the data ring tail to at least @lpos. This function puts
629  * descriptors into the reusable state if the tail is pushed beyond
630  * their associated data block.
631  */
632 static bool data_push_tail(struct printk_ringbuffer *rb, unsigned long lpos)
633 {
634 	struct prb_data_ring *data_ring = &rb->text_data_ring;
635 	unsigned long tail_lpos_new;
636 	unsigned long tail_lpos;
637 	unsigned long next_lpos;
638 
639 	/* If @lpos is from a data-less block, there is nothing to do. */
640 	if (LPOS_DATALESS(lpos))
641 		return true;
642 
643 	/*
644 	 * Any descriptor states that have transitioned to reusable due to the
645 	 * data tail being pushed to this loaded value will be visible to this
646 	 * CPU. This pairs with data_push_tail:D.
647 	 *
648 	 * Memory barrier involvement:
649 	 *
650 	 * If data_push_tail:A reads from data_push_tail:D, then this CPU can
651 	 * see desc_make_reusable:A.
652 	 *
653 	 * Relies on:
654 	 *
655 	 * MB from desc_make_reusable:A to data_push_tail:D
656 	 *    matches
657 	 * READFROM from data_push_tail:D to data_push_tail:A
658 	 *    thus
659 	 * READFROM from desc_make_reusable:A to this CPU
660 	 */
661 	tail_lpos = atomic_long_read(&data_ring->tail_lpos); /* LMM(data_push_tail:A) */
662 
663 	/*
664 	 * Loop until the tail lpos is at or beyond @lpos. This condition
665 	 * may already be satisfied, resulting in no full memory barrier
666 	 * from data_push_tail:D being performed. However, since this CPU
667 	 * sees the new tail lpos, any descriptor states that transitioned to
668 	 * the reusable state must already be visible.
669 	 */
670 	while ((lpos - tail_lpos) - 1 < DATA_SIZE(data_ring)) {
671 		/*
672 		 * Make all descriptors reusable that are associated with
673 		 * data blocks before @lpos.
674 		 */
675 		if (!data_make_reusable(rb, tail_lpos, lpos, &next_lpos)) {
676 			/*
677 			 * 1. Guarantee the block ID loaded in
678 			 *    data_make_reusable() is performed before
679 			 *    reloading the tail lpos. The failed
680 			 *    data_make_reusable() may be due to a newly
681 			 *    recycled data area causing the tail lpos to
682 			 *    have been previously pushed. This pairs with
683 			 *    data_alloc:A and data_realloc:A.
684 			 *
685 			 *    Memory barrier involvement:
686 			 *
687 			 *    If data_make_reusable:A reads from data_alloc:B,
688 			 *    then data_push_tail:C reads from
689 			 *    data_push_tail:D.
690 			 *
691 			 *    Relies on:
692 			 *
693 			 *    MB from data_push_tail:D to data_alloc:B
694 			 *       matching
695 			 *    RMB from data_make_reusable:A to
696 			 *    data_push_tail:C
697 			 *
698 			 *    Note: data_push_tail:D and data_alloc:B can be
699 			 *          different CPUs. However, the data_alloc:B
700 			 *          CPU (which performs the full memory
701 			 *          barrier) must have previously seen
702 			 *          data_push_tail:D.
703 			 *
704 			 * 2. Guarantee the descriptor state loaded in
705 			 *    data_make_reusable() is performed before
706 			 *    reloading the tail lpos. The failed
707 			 *    data_make_reusable() may be due to a newly
708 			 *    recycled descriptor causing the tail lpos to
709 			 *    have been previously pushed. This pairs with
710 			 *    desc_reserve:D.
711 			 *
712 			 *    Memory barrier involvement:
713 			 *
714 			 *    If data_make_reusable:B reads from
715 			 *    desc_reserve:F, then data_push_tail:C reads
716 			 *    from data_push_tail:D.
717 			 *
718 			 *    Relies on:
719 			 *
720 			 *    MB from data_push_tail:D to desc_reserve:F
721 			 *       matching
722 			 *    RMB from data_make_reusable:B to
723 			 *    data_push_tail:C
724 			 *
725 			 *    Note: data_push_tail:D and desc_reserve:F can
726 			 *          be different CPUs. However, the
727 			 *          desc_reserve:F CPU (which performs the
728 			 *          full memory barrier) must have previously
729 			 *          seen data_push_tail:D.
730 			 */
731 			smp_rmb(); /* LMM(data_push_tail:B) */
732 
733 			tail_lpos_new = atomic_long_read(&data_ring->tail_lpos
734 							); /* LMM(data_push_tail:C) */
735 			if (tail_lpos_new == tail_lpos)
736 				return false;
737 
738 			/* Another CPU pushed the tail. Try again. */
739 			tail_lpos = tail_lpos_new;
740 			continue;
741 		}
742 
743 		/*
744 		 * Guarantee any descriptor states that have transitioned to
745 		 * reusable are stored before pushing the tail lpos. A full
746 		 * memory barrier is needed since other CPUs may have made
747 		 * the descriptor states reusable. This pairs with
748 		 * data_push_tail:A.
749 		 */
750 		if (atomic_long_try_cmpxchg(&data_ring->tail_lpos, &tail_lpos,
751 					    next_lpos)) { /* LMM(data_push_tail:D) */
752 			break;
753 		}
754 	}
755 
756 	return true;
757 }
758 
759 /*
760  * Advance the desc ring tail. This function advances the tail by one
761  * descriptor, thus invalidating the oldest descriptor. Before advancing
762  * the tail, the tail descriptor is made reusable and all data blocks up to
763  * and including the descriptor's data block are invalidated (i.e. the data
764  * ring tail is pushed past the data block of the descriptor being made
765  * reusable).
766  */
767 static bool desc_push_tail(struct printk_ringbuffer *rb,
768 			   unsigned long tail_id)
769 {
770 	struct prb_desc_ring *desc_ring = &rb->desc_ring;
771 	enum desc_state d_state;
772 	struct prb_desc desc;
773 
774 	d_state = desc_read(desc_ring, tail_id, &desc, NULL, NULL);
775 
776 	switch (d_state) {
777 	case desc_miss:
778 		/*
779 		 * If the ID is exactly 1 wrap behind the expected, it is
780 		 * in the process of being reserved by another writer and
781 		 * must be considered reserved.
782 		 */
783 		if (DESC_ID(atomic_long_read(&desc.state_var)) ==
784 		    DESC_ID_PREV_WRAP(desc_ring, tail_id)) {
785 			return false;
786 		}
787 
788 		/*
789 		 * The ID has changed. Another writer must have pushed the
790 		 * tail and recycled the descriptor already. Success is
791 		 * returned because the caller is only interested in the
792 		 * specified tail being pushed, which it was.
793 		 */
794 		return true;
795 	case desc_reserved:
796 	case desc_committed:
797 		return false;
798 	case desc_finalized:
799 		desc_make_reusable(desc_ring, tail_id);
800 		break;
801 	case desc_reusable:
802 		break;
803 	}
804 
805 	/*
806 	 * Data blocks must be invalidated before their associated
807 	 * descriptor can be made available for recycling. Invalidating
808 	 * them later is not possible because there is no way to trust
809 	 * data blocks once their associated descriptor is gone.
810 	 */
811 
812 	if (!data_push_tail(rb, desc.text_blk_lpos.next))
813 		return false;
814 
815 	/*
816 	 * Check the next descriptor after @tail_id before pushing the tail
817 	 * to it because the tail must always be in a finalized or reusable
818 	 * state. The implementation of prb_first_seq() relies on this.
819 	 *
820 	 * A successful read implies that the next descriptor is less than or
821 	 * equal to @head_id so there is no risk of pushing the tail past the
822 	 * head.
823 	 */
824 	d_state = desc_read(desc_ring, DESC_ID(tail_id + 1), &desc,
825 			    NULL, NULL); /* LMM(desc_push_tail:A) */
826 
827 	if (d_state == desc_finalized || d_state == desc_reusable) {
828 		/*
829 		 * Guarantee any descriptor states that have transitioned to
830 		 * reusable are stored before pushing the tail ID. This allows
831 		 * verifying the recycled descriptor state. A full memory
832 		 * barrier is needed since other CPUs may have made the
833 		 * descriptor states reusable. This pairs with desc_reserve:D.
834 		 */
835 		atomic_long_cmpxchg(&desc_ring->tail_id, tail_id,
836 				    DESC_ID(tail_id + 1)); /* LMM(desc_push_tail:B) */
837 	} else {
838 		/*
839 		 * Guarantee the last state load from desc_read() is before
840 		 * reloading @tail_id in order to see a new tail ID in the
841 		 * case that the descriptor has been recycled. This pairs
842 		 * with desc_reserve:D.
843 		 *
844 		 * Memory barrier involvement:
845 		 *
846 		 * If desc_push_tail:A reads from desc_reserve:F, then
847 		 * desc_push_tail:D reads from desc_push_tail:B.
848 		 *
849 		 * Relies on:
850 		 *
851 		 * MB from desc_push_tail:B to desc_reserve:F
852 		 *    matching
853 		 * RMB from desc_push_tail:A to desc_push_tail:D
854 		 *
855 		 * Note: desc_push_tail:B and desc_reserve:F can be different
856 		 *       CPUs. However, the desc_reserve:F CPU (which performs
857 		 *       the full memory barrier) must have previously seen
858 		 *       desc_push_tail:B.
859 		 */
860 		smp_rmb(); /* LMM(desc_push_tail:C) */
861 
862 		/*
863 		 * Re-check the tail ID. The descriptor following @tail_id is
864 		 * not in an allowed tail state. But if the tail has since
865 		 * been moved by another CPU, then it does not matter.
866 		 */
867 		if (atomic_long_read(&desc_ring->tail_id) == tail_id) /* LMM(desc_push_tail:D) */
868 			return false;
869 	}
870 
871 	return true;
872 }
873 
874 /* Reserve a new descriptor, invalidating the oldest if necessary. */
875 static bool desc_reserve(struct printk_ringbuffer *rb, unsigned long *id_out)
876 {
877 	struct prb_desc_ring *desc_ring = &rb->desc_ring;
878 	unsigned long prev_state_val;
879 	unsigned long id_prev_wrap;
880 	struct prb_desc *desc;
881 	unsigned long head_id;
882 	unsigned long id;
883 
884 	head_id = atomic_long_read(&desc_ring->head_id); /* LMM(desc_reserve:A) */
885 
886 	do {
887 		id = DESC_ID(head_id + 1);
888 		id_prev_wrap = DESC_ID_PREV_WRAP(desc_ring, id);
889 
890 		/*
891 		 * Guarantee the head ID is read before reading the tail ID.
892 		 * Since the tail ID is updated before the head ID, this
893 		 * guarantees that @id_prev_wrap is never ahead of the tail
894 		 * ID. This pairs with desc_reserve:D.
895 		 *
896 		 * Memory barrier involvement:
897 		 *
898 		 * If desc_reserve:A reads from desc_reserve:D, then
899 		 * desc_reserve:C reads from desc_push_tail:B.
900 		 *
901 		 * Relies on:
902 		 *
903 		 * MB from desc_push_tail:B to desc_reserve:D
904 		 *    matching
905 		 * RMB from desc_reserve:A to desc_reserve:C
906 		 *
907 		 * Note: desc_push_tail:B and desc_reserve:D can be different
908 		 *       CPUs. However, the desc_reserve:D CPU (which performs
909 		 *       the full memory barrier) must have previously seen
910 		 *       desc_push_tail:B.
911 		 */
912 		smp_rmb(); /* LMM(desc_reserve:B) */
913 
914 		if (id_prev_wrap == atomic_long_read(&desc_ring->tail_id
915 						    )) { /* LMM(desc_reserve:C) */
916 			/*
917 			 * Make space for the new descriptor by
918 			 * advancing the tail.
919 			 */
920 			if (!desc_push_tail(rb, id_prev_wrap))
921 				return false;
922 		}
923 
924 		/*
925 		 * 1. Guarantee the tail ID is read before validating the
926 		 *    recycled descriptor state. A read memory barrier is
927 		 *    sufficient for this. This pairs with desc_push_tail:B.
928 		 *
929 		 *    Memory barrier involvement:
930 		 *
931 		 *    If desc_reserve:C reads from desc_push_tail:B, then
932 		 *    desc_reserve:E reads from desc_make_reusable:A.
933 		 *
934 		 *    Relies on:
935 		 *
936 		 *    MB from desc_make_reusable:A to desc_push_tail:B
937 		 *       matching
938 		 *    RMB from desc_reserve:C to desc_reserve:E
939 		 *
940 		 *    Note: desc_make_reusable:A and desc_push_tail:B can be
941 		 *          different CPUs. However, the desc_push_tail:B CPU
942 		 *          (which performs the full memory barrier) must have
943 		 *          previously seen desc_make_reusable:A.
944 		 *
945 		 * 2. Guarantee the tail ID is stored before storing the head
946 		 *    ID. This pairs with desc_reserve:B.
947 		 *
948 		 * 3. Guarantee any data ring tail changes are stored before
949 		 *    recycling the descriptor. Data ring tail changes can
950 		 *    happen via desc_push_tail()->data_push_tail(). A full
951 		 *    memory barrier is needed since another CPU may have
952 		 *    pushed the data ring tails. This pairs with
953 		 *    data_push_tail:B.
954 		 *
955 		 * 4. Guarantee a new tail ID is stored before recycling the
956 		 *    descriptor. A full memory barrier is needed since
957 		 *    another CPU may have pushed the tail ID. This pairs
958 		 *    with desc_push_tail:C and this also pairs with
959 		 *    prb_first_seq:C.
960 		 *
961 		 * 5. Guarantee the head ID is stored before trying to
962 		 *    finalize the previous descriptor. This pairs with
963 		 *    _prb_commit:B.
964 		 */
965 	} while (!atomic_long_try_cmpxchg(&desc_ring->head_id, &head_id,
966 					  id)); /* LMM(desc_reserve:D) */
967 
968 	desc = to_desc(desc_ring, id);
969 
970 	/*
971 	 * If the descriptor has been recycled, verify the old state val.
972 	 * See "ABA Issues" about why this verification is performed.
973 	 */
974 	prev_state_val = atomic_long_read(&desc->state_var); /* LMM(desc_reserve:E) */
975 	if (prev_state_val &&
976 	    get_desc_state(id_prev_wrap, prev_state_val) != desc_reusable) {
977 		WARN_ON_ONCE(1);
978 		return false;
979 	}
980 
981 	/*
982 	 * Assign the descriptor a new ID and set its state to reserved.
983 	 * See "ABA Issues" about why cmpxchg() instead of set() is used.
984 	 *
985 	 * Guarantee the new descriptor ID and state is stored before making
986 	 * any other changes. A write memory barrier is sufficient for this.
987 	 * This pairs with desc_read:D.
988 	 */
989 	if (!atomic_long_try_cmpxchg(&desc->state_var, &prev_state_val,
990 			DESC_SV(id, desc_reserved))) { /* LMM(desc_reserve:F) */
991 		WARN_ON_ONCE(1);
992 		return false;
993 	}
994 
995 	/* Now data in @desc can be modified: LMM(desc_reserve:G) */
996 
997 	*id_out = id;
998 	return true;
999 }
1000 
1001 /* Determine the end of a data block. */
1002 static unsigned long get_next_lpos(struct prb_data_ring *data_ring,
1003 				   unsigned long lpos, unsigned int size)
1004 {
1005 	unsigned long begin_lpos;
1006 	unsigned long next_lpos;
1007 
1008 	begin_lpos = lpos;
1009 	next_lpos = lpos + size;
1010 
1011 	/* First check if the data block does not wrap. */
1012 	if (DATA_WRAPS(data_ring, begin_lpos) == DATA_WRAPS(data_ring, next_lpos))
1013 		return next_lpos;
1014 
1015 	/* Wrapping data blocks store their data at the beginning. */
1016 	return (DATA_THIS_WRAP_START_LPOS(data_ring, next_lpos) + size);
1017 }
1018 
1019 /*
1020  * Allocate a new data block, invalidating the oldest data block(s)
1021  * if necessary. This function also associates the data block with
1022  * a specified descriptor.
1023  */
1024 static char *data_alloc(struct printk_ringbuffer *rb, unsigned int size,
1025 			struct prb_data_blk_lpos *blk_lpos, unsigned long id)
1026 {
1027 	struct prb_data_ring *data_ring = &rb->text_data_ring;
1028 	struct prb_data_block *blk;
1029 	unsigned long begin_lpos;
1030 	unsigned long next_lpos;
1031 
1032 	if (size == 0) {
1033 		/* Specify a data-less block. */
1034 		blk_lpos->begin = NO_LPOS;
1035 		blk_lpos->next = NO_LPOS;
1036 		return NULL;
1037 	}
1038 
1039 	size = to_blk_size(size);
1040 
1041 	begin_lpos = atomic_long_read(&data_ring->head_lpos);
1042 
1043 	do {
1044 		next_lpos = get_next_lpos(data_ring, begin_lpos, size);
1045 
1046 		if (!data_push_tail(rb, next_lpos - DATA_SIZE(data_ring))) {
1047 			/* Failed to allocate, specify a data-less block. */
1048 			blk_lpos->begin = FAILED_LPOS;
1049 			blk_lpos->next = FAILED_LPOS;
1050 			return NULL;
1051 		}
1052 
1053 		/*
1054 		 * 1. Guarantee any descriptor states that have transitioned
1055 		 *    to reusable are stored before modifying the newly
1056 		 *    allocated data area. A full memory barrier is needed
1057 		 *    since other CPUs may have made the descriptor states
1058 		 *    reusable. See data_push_tail:A about why the reusable
1059 		 *    states are visible. This pairs with desc_read:D.
1060 		 *
1061 		 * 2. Guarantee any updated tail lpos is stored before
1062 		 *    modifying the newly allocated data area. Another CPU may
1063 		 *    be in data_make_reusable() and is reading a block ID
1064 		 *    from this area. data_make_reusable() can handle reading
1065 		 *    a garbage block ID value, but then it must be able to
1066 		 *    load a new tail lpos. A full memory barrier is needed
1067 		 *    since other CPUs may have updated the tail lpos. This
1068 		 *    pairs with data_push_tail:B.
1069 		 */
1070 	} while (!atomic_long_try_cmpxchg(&data_ring->head_lpos, &begin_lpos,
1071 					  next_lpos)); /* LMM(data_alloc:A) */
1072 
1073 	blk = to_block(data_ring, begin_lpos);
1074 	blk->id = id; /* LMM(data_alloc:B) */
1075 
1076 	if (DATA_WRAPS(data_ring, begin_lpos) != DATA_WRAPS(data_ring, next_lpos)) {
1077 		/* Wrapping data blocks store their data at the beginning. */
1078 		blk = to_block(data_ring, 0);
1079 
1080 		/*
1081 		 * Store the ID on the wrapped block for consistency.
1082 		 * The printk_ringbuffer does not actually use it.
1083 		 */
1084 		blk->id = id;
1085 	}
1086 
1087 	blk_lpos->begin = begin_lpos;
1088 	blk_lpos->next = next_lpos;
1089 
1090 	return &blk->data[0];
1091 }
1092 
1093 /*
1094  * Try to resize an existing data block associated with the descriptor
1095  * specified by @id. If the resized data block should become wrapped, it
1096  * copies the old data to the new data block. If @size yields a data block
1097  * with the same or less size, the data block is left as is.
1098  *
1099  * Fail if this is not the last allocated data block or if there is not
1100  * enough space or it is not possible make enough space.
1101  *
1102  * Return a pointer to the beginning of the entire data buffer or NULL on
1103  * failure.
1104  */
1105 static char *data_realloc(struct printk_ringbuffer *rb, unsigned int size,
1106 			  struct prb_data_blk_lpos *blk_lpos, unsigned long id)
1107 {
1108 	struct prb_data_ring *data_ring = &rb->text_data_ring;
1109 	struct prb_data_block *blk;
1110 	unsigned long head_lpos;
1111 	unsigned long next_lpos;
1112 	bool wrapped;
1113 
1114 	/* Reallocation only works if @blk_lpos is the newest data block. */
1115 	head_lpos = atomic_long_read(&data_ring->head_lpos);
1116 	if (head_lpos != blk_lpos->next)
1117 		return NULL;
1118 
1119 	/* Keep track if @blk_lpos was a wrapping data block. */
1120 	wrapped = (DATA_WRAPS(data_ring, blk_lpos->begin) != DATA_WRAPS(data_ring, blk_lpos->next));
1121 
1122 	size = to_blk_size(size);
1123 
1124 	next_lpos = get_next_lpos(data_ring, blk_lpos->begin, size);
1125 
1126 	/* If the data block does not increase, there is nothing to do. */
1127 	if (head_lpos - next_lpos < DATA_SIZE(data_ring)) {
1128 		if (wrapped)
1129 			blk = to_block(data_ring, 0);
1130 		else
1131 			blk = to_block(data_ring, blk_lpos->begin);
1132 		return &blk->data[0];
1133 	}
1134 
1135 	if (!data_push_tail(rb, next_lpos - DATA_SIZE(data_ring)))
1136 		return NULL;
1137 
1138 	/* The memory barrier involvement is the same as data_alloc:A. */
1139 	if (!atomic_long_try_cmpxchg(&data_ring->head_lpos, &head_lpos,
1140 				     next_lpos)) { /* LMM(data_realloc:A) */
1141 		return NULL;
1142 	}
1143 
1144 	blk = to_block(data_ring, blk_lpos->begin);
1145 
1146 	if (DATA_WRAPS(data_ring, blk_lpos->begin) != DATA_WRAPS(data_ring, next_lpos)) {
1147 		struct prb_data_block *old_blk = blk;
1148 
1149 		/* Wrapping data blocks store their data at the beginning. */
1150 		blk = to_block(data_ring, 0);
1151 
1152 		/*
1153 		 * Store the ID on the wrapped block for consistency.
1154 		 * The printk_ringbuffer does not actually use it.
1155 		 */
1156 		blk->id = id;
1157 
1158 		if (!wrapped) {
1159 			/*
1160 			 * Since the allocated space is now in the newly
1161 			 * created wrapping data block, copy the content
1162 			 * from the old data block.
1163 			 */
1164 			memcpy(&blk->data[0], &old_blk->data[0],
1165 			       (blk_lpos->next - blk_lpos->begin) - sizeof(blk->id));
1166 		}
1167 	}
1168 
1169 	blk_lpos->next = next_lpos;
1170 
1171 	return &blk->data[0];
1172 }
1173 
1174 /* Return the number of bytes used by a data block. */
1175 static unsigned int space_used(struct prb_data_ring *data_ring,
1176 			       struct prb_data_blk_lpos *blk_lpos)
1177 {
1178 	/* Data-less blocks take no space. */
1179 	if (BLK_DATALESS(blk_lpos))
1180 		return 0;
1181 
1182 	if (DATA_WRAPS(data_ring, blk_lpos->begin) == DATA_WRAPS(data_ring, blk_lpos->next)) {
1183 		/* Data block does not wrap. */
1184 		return (DATA_INDEX(data_ring, blk_lpos->next) -
1185 			DATA_INDEX(data_ring, blk_lpos->begin));
1186 	}
1187 
1188 	/*
1189 	 * For wrapping data blocks, the trailing (wasted) space is
1190 	 * also counted.
1191 	 */
1192 	return (DATA_INDEX(data_ring, blk_lpos->next) +
1193 		DATA_SIZE(data_ring) - DATA_INDEX(data_ring, blk_lpos->begin));
1194 }
1195 
1196 /*
1197  * Given @blk_lpos, return a pointer to the writer data from the data block
1198  * and calculate the size of the data part. A NULL pointer is returned if
1199  * @blk_lpos specifies values that could never be legal.
1200  *
1201  * This function (used by readers) performs strict validation on the lpos
1202  * values to possibly detect bugs in the writer code. A WARN_ON_ONCE() is
1203  * triggered if an internal error is detected.
1204  */
1205 static const char *get_data(struct prb_data_ring *data_ring,
1206 			    struct prb_data_blk_lpos *blk_lpos,
1207 			    unsigned int *data_size)
1208 {
1209 	struct prb_data_block *db;
1210 
1211 	/* Data-less data block description. */
1212 	if (BLK_DATALESS(blk_lpos)) {
1213 		if (blk_lpos->begin == NO_LPOS && blk_lpos->next == NO_LPOS) {
1214 			*data_size = 0;
1215 			return "";
1216 		}
1217 		return NULL;
1218 	}
1219 
1220 	/* Regular data block: @begin less than @next and in same wrap. */
1221 	if (DATA_WRAPS(data_ring, blk_lpos->begin) == DATA_WRAPS(data_ring, blk_lpos->next) &&
1222 	    blk_lpos->begin < blk_lpos->next) {
1223 		db = to_block(data_ring, blk_lpos->begin);
1224 		*data_size = blk_lpos->next - blk_lpos->begin;
1225 
1226 	/* Wrapping data block: @begin is one wrap behind @next. */
1227 	} else if (DATA_WRAPS(data_ring, blk_lpos->begin + DATA_SIZE(data_ring)) ==
1228 		   DATA_WRAPS(data_ring, blk_lpos->next)) {
1229 		db = to_block(data_ring, 0);
1230 		*data_size = DATA_INDEX(data_ring, blk_lpos->next);
1231 
1232 	/* Illegal block description. */
1233 	} else {
1234 		WARN_ON_ONCE(1);
1235 		return NULL;
1236 	}
1237 
1238 	/* A valid data block will always be aligned to the ID size. */
1239 	if (WARN_ON_ONCE(blk_lpos->begin != ALIGN(blk_lpos->begin, sizeof(db->id))) ||
1240 	    WARN_ON_ONCE(blk_lpos->next != ALIGN(blk_lpos->next, sizeof(db->id)))) {
1241 		return NULL;
1242 	}
1243 
1244 	/* A valid data block will always have at least an ID. */
1245 	if (WARN_ON_ONCE(*data_size < sizeof(db->id)))
1246 		return NULL;
1247 
1248 	/* Subtract block ID space from size to reflect data size. */
1249 	*data_size -= sizeof(db->id);
1250 
1251 	return &db->data[0];
1252 }
1253 
1254 /*
1255  * Attempt to transition the newest descriptor from committed back to reserved
1256  * so that the record can be modified by a writer again. This is only possible
1257  * if the descriptor is not yet finalized and the provided @caller_id matches.
1258  */
1259 static struct prb_desc *desc_reopen_last(struct prb_desc_ring *desc_ring,
1260 					 u32 caller_id, unsigned long *id_out)
1261 {
1262 	unsigned long prev_state_val;
1263 	enum desc_state d_state;
1264 	struct prb_desc desc;
1265 	struct prb_desc *d;
1266 	unsigned long id;
1267 	u32 cid;
1268 
1269 	id = atomic_long_read(&desc_ring->head_id);
1270 
1271 	/*
1272 	 * To reduce unnecessarily reopening, first check if the descriptor
1273 	 * state and caller ID are correct.
1274 	 */
1275 	d_state = desc_read(desc_ring, id, &desc, NULL, &cid);
1276 	if (d_state != desc_committed || cid != caller_id)
1277 		return NULL;
1278 
1279 	d = to_desc(desc_ring, id);
1280 
1281 	prev_state_val = DESC_SV(id, desc_committed);
1282 
1283 	/*
1284 	 * Guarantee the reserved state is stored before reading any
1285 	 * record data. A full memory barrier is needed because @state_var
1286 	 * modification is followed by reading. This pairs with _prb_commit:B.
1287 	 *
1288 	 * Memory barrier involvement:
1289 	 *
1290 	 * If desc_reopen_last:A reads from _prb_commit:B, then
1291 	 * prb_reserve_in_last:A reads from _prb_commit:A.
1292 	 *
1293 	 * Relies on:
1294 	 *
1295 	 * WMB from _prb_commit:A to _prb_commit:B
1296 	 *    matching
1297 	 * MB If desc_reopen_last:A to prb_reserve_in_last:A
1298 	 */
1299 	if (!atomic_long_try_cmpxchg(&d->state_var, &prev_state_val,
1300 			DESC_SV(id, desc_reserved))) { /* LMM(desc_reopen_last:A) */
1301 		return NULL;
1302 	}
1303 
1304 	*id_out = id;
1305 	return d;
1306 }
1307 
1308 /**
1309  * prb_reserve_in_last() - Re-reserve and extend the space in the ringbuffer
1310  *                         used by the newest record.
1311  *
1312  * @e:         The entry structure to setup.
1313  * @rb:        The ringbuffer to re-reserve and extend data in.
1314  * @r:         The record structure to allocate buffers for.
1315  * @caller_id: The caller ID of the caller (reserving writer).
1316  * @max_size:  Fail if the extended size would be greater than this.
1317  *
1318  * This is the public function available to writers to re-reserve and extend
1319  * data.
1320  *
1321  * The writer specifies the text size to extend (not the new total size) by
1322  * setting the @text_buf_size field of @r. To ensure proper initialization
1323  * of @r, prb_rec_init_wr() should be used.
1324  *
1325  * This function will fail if @caller_id does not match the caller ID of the
1326  * newest record. In that case the caller must reserve new data using
1327  * prb_reserve().
1328  *
1329  * Context: Any context. Disables local interrupts on success.
1330  * Return: true if text data could be extended, otherwise false.
1331  *
1332  * On success:
1333  *
1334  *   - @r->text_buf points to the beginning of the entire text buffer.
1335  *
1336  *   - @r->text_buf_size is set to the new total size of the buffer.
1337  *
1338  *   - @r->info is not touched so that @r->info->text_len could be used
1339  *     to append the text.
1340  *
1341  *   - prb_record_text_space() can be used on @e to query the new
1342  *     actually used space.
1343  *
1344  * Important: All @r->info fields will already be set with the current values
1345  *            for the record. I.e. @r->info->text_len will be less than
1346  *            @text_buf_size. Writers can use @r->info->text_len to know
1347  *            where concatenation begins and writers should update
1348  *            @r->info->text_len after concatenating.
1349  */
1350 bool prb_reserve_in_last(struct prb_reserved_entry *e, struct printk_ringbuffer *rb,
1351 			 struct printk_record *r, u32 caller_id, unsigned int max_size)
1352 {
1353 	struct prb_desc_ring *desc_ring = &rb->desc_ring;
1354 	struct printk_info *info;
1355 	unsigned int data_size;
1356 	struct prb_desc *d;
1357 	unsigned long id;
1358 
1359 	local_irq_save(e->irqflags);
1360 
1361 	/* Transition the newest descriptor back to the reserved state. */
1362 	d = desc_reopen_last(desc_ring, caller_id, &id);
1363 	if (!d) {
1364 		local_irq_restore(e->irqflags);
1365 		goto fail_reopen;
1366 	}
1367 
1368 	/* Now the writer has exclusive access: LMM(prb_reserve_in_last:A) */
1369 
1370 	info = to_info(desc_ring, id);
1371 
1372 	/*
1373 	 * Set the @e fields here so that prb_commit() can be used if
1374 	 * anything fails from now on.
1375 	 */
1376 	e->rb = rb;
1377 	e->id = id;
1378 
1379 	/*
1380 	 * desc_reopen_last() checked the caller_id, but there was no
1381 	 * exclusive access at that point. The descriptor may have
1382 	 * changed since then.
1383 	 */
1384 	if (caller_id != info->caller_id)
1385 		goto fail;
1386 
1387 	if (BLK_DATALESS(&d->text_blk_lpos)) {
1388 		if (WARN_ON_ONCE(info->text_len != 0)) {
1389 			pr_warn_once("wrong text_len value (%hu, expecting 0)\n",
1390 				     info->text_len);
1391 			info->text_len = 0;
1392 		}
1393 
1394 		if (!data_check_size(&rb->text_data_ring, r->text_buf_size))
1395 			goto fail;
1396 
1397 		if (r->text_buf_size > max_size)
1398 			goto fail;
1399 
1400 		r->text_buf = data_alloc(rb, r->text_buf_size,
1401 					 &d->text_blk_lpos, id);
1402 	} else {
1403 		if (!get_data(&rb->text_data_ring, &d->text_blk_lpos, &data_size))
1404 			goto fail;
1405 
1406 		/*
1407 		 * Increase the buffer size to include the original size. If
1408 		 * the meta data (@text_len) is not sane, use the full data
1409 		 * block size.
1410 		 */
1411 		if (WARN_ON_ONCE(info->text_len > data_size)) {
1412 			pr_warn_once("wrong text_len value (%hu, expecting <=%u)\n",
1413 				     info->text_len, data_size);
1414 			info->text_len = data_size;
1415 		}
1416 		r->text_buf_size += info->text_len;
1417 
1418 		if (!data_check_size(&rb->text_data_ring, r->text_buf_size))
1419 			goto fail;
1420 
1421 		if (r->text_buf_size > max_size)
1422 			goto fail;
1423 
1424 		r->text_buf = data_realloc(rb, r->text_buf_size,
1425 					   &d->text_blk_lpos, id);
1426 	}
1427 	if (r->text_buf_size && !r->text_buf)
1428 		goto fail;
1429 
1430 	r->info = info;
1431 
1432 	e->text_space = space_used(&rb->text_data_ring, &d->text_blk_lpos);
1433 
1434 	return true;
1435 fail:
1436 	prb_commit(e);
1437 	/* prb_commit() re-enabled interrupts. */
1438 fail_reopen:
1439 	/* Make it clear to the caller that the re-reserve failed. */
1440 	memset(r, 0, sizeof(*r));
1441 	return false;
1442 }
1443 
1444 /*
1445  * Attempt to finalize a specified descriptor. If this fails, the descriptor
1446  * is either already final or it will finalize itself when the writer commits.
1447  */
1448 static void desc_make_final(struct prb_desc_ring *desc_ring, unsigned long id)
1449 {
1450 	unsigned long prev_state_val = DESC_SV(id, desc_committed);
1451 	struct prb_desc *d = to_desc(desc_ring, id);
1452 
1453 	atomic_long_cmpxchg_relaxed(&d->state_var, prev_state_val,
1454 			DESC_SV(id, desc_finalized)); /* LMM(desc_make_final:A) */
1455 
1456 	/* Best effort to remember the last finalized @id. */
1457 	atomic_long_set(&desc_ring->last_finalized_id, id);
1458 }
1459 
1460 /**
1461  * prb_reserve() - Reserve space in the ringbuffer.
1462  *
1463  * @e:  The entry structure to setup.
1464  * @rb: The ringbuffer to reserve data in.
1465  * @r:  The record structure to allocate buffers for.
1466  *
1467  * This is the public function available to writers to reserve data.
1468  *
1469  * The writer specifies the text size to reserve by setting the
1470  * @text_buf_size field of @r. To ensure proper initialization of @r,
1471  * prb_rec_init_wr() should be used.
1472  *
1473  * Context: Any context. Disables local interrupts on success.
1474  * Return: true if at least text data could be allocated, otherwise false.
1475  *
1476  * On success, the fields @info and @text_buf of @r will be set by this
1477  * function and should be filled in by the writer before committing. Also
1478  * on success, prb_record_text_space() can be used on @e to query the actual
1479  * space used for the text data block.
1480  *
1481  * Important: @info->text_len needs to be set correctly by the writer in
1482  *            order for data to be readable and/or extended. Its value
1483  *            is initialized to 0.
1484  */
1485 bool prb_reserve(struct prb_reserved_entry *e, struct printk_ringbuffer *rb,
1486 		 struct printk_record *r)
1487 {
1488 	struct prb_desc_ring *desc_ring = &rb->desc_ring;
1489 	struct printk_info *info;
1490 	struct prb_desc *d;
1491 	unsigned long id;
1492 	u64 seq;
1493 
1494 	if (!data_check_size(&rb->text_data_ring, r->text_buf_size))
1495 		goto fail;
1496 
1497 	/*
1498 	 * Descriptors in the reserved state act as blockers to all further
1499 	 * reservations once the desc_ring has fully wrapped. Disable
1500 	 * interrupts during the reserve/commit window in order to minimize
1501 	 * the likelihood of this happening.
1502 	 */
1503 	local_irq_save(e->irqflags);
1504 
1505 	if (!desc_reserve(rb, &id)) {
1506 		/* Descriptor reservation failures are tracked. */
1507 		atomic_long_inc(&rb->fail);
1508 		local_irq_restore(e->irqflags);
1509 		goto fail;
1510 	}
1511 
1512 	d = to_desc(desc_ring, id);
1513 	info = to_info(desc_ring, id);
1514 
1515 	/*
1516 	 * All @info fields (except @seq) are cleared and must be filled in
1517 	 * by the writer. Save @seq before clearing because it is used to
1518 	 * determine the new sequence number.
1519 	 */
1520 	seq = info->seq;
1521 	memset(info, 0, sizeof(*info));
1522 
1523 	/*
1524 	 * Set the @e fields here so that prb_commit() can be used if
1525 	 * text data allocation fails.
1526 	 */
1527 	e->rb = rb;
1528 	e->id = id;
1529 
1530 	/*
1531 	 * Initialize the sequence number if it has "never been set".
1532 	 * Otherwise just increment it by a full wrap.
1533 	 *
1534 	 * @seq is considered "never been set" if it has a value of 0,
1535 	 * _except_ for @infos[0], which was specially setup by the ringbuffer
1536 	 * initializer and therefore is always considered as set.
1537 	 *
1538 	 * See the "Bootstrap" comment block in printk_ringbuffer.h for
1539 	 * details about how the initializer bootstraps the descriptors.
1540 	 */
1541 	if (seq == 0 && DESC_INDEX(desc_ring, id) != 0)
1542 		info->seq = DESC_INDEX(desc_ring, id);
1543 	else
1544 		info->seq = seq + DESCS_COUNT(desc_ring);
1545 
1546 	/*
1547 	 * New data is about to be reserved. Once that happens, previous
1548 	 * descriptors are no longer able to be extended. Finalize the
1549 	 * previous descriptor now so that it can be made available to
1550 	 * readers. (For seq==0 there is no previous descriptor.)
1551 	 */
1552 	if (info->seq > 0)
1553 		desc_make_final(desc_ring, DESC_ID(id - 1));
1554 
1555 	r->text_buf = data_alloc(rb, r->text_buf_size, &d->text_blk_lpos, id);
1556 	/* If text data allocation fails, a data-less record is committed. */
1557 	if (r->text_buf_size && !r->text_buf) {
1558 		prb_commit(e);
1559 		/* prb_commit() re-enabled interrupts. */
1560 		goto fail;
1561 	}
1562 
1563 	r->info = info;
1564 
1565 	/* Record full text space used by record. */
1566 	e->text_space = space_used(&rb->text_data_ring, &d->text_blk_lpos);
1567 
1568 	return true;
1569 fail:
1570 	/* Make it clear to the caller that the reserve failed. */
1571 	memset(r, 0, sizeof(*r));
1572 	return false;
1573 }
1574 
1575 /* Commit the data (possibly finalizing it) and restore interrupts. */
1576 static void _prb_commit(struct prb_reserved_entry *e, unsigned long state_val)
1577 {
1578 	struct prb_desc_ring *desc_ring = &e->rb->desc_ring;
1579 	struct prb_desc *d = to_desc(desc_ring, e->id);
1580 	unsigned long prev_state_val = DESC_SV(e->id, desc_reserved);
1581 
1582 	/* Now the writer has finished all writing: LMM(_prb_commit:A) */
1583 
1584 	/*
1585 	 * Set the descriptor as committed. See "ABA Issues" about why
1586 	 * cmpxchg() instead of set() is used.
1587 	 *
1588 	 * 1  Guarantee all record data is stored before the descriptor state
1589 	 *    is stored as committed. A write memory barrier is sufficient
1590 	 *    for this. This pairs with desc_read:B and desc_reopen_last:A.
1591 	 *
1592 	 * 2. Guarantee the descriptor state is stored as committed before
1593 	 *    re-checking the head ID in order to possibly finalize this
1594 	 *    descriptor. This pairs with desc_reserve:D.
1595 	 *
1596 	 *    Memory barrier involvement:
1597 	 *
1598 	 *    If prb_commit:A reads from desc_reserve:D, then
1599 	 *    desc_make_final:A reads from _prb_commit:B.
1600 	 *
1601 	 *    Relies on:
1602 	 *
1603 	 *    MB _prb_commit:B to prb_commit:A
1604 	 *       matching
1605 	 *    MB desc_reserve:D to desc_make_final:A
1606 	 */
1607 	if (!atomic_long_try_cmpxchg(&d->state_var, &prev_state_val,
1608 			DESC_SV(e->id, state_val))) { /* LMM(_prb_commit:B) */
1609 		WARN_ON_ONCE(1);
1610 	}
1611 
1612 	/* Restore interrupts, the reserve/commit window is finished. */
1613 	local_irq_restore(e->irqflags);
1614 }
1615 
1616 /**
1617  * prb_commit() - Commit (previously reserved) data to the ringbuffer.
1618  *
1619  * @e: The entry containing the reserved data information.
1620  *
1621  * This is the public function available to writers to commit data.
1622  *
1623  * Note that the data is not yet available to readers until it is finalized.
1624  * Finalizing happens automatically when space for the next record is
1625  * reserved.
1626  *
1627  * See prb_final_commit() for a version of this function that finalizes
1628  * immediately.
1629  *
1630  * Context: Any context. Enables local interrupts.
1631  */
1632 void prb_commit(struct prb_reserved_entry *e)
1633 {
1634 	struct prb_desc_ring *desc_ring = &e->rb->desc_ring;
1635 	unsigned long head_id;
1636 
1637 	_prb_commit(e, desc_committed);
1638 
1639 	/*
1640 	 * If this descriptor is no longer the head (i.e. a new record has
1641 	 * been allocated), extending the data for this record is no longer
1642 	 * allowed and therefore it must be finalized.
1643 	 */
1644 	head_id = atomic_long_read(&desc_ring->head_id); /* LMM(prb_commit:A) */
1645 	if (head_id != e->id)
1646 		desc_make_final(desc_ring, e->id);
1647 }
1648 
1649 /**
1650  * prb_final_commit() - Commit and finalize (previously reserved) data to
1651  *                      the ringbuffer.
1652  *
1653  * @e: The entry containing the reserved data information.
1654  *
1655  * This is the public function available to writers to commit+finalize data.
1656  *
1657  * By finalizing, the data is made immediately available to readers.
1658  *
1659  * This function should only be used if there are no intentions of extending
1660  * this data using prb_reserve_in_last().
1661  *
1662  * Context: Any context. Enables local interrupts.
1663  */
1664 void prb_final_commit(struct prb_reserved_entry *e)
1665 {
1666 	struct prb_desc_ring *desc_ring = &e->rb->desc_ring;
1667 
1668 	_prb_commit(e, desc_finalized);
1669 
1670 	/* Best effort to remember the last finalized @id. */
1671 	atomic_long_set(&desc_ring->last_finalized_id, e->id);
1672 }
1673 
1674 /*
1675  * Count the number of lines in provided text. All text has at least 1 line
1676  * (even if @text_size is 0). Each '\n' processed is counted as an additional
1677  * line.
1678  */
1679 static unsigned int count_lines(const char *text, unsigned int text_size)
1680 {
1681 	unsigned int next_size = text_size;
1682 	unsigned int line_count = 1;
1683 	const char *next = text;
1684 
1685 	while (next_size) {
1686 		next = memchr(next, '\n', next_size);
1687 		if (!next)
1688 			break;
1689 		line_count++;
1690 		next++;
1691 		next_size = text_size - (next - text);
1692 	}
1693 
1694 	return line_count;
1695 }
1696 
1697 /*
1698  * Given @blk_lpos, copy an expected @len of data into the provided buffer.
1699  * If @line_count is provided, count the number of lines in the data.
1700  *
1701  * This function (used by readers) performs strict validation on the data
1702  * size to possibly detect bugs in the writer code. A WARN_ON_ONCE() is
1703  * triggered if an internal error is detected.
1704  */
1705 static bool copy_data(struct prb_data_ring *data_ring,
1706 		      struct prb_data_blk_lpos *blk_lpos, u16 len, char *buf,
1707 		      unsigned int buf_size, unsigned int *line_count)
1708 {
1709 	unsigned int data_size;
1710 	const char *data;
1711 
1712 	/* Caller might not want any data. */
1713 	if ((!buf || !buf_size) && !line_count)
1714 		return true;
1715 
1716 	data = get_data(data_ring, blk_lpos, &data_size);
1717 	if (!data)
1718 		return false;
1719 
1720 	/*
1721 	 * Actual cannot be less than expected. It can be more than expected
1722 	 * because of the trailing alignment padding.
1723 	 *
1724 	 * Note that invalid @len values can occur because the caller loads
1725 	 * the value during an allowed data race.
1726 	 */
1727 	if (data_size < (unsigned int)len)
1728 		return false;
1729 
1730 	/* Caller interested in the line count? */
1731 	if (line_count)
1732 		*line_count = count_lines(data, len);
1733 
1734 	/* Caller interested in the data content? */
1735 	if (!buf || !buf_size)
1736 		return true;
1737 
1738 	data_size = min_t(u16, buf_size, len);
1739 
1740 	memcpy(&buf[0], data, data_size); /* LMM(copy_data:A) */
1741 	return true;
1742 }
1743 
1744 /*
1745  * This is an extended version of desc_read(). It gets a copy of a specified
1746  * descriptor. However, it also verifies that the record is finalized and has
1747  * the sequence number @seq. On success, 0 is returned.
1748  *
1749  * Error return values:
1750  * -EINVAL: A finalized record with sequence number @seq does not exist.
1751  * -ENOENT: A finalized record with sequence number @seq exists, but its data
1752  *          is not available. This is a valid record, so readers should
1753  *          continue with the next record.
1754  */
1755 static int desc_read_finalized_seq(struct prb_desc_ring *desc_ring,
1756 				   unsigned long id, u64 seq,
1757 				   struct prb_desc *desc_out)
1758 {
1759 	struct prb_data_blk_lpos *blk_lpos = &desc_out->text_blk_lpos;
1760 	enum desc_state d_state;
1761 	u64 s;
1762 
1763 	d_state = desc_read(desc_ring, id, desc_out, &s, NULL);
1764 
1765 	/*
1766 	 * An unexpected @id (desc_miss) or @seq mismatch means the record
1767 	 * does not exist. A descriptor in the reserved or committed state
1768 	 * means the record does not yet exist for the reader.
1769 	 */
1770 	if (d_state == desc_miss ||
1771 	    d_state == desc_reserved ||
1772 	    d_state == desc_committed ||
1773 	    s != seq) {
1774 		return -EINVAL;
1775 	}
1776 
1777 	/*
1778 	 * A descriptor in the reusable state may no longer have its data
1779 	 * available; report it as existing but with lost data. Or the record
1780 	 * may actually be a record with lost data.
1781 	 */
1782 	if (d_state == desc_reusable ||
1783 	    (blk_lpos->begin == FAILED_LPOS && blk_lpos->next == FAILED_LPOS)) {
1784 		return -ENOENT;
1785 	}
1786 
1787 	return 0;
1788 }
1789 
1790 /*
1791  * Copy the ringbuffer data from the record with @seq to the provided
1792  * @r buffer. On success, 0 is returned.
1793  *
1794  * See desc_read_finalized_seq() for error return values.
1795  */
1796 static int prb_read(struct printk_ringbuffer *rb, u64 seq,
1797 		    struct printk_record *r, unsigned int *line_count)
1798 {
1799 	struct prb_desc_ring *desc_ring = &rb->desc_ring;
1800 	struct printk_info *info = to_info(desc_ring, seq);
1801 	struct prb_desc *rdesc = to_desc(desc_ring, seq);
1802 	atomic_long_t *state_var = &rdesc->state_var;
1803 	struct prb_desc desc;
1804 	unsigned long id;
1805 	int err;
1806 
1807 	/* Extract the ID, used to specify the descriptor to read. */
1808 	id = DESC_ID(atomic_long_read(state_var));
1809 
1810 	/* Get a local copy of the correct descriptor (if available). */
1811 	err = desc_read_finalized_seq(desc_ring, id, seq, &desc);
1812 
1813 	/*
1814 	 * If @r is NULL, the caller is only interested in the availability
1815 	 * of the record.
1816 	 */
1817 	if (err || !r)
1818 		return err;
1819 
1820 	/* If requested, copy meta data. */
1821 	if (r->info)
1822 		memcpy(r->info, info, sizeof(*(r->info)));
1823 
1824 	/* Copy text data. If it fails, this is a data-less record. */
1825 	if (!copy_data(&rb->text_data_ring, &desc.text_blk_lpos, info->text_len,
1826 		       r->text_buf, r->text_buf_size, line_count)) {
1827 		return -ENOENT;
1828 	}
1829 
1830 	/* Ensure the record is still finalized and has the same @seq. */
1831 	return desc_read_finalized_seq(desc_ring, id, seq, &desc);
1832 }
1833 
1834 /* Get the sequence number of the tail descriptor. */
1835 static u64 prb_first_seq(struct printk_ringbuffer *rb)
1836 {
1837 	struct prb_desc_ring *desc_ring = &rb->desc_ring;
1838 	enum desc_state d_state;
1839 	struct prb_desc desc;
1840 	unsigned long id;
1841 	u64 seq;
1842 
1843 	for (;;) {
1844 		id = atomic_long_read(&rb->desc_ring.tail_id); /* LMM(prb_first_seq:A) */
1845 
1846 		d_state = desc_read(desc_ring, id, &desc, &seq, NULL); /* LMM(prb_first_seq:B) */
1847 
1848 		/*
1849 		 * This loop will not be infinite because the tail is
1850 		 * _always_ in the finalized or reusable state.
1851 		 */
1852 		if (d_state == desc_finalized || d_state == desc_reusable)
1853 			break;
1854 
1855 		/*
1856 		 * Guarantee the last state load from desc_read() is before
1857 		 * reloading @tail_id in order to see a new tail in the case
1858 		 * that the descriptor has been recycled. This pairs with
1859 		 * desc_reserve:D.
1860 		 *
1861 		 * Memory barrier involvement:
1862 		 *
1863 		 * If prb_first_seq:B reads from desc_reserve:F, then
1864 		 * prb_first_seq:A reads from desc_push_tail:B.
1865 		 *
1866 		 * Relies on:
1867 		 *
1868 		 * MB from desc_push_tail:B to desc_reserve:F
1869 		 *    matching
1870 		 * RMB prb_first_seq:B to prb_first_seq:A
1871 		 */
1872 		smp_rmb(); /* LMM(prb_first_seq:C) */
1873 	}
1874 
1875 	return seq;
1876 }
1877 
1878 /*
1879  * Non-blocking read of a record. Updates @seq to the last finalized record
1880  * (which may have no data available).
1881  *
1882  * See the description of prb_read_valid() and prb_read_valid_info()
1883  * for details.
1884  */
1885 static bool _prb_read_valid(struct printk_ringbuffer *rb, u64 *seq,
1886 			    struct printk_record *r, unsigned int *line_count)
1887 {
1888 	u64 tail_seq;
1889 	int err;
1890 
1891 	while ((err = prb_read(rb, *seq, r, line_count))) {
1892 		tail_seq = prb_first_seq(rb);
1893 
1894 		if (*seq < tail_seq) {
1895 			/*
1896 			 * Behind the tail. Catch up and try again. This
1897 			 * can happen for -ENOENT and -EINVAL cases.
1898 			 */
1899 			*seq = tail_seq;
1900 
1901 		} else if (err == -ENOENT) {
1902 			/* Record exists, but no data available. Skip. */
1903 			(*seq)++;
1904 
1905 		} else {
1906 			/* Non-existent/non-finalized record. Must stop. */
1907 			return false;
1908 		}
1909 	}
1910 
1911 	return true;
1912 }
1913 
1914 /**
1915  * prb_read_valid() - Non-blocking read of a requested record or (if gone)
1916  *                    the next available record.
1917  *
1918  * @rb:  The ringbuffer to read from.
1919  * @seq: The sequence number of the record to read.
1920  * @r:   A record data buffer to store the read record to.
1921  *
1922  * This is the public function available to readers to read a record.
1923  *
1924  * The reader provides the @info and @text_buf buffers of @r to be
1925  * filled in. Any of the buffer pointers can be set to NULL if the reader
1926  * is not interested in that data. To ensure proper initialization of @r,
1927  * prb_rec_init_rd() should be used.
1928  *
1929  * Context: Any context.
1930  * Return: true if a record was read, otherwise false.
1931  *
1932  * On success, the reader must check r->info.seq to see which record was
1933  * actually read. This allows the reader to detect dropped records.
1934  *
1935  * Failure means @seq refers to a not yet written record.
1936  */
1937 bool prb_read_valid(struct printk_ringbuffer *rb, u64 seq,
1938 		    struct printk_record *r)
1939 {
1940 	return _prb_read_valid(rb, &seq, r, NULL);
1941 }
1942 
1943 /**
1944  * prb_read_valid_info() - Non-blocking read of meta data for a requested
1945  *                         record or (if gone) the next available record.
1946  *
1947  * @rb:         The ringbuffer to read from.
1948  * @seq:        The sequence number of the record to read.
1949  * @info:       A buffer to store the read record meta data to.
1950  * @line_count: A buffer to store the number of lines in the record text.
1951  *
1952  * This is the public function available to readers to read only the
1953  * meta data of a record.
1954  *
1955  * The reader provides the @info, @line_count buffers to be filled in.
1956  * Either of the buffer pointers can be set to NULL if the reader is not
1957  * interested in that data.
1958  *
1959  * Context: Any context.
1960  * Return: true if a record's meta data was read, otherwise false.
1961  *
1962  * On success, the reader must check info->seq to see which record meta data
1963  * was actually read. This allows the reader to detect dropped records.
1964  *
1965  * Failure means @seq refers to a not yet written record.
1966  */
1967 bool prb_read_valid_info(struct printk_ringbuffer *rb, u64 seq,
1968 			 struct printk_info *info, unsigned int *line_count)
1969 {
1970 	struct printk_record r;
1971 
1972 	prb_rec_init_rd(&r, info, NULL, 0);
1973 
1974 	return _prb_read_valid(rb, &seq, &r, line_count);
1975 }
1976 
1977 /**
1978  * prb_first_valid_seq() - Get the sequence number of the oldest available
1979  *                         record.
1980  *
1981  * @rb: The ringbuffer to get the sequence number from.
1982  *
1983  * This is the public function available to readers to see what the
1984  * first/oldest valid sequence number is.
1985  *
1986  * This provides readers a starting point to begin iterating the ringbuffer.
1987  *
1988  * Context: Any context.
1989  * Return: The sequence number of the first/oldest record or, if the
1990  *         ringbuffer is empty, 0 is returned.
1991  */
1992 u64 prb_first_valid_seq(struct printk_ringbuffer *rb)
1993 {
1994 	u64 seq = 0;
1995 
1996 	if (!_prb_read_valid(rb, &seq, NULL, NULL))
1997 		return 0;
1998 
1999 	return seq;
2000 }
2001 
2002 /**
2003  * prb_next_seq() - Get the sequence number after the last available record.
2004  *
2005  * @rb:  The ringbuffer to get the sequence number from.
2006  *
2007  * This is the public function available to readers to see what the next
2008  * newest sequence number available to readers will be.
2009  *
2010  * This provides readers a sequence number to jump to if all currently
2011  * available records should be skipped.
2012  *
2013  * Context: Any context.
2014  * Return: The sequence number of the next newest (not yet available) record
2015  *         for readers.
2016  */
2017 u64 prb_next_seq(struct printk_ringbuffer *rb)
2018 {
2019 	struct prb_desc_ring *desc_ring = &rb->desc_ring;
2020 	enum desc_state d_state;
2021 	unsigned long id;
2022 	u64 seq;
2023 
2024 	/* Check if the cached @id still points to a valid @seq. */
2025 	id = atomic_long_read(&desc_ring->last_finalized_id);
2026 	d_state = desc_read(desc_ring, id, NULL, &seq, NULL);
2027 
2028 	if (d_state == desc_finalized || d_state == desc_reusable) {
2029 		/*
2030 		 * Begin searching after the last finalized record.
2031 		 *
2032 		 * On 0, the search must begin at 0 because of hack#2
2033 		 * of the bootstrapping phase it is not known if a
2034 		 * record at index 0 exists.
2035 		 */
2036 		if (seq != 0)
2037 			seq++;
2038 	} else {
2039 		/*
2040 		 * The information about the last finalized sequence number
2041 		 * has gone. It should happen only when there is a flood of
2042 		 * new messages and the ringbuffer is rapidly recycled.
2043 		 * Give up and start from the beginning.
2044 		 */
2045 		seq = 0;
2046 	}
2047 
2048 	/*
2049 	 * The information about the last finalized @seq might be inaccurate.
2050 	 * Search forward to find the current one.
2051 	 */
2052 	while (_prb_read_valid(rb, &seq, NULL, NULL))
2053 		seq++;
2054 
2055 	return seq;
2056 }
2057 
2058 /**
2059  * prb_init() - Initialize a ringbuffer to use provided external buffers.
2060  *
2061  * @rb:       The ringbuffer to initialize.
2062  * @text_buf: The data buffer for text data.
2063  * @textbits: The size of @text_buf as a power-of-2 value.
2064  * @descs:    The descriptor buffer for ringbuffer records.
2065  * @descbits: The count of @descs items as a power-of-2 value.
2066  * @infos:    The printk_info buffer for ringbuffer records.
2067  *
2068  * This is the public function available to writers to setup a ringbuffer
2069  * during runtime using provided buffers.
2070  *
2071  * This must match the initialization of DEFINE_PRINTKRB().
2072  *
2073  * Context: Any context.
2074  */
2075 void prb_init(struct printk_ringbuffer *rb,
2076 	      char *text_buf, unsigned int textbits,
2077 	      struct prb_desc *descs, unsigned int descbits,
2078 	      struct printk_info *infos)
2079 {
2080 	memset(descs, 0, _DESCS_COUNT(descbits) * sizeof(descs[0]));
2081 	memset(infos, 0, _DESCS_COUNT(descbits) * sizeof(infos[0]));
2082 
2083 	rb->desc_ring.count_bits = descbits;
2084 	rb->desc_ring.descs = descs;
2085 	rb->desc_ring.infos = infos;
2086 	atomic_long_set(&rb->desc_ring.head_id, DESC0_ID(descbits));
2087 	atomic_long_set(&rb->desc_ring.tail_id, DESC0_ID(descbits));
2088 	atomic_long_set(&rb->desc_ring.last_finalized_id, DESC0_ID(descbits));
2089 
2090 	rb->text_data_ring.size_bits = textbits;
2091 	rb->text_data_ring.data = text_buf;
2092 	atomic_long_set(&rb->text_data_ring.head_lpos, BLK0_LPOS(textbits));
2093 	atomic_long_set(&rb->text_data_ring.tail_lpos, BLK0_LPOS(textbits));
2094 
2095 	atomic_long_set(&rb->fail, 0);
2096 
2097 	atomic_long_set(&(descs[_DESCS_COUNT(descbits) - 1].state_var), DESC0_SV(descbits));
2098 	descs[_DESCS_COUNT(descbits) - 1].text_blk_lpos.begin = FAILED_LPOS;
2099 	descs[_DESCS_COUNT(descbits) - 1].text_blk_lpos.next = FAILED_LPOS;
2100 
2101 	infos[0].seq = -(u64)_DESCS_COUNT(descbits);
2102 	infos[_DESCS_COUNT(descbits) - 1].seq = 0;
2103 }
2104 
2105 /**
2106  * prb_record_text_space() - Query the full actual used ringbuffer space for
2107  *                           the text data of a reserved entry.
2108  *
2109  * @e: The successfully reserved entry to query.
2110  *
2111  * This is the public function available to writers to see how much actual
2112  * space is used in the ringbuffer to store the text data of the specified
2113  * entry.
2114  *
2115  * This function is only valid if @e has been successfully reserved using
2116  * prb_reserve().
2117  *
2118  * Context: Any context.
2119  * Return: The size in bytes used by the text data of the associated record.
2120  */
2121 unsigned int prb_record_text_space(struct prb_reserved_entry *e)
2122 {
2123 	return e->text_space;
2124 }
2125