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) \
 334DESC_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 */
 346struct 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 */
 355static 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 */
 364static 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
 369static 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 */
 379static 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 */
 394static 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. */
 414static 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 */
 432static 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);
 532out:
 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 */
 543static 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 */
 564static 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 */
 632static 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 */
 767static 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. */
 875static 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. */
1002static 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 */
1024static 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 */
1105static 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. */
1175static 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 */
1205static 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 */
1259static 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 */
1350bool 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;
1435fail:
1436	prb_commit(e);
1437	/* prb_commit() re-enabled interrupts. */
1438fail_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 */
1448static 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 */
1485bool 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;
1569fail:
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. */
1576static 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 */
1632void 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 */
1664void 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 */
1679static 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 */
1705static 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 */
1755static 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 */
1796static 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. */
1835static 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 */
1885static 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 */
1937bool 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 */
1967bool 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 */
1992u64 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 */
2017u64 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 */
2075void 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 */
2121unsigned int prb_record_text_space(struct prb_reserved_entry *e)
2122{
2123	return e->text_space;
2124}