relay.c

   1/*
   2 * Public API and common code for kernel->userspace relay file support.
   3 *
   4 * See Documentation/filesystems/relay.rst for an overview.
   5 *
   6 * Copyright (C) 2002-2005 - Tom Zanussi (zanussi@us.ibm.com), IBM Corp
   7 * Copyright (C) 1999-2005 - Karim Yaghmour (karim@opersys.com)
   8 *
   9 * Moved to kernel/relay.c by Paul Mundt, 2006.
  10 * November 2006 - CPU hotplug support by Mathieu Desnoyers
  11 * 	(mathieu.desnoyers@polymtl.ca)
  12 *
  13 * This file is released under the GPL.
  14 */
  15#include <linux/errno.h>
  16#include <linux/stddef.h>
  17#include <linux/slab.h>
  18#include <linux/export.h>
  19#include <linux/string.h>
  20#include <linux/relay.h>
  21#include <linux/vmalloc.h>
  22#include <linux/mm.h>
  23#include <linux/cpu.h>
  24#include <linux/splice.h>
  25
  26/* list of open channels, for cpu hotplug */
  27static DEFINE_MUTEX(relay_channels_mutex);
  28static LIST_HEAD(relay_channels);
  29
  30/*
  31 * fault() vm_op implementation for relay file mapping.
  32 */
  33static vm_fault_t relay_buf_fault(struct vm_fault *vmf)
  34{
  35	struct page *page;
  36	struct rchan_buf *buf = vmf->vma->vm_private_data;
  37	pgoff_t pgoff = vmf->pgoff;
  38
  39	if (!buf)
  40		return VM_FAULT_OOM;
  41
  42	page = vmalloc_to_page(buf->start + (pgoff << PAGE_SHIFT));
  43	if (!page)
  44		return VM_FAULT_SIGBUS;
  45	get_page(page);
  46	vmf->page = page;
  47
  48	return 0;
  49}
  50
  51/*
  52 * vm_ops for relay file mappings.
  53 */
  54static const struct vm_operations_struct relay_file_mmap_ops = {
  55	.fault = relay_buf_fault,
  56};
  57
  58/*
  59 * allocate an array of pointers of struct page
  60 */
  61static struct page **relay_alloc_page_array(unsigned int n_pages)
  62{
  63	return kvcalloc(n_pages, sizeof(struct page *), GFP_KERNEL);
  64}
  65
  66/*
  67 * free an array of pointers of struct page
  68 */
  69static void relay_free_page_array(struct page **array)
  70{
  71	kvfree(array);
  72}
  73
  74/**
  75 *	relay_mmap_prepare_buf: - mmap channel buffer to process address space
  76 *	@buf: the relay channel buffer
  77 *	@desc: describing what to map
  78 *
  79 *	Returns 0 if ok, negative on error
  80 *
  81 *	Caller should already have grabbed mmap_lock.
  82 */
  83static int relay_mmap_prepare_buf(struct rchan_buf *buf,
  84				  struct vm_area_desc *desc)
  85{
  86	unsigned long length = vma_desc_size(desc);
  87
  88	if (!buf)
  89		return -EBADF;
  90
  91	if (length != (unsigned long)buf->chan->alloc_size)
  92		return -EINVAL;
  93
  94	desc->vm_ops = &relay_file_mmap_ops;
  95	desc->vm_flags |= VM_DONTEXPAND;
  96	desc->private_data = buf;
  97
  98	return 0;
  99}
 100
 101/**
 102 *	relay_alloc_buf - allocate a channel buffer
 103 *	@buf: the buffer struct
 104 *	@size: total size of the buffer
 105 *
 106 *	Returns a pointer to the resulting buffer, %NULL if unsuccessful. The
 107 *	passed in size will get page aligned, if it isn't already.
 108 */
 109static void *relay_alloc_buf(struct rchan_buf *buf, size_t *size)
 110{
 111	void *mem;
 112	unsigned int i, j, n_pages;
 113
 114	*size = PAGE_ALIGN(*size);
 115	n_pages = *size >> PAGE_SHIFT;
 116
 117	buf->page_array = relay_alloc_page_array(n_pages);
 118	if (!buf->page_array)
 119		return NULL;
 120
 121	for (i = 0; i < n_pages; i++) {
 122		buf->page_array[i] = alloc_page(GFP_KERNEL | __GFP_ZERO);
 123		if (unlikely(!buf->page_array[i]))
 124			goto depopulate;
 125		set_page_private(buf->page_array[i], (unsigned long)buf);
 126	}
 127	mem = vmap(buf->page_array, n_pages, VM_MAP, PAGE_KERNEL);
 128	if (!mem)
 129		goto depopulate;
 130
 131	buf->page_count = n_pages;
 132	return mem;
 133
 134depopulate:
 135	for (j = 0; j < i; j++)
 136		__free_page(buf->page_array[j]);
 137	relay_free_page_array(buf->page_array);
 138	return NULL;
 139}
 140
 141/**
 142 *	relay_create_buf - allocate and initialize a channel buffer
 143 *	@chan: the relay channel
 144 *
 145 *	Returns channel buffer if successful, %NULL otherwise.
 146 */
 147static struct rchan_buf *relay_create_buf(struct rchan *chan)
 148{
 149	struct rchan_buf *buf;
 150
 151	if (chan->n_subbufs > KMALLOC_MAX_SIZE / sizeof(size_t))
 152		return NULL;
 153
 154	buf = kzalloc(sizeof(struct rchan_buf), GFP_KERNEL);
 155	if (!buf)
 156		return NULL;
 157	buf->padding = kmalloc_array(chan->n_subbufs, sizeof(size_t),
 158				     GFP_KERNEL);
 159	if (!buf->padding)
 160		goto free_buf;
 161
 162	buf->start = relay_alloc_buf(buf, &chan->alloc_size);
 163	if (!buf->start)
 164		goto free_buf;
 165
 166	buf->chan = chan;
 167	kref_get(&buf->chan->kref);
 168	return buf;
 169
 170free_buf:
 171	kfree(buf->padding);
 172	kfree(buf);
 173	return NULL;
 174}
 175
 176/**
 177 *	relay_destroy_channel - free the channel struct
 178 *	@kref: target kernel reference that contains the relay channel
 179 *
 180 *	Should only be called from kref_put().
 181 */
 182static void relay_destroy_channel(struct kref *kref)
 183{
 184	struct rchan *chan = container_of(kref, struct rchan, kref);
 185	free_percpu(chan->buf);
 186	kfree(chan);
 187}
 188
 189/**
 190 *	relay_destroy_buf - destroy an rchan_buf struct and associated buffer
 191 *	@buf: the buffer struct
 192 */
 193static void relay_destroy_buf(struct rchan_buf *buf)
 194{
 195	struct rchan *chan = buf->chan;
 196	unsigned int i;
 197
 198	if (likely(buf->start)) {
 199		vunmap(buf->start);
 200		for (i = 0; i < buf->page_count; i++)
 201			__free_page(buf->page_array[i]);
 202		relay_free_page_array(buf->page_array);
 203	}
 204	*per_cpu_ptr(chan->buf, buf->cpu) = NULL;
 205	kfree(buf->padding);
 206	kfree(buf);
 207	kref_put(&chan->kref, relay_destroy_channel);
 208}
 209
 210/**
 211 *	relay_remove_buf - remove a channel buffer
 212 *	@kref: target kernel reference that contains the relay buffer
 213 *
 214 *	Removes the file from the filesystem, which also frees the
 215 *	rchan_buf_struct and the channel buffer.  Should only be called from
 216 *	kref_put().
 217 */
 218static void relay_remove_buf(struct kref *kref)
 219{
 220	struct rchan_buf *buf = container_of(kref, struct rchan_buf, kref);
 221	relay_destroy_buf(buf);
 222}
 223
 224/**
 225 *	relay_buf_empty - boolean, is the channel buffer empty?
 226 *	@buf: channel buffer
 227 *
 228 *	Returns 1 if the buffer is empty, 0 otherwise.
 229 */
 230static int relay_buf_empty(struct rchan_buf *buf)
 231{
 232	return (buf->subbufs_produced - buf->subbufs_consumed) ? 0 : 1;
 233}
 234
 235/**
 236 *	relay_buf_full - boolean, is the channel buffer full?
 237 *	@buf: channel buffer
 238 *
 239 *	Returns 1 if the buffer is full, 0 otherwise.
 240 */
 241int relay_buf_full(struct rchan_buf *buf)
 242{
 243	size_t ready = buf->subbufs_produced - buf->subbufs_consumed;
 244	return (ready >= buf->chan->n_subbufs) ? 1 : 0;
 245}
 246EXPORT_SYMBOL_GPL(relay_buf_full);
 247
 248/*
 249 * High-level relay kernel API and associated functions.
 250 */
 251
 252static int relay_subbuf_start(struct rchan_buf *buf, void *subbuf,
 253			      void *prev_subbuf)
 254{
 255	int full = relay_buf_full(buf);
 256
 257	if (full)
 258		buf->stats.full_count++;
 259
 260	if (!buf->chan->cb->subbuf_start)
 261		return !full;
 262
 263	return buf->chan->cb->subbuf_start(buf, subbuf,
 264					   prev_subbuf);
 265}
 266
 267/**
 268 *	wakeup_readers - wake up readers waiting on a channel
 269 *	@work: contains the channel buffer
 270 *
 271 *	This is the function used to defer reader waking
 272 */
 273static void wakeup_readers(struct irq_work *work)
 274{
 275	struct rchan_buf *buf;
 276
 277	buf = container_of(work, struct rchan_buf, wakeup_work);
 278	wake_up_interruptible(&buf->read_wait);
 279}
 280
 281/**
 282 *	__relay_reset - reset a channel buffer
 283 *	@buf: the channel buffer
 284 *	@init: 1 if this is a first-time initialization
 285 *
 286 *	See relay_reset() for description of effect.
 287 */
 288static void __relay_reset(struct rchan_buf *buf, unsigned int init)
 289{
 290	size_t i;
 291
 292	if (init) {
 293		init_waitqueue_head(&buf->read_wait);
 294		kref_init(&buf->kref);
 295		init_irq_work(&buf->wakeup_work, wakeup_readers);
 296	} else {
 297		irq_work_sync(&buf->wakeup_work);
 298	}
 299
 300	buf->subbufs_produced = 0;
 301	buf->subbufs_consumed = 0;
 302	buf->bytes_consumed = 0;
 303	buf->finalized = 0;
 304	buf->data = buf->start;
 305	buf->offset = 0;
 306	buf->stats.full_count = 0;
 307	buf->stats.big_count = 0;
 308
 309	for (i = 0; i < buf->chan->n_subbufs; i++)
 310		buf->padding[i] = 0;
 311
 312	relay_subbuf_start(buf, buf->data, NULL);
 313}
 314
 315/**
 316 *	relay_reset - reset the channel
 317 *	@chan: the channel
 318 *
 319 *	This has the effect of erasing all data from all channel buffers
 320 *	and restarting the channel in its initial state.  The buffers
 321 *	are not freed, so any mappings are still in effect.
 322 *
 323 *	NOTE. Care should be taken that the channel isn't actually
 324 *	being used by anything when this call is made.
 325 */
 326void relay_reset(struct rchan *chan)
 327{
 328	struct rchan_buf *buf;
 329	unsigned int i;
 330
 331	if (!chan)
 332		return;
 333
 334	if (chan->is_global && (buf = *per_cpu_ptr(chan->buf, 0))) {
 335		__relay_reset(buf, 0);
 336		return;
 337	}
 338
 339	mutex_lock(&relay_channels_mutex);
 340	for_each_possible_cpu(i)
 341		if ((buf = *per_cpu_ptr(chan->buf, i)))
 342			__relay_reset(buf, 0);
 343	mutex_unlock(&relay_channels_mutex);
 344}
 345EXPORT_SYMBOL_GPL(relay_reset);
 346
 347static inline void relay_set_buf_dentry(struct rchan_buf *buf,
 348					struct dentry *dentry)
 349{
 350	buf->dentry = dentry;
 351	d_inode(buf->dentry)->i_size = buf->early_bytes;
 352}
 353
 354static struct dentry *relay_create_buf_file(struct rchan *chan,
 355					    struct rchan_buf *buf,
 356					    unsigned int cpu)
 357{
 358	struct dentry *dentry;
 359	char *tmpname;
 360
 361	tmpname = kasprintf(GFP_KERNEL, "%s%d", chan->base_filename, cpu);
 362	if (!tmpname)
 363		return NULL;
 364
 365	/* Create file in fs */
 366	dentry = chan->cb->create_buf_file(tmpname, chan->parent,
 367					   S_IRUSR, buf,
 368					   &chan->is_global);
 369	if (IS_ERR(dentry))
 370		dentry = NULL;
 371
 372	kfree(tmpname);
 373
 374	return dentry;
 375}
 376
 377/*
 378 *	relay_open_buf - create a new relay channel buffer
 379 *
 380 *	used by relay_open() and CPU hotplug.
 381 */
 382static struct rchan_buf *relay_open_buf(struct rchan *chan, unsigned int cpu)
 383{
 384	struct rchan_buf *buf;
 385	struct dentry *dentry;
 386
 387 	if (chan->is_global)
 388		return *per_cpu_ptr(chan->buf, 0);
 389
 390	buf = relay_create_buf(chan);
 391	if (!buf)
 392		return NULL;
 393
 394	if (chan->has_base_filename) {
 395		dentry = relay_create_buf_file(chan, buf, cpu);
 396		if (!dentry)
 397			goto free_buf;
 398		relay_set_buf_dentry(buf, dentry);
 399	} else {
 400		/* Only retrieve global info, nothing more, nothing less */
 401		dentry = chan->cb->create_buf_file(NULL, NULL,
 402						   S_IRUSR, buf,
 403						   &chan->is_global);
 404		if (IS_ERR_OR_NULL(dentry))
 405			goto free_buf;
 406	}
 407
 408 	buf->cpu = cpu;
 409 	__relay_reset(buf, 1);
 410
 411 	if(chan->is_global) {
 412		*per_cpu_ptr(chan->buf, 0) = buf;
 413 		buf->cpu = 0;
 414  	}
 415
 416	return buf;
 417
 418free_buf:
 419 	relay_destroy_buf(buf);
 420	return NULL;
 421}
 422
 423/**
 424 *	relay_close_buf - close a channel buffer
 425 *	@buf: channel buffer
 426 *
 427 *	Marks the buffer finalized and restores the default callbacks.
 428 *	The channel buffer and channel buffer data structure are then freed
 429 *	automatically when the last reference is given up.
 430 */
 431static void relay_close_buf(struct rchan_buf *buf)
 432{
 433	buf->finalized = 1;
 434	irq_work_sync(&buf->wakeup_work);
 435	buf->chan->cb->remove_buf_file(buf->dentry);
 436	kref_put(&buf->kref, relay_remove_buf);
 437}
 438
 439int relay_prepare_cpu(unsigned int cpu)
 440{
 441	struct rchan *chan;
 442	struct rchan_buf *buf;
 443
 444	mutex_lock(&relay_channels_mutex);
 445	list_for_each_entry(chan, &relay_channels, list) {
 446		if (*per_cpu_ptr(chan->buf, cpu))
 447			continue;
 448		buf = relay_open_buf(chan, cpu);
 449		if (!buf) {
 450			pr_err("relay: cpu %d buffer creation failed\n", cpu);
 451			mutex_unlock(&relay_channels_mutex);
 452			return -ENOMEM;
 453		}
 454		*per_cpu_ptr(chan->buf, cpu) = buf;
 455	}
 456	mutex_unlock(&relay_channels_mutex);
 457	return 0;
 458}
 459
 460/**
 461 *	relay_open - create a new relay channel
 462 *	@base_filename: base name of files to create
 463 *	@parent: dentry of parent directory, %NULL for root directory or buffer
 464 *	@subbuf_size: size of sub-buffers
 465 *	@n_subbufs: number of sub-buffers
 466 *	@cb: client callback functions
 467 *	@private_data: user-defined data
 468 *
 469 *	Returns channel pointer if successful, %NULL otherwise.
 470 *
 471 *	Creates a channel buffer for each cpu using the sizes and
 472 *	attributes specified.  The created channel buffer files
 473 *	will be named base_filename0...base_filenameN-1.  File
 474 *	permissions will be %S_IRUSR.
 475 */
 476struct rchan *relay_open(const char *base_filename,
 477			 struct dentry *parent,
 478			 size_t subbuf_size,
 479			 size_t n_subbufs,
 480			 const struct rchan_callbacks *cb,
 481			 void *private_data)
 482{
 483	unsigned int i;
 484	struct rchan *chan;
 485	struct rchan_buf *buf;
 486
 487	if (!(subbuf_size && n_subbufs))
 488		return NULL;
 489	if (subbuf_size > UINT_MAX / n_subbufs)
 490		return NULL;
 491	if (!cb || !cb->create_buf_file || !cb->remove_buf_file)
 492		return NULL;
 493
 494	chan = kzalloc(sizeof(struct rchan), GFP_KERNEL);
 495	if (!chan)
 496		return NULL;
 497
 498	chan->buf = alloc_percpu(struct rchan_buf *);
 499	if (!chan->buf) {
 500		kfree(chan);
 501		return NULL;
 502	}
 503
 504	chan->version = RELAYFS_CHANNEL_VERSION;
 505	chan->n_subbufs = n_subbufs;
 506	chan->subbuf_size = subbuf_size;
 507	chan->alloc_size = PAGE_ALIGN(subbuf_size * n_subbufs);
 508	chan->parent = parent;
 509	chan->private_data = private_data;
 510	if (base_filename) {
 511		chan->has_base_filename = 1;
 512		strscpy(chan->base_filename, base_filename, NAME_MAX);
 513	}
 514	chan->cb = cb;
 515	kref_init(&chan->kref);
 516
 517	mutex_lock(&relay_channels_mutex);
 518	for_each_online_cpu(i) {
 519		buf = relay_open_buf(chan, i);
 520		if (!buf)
 521			goto free_bufs;
 522		*per_cpu_ptr(chan->buf, i) = buf;
 523	}
 524	list_add(&chan->list, &relay_channels);
 525	mutex_unlock(&relay_channels_mutex);
 526
 527	return chan;
 528
 529free_bufs:
 530	for_each_possible_cpu(i) {
 531		if ((buf = *per_cpu_ptr(chan->buf, i)))
 532			relay_close_buf(buf);
 533	}
 534
 535	kref_put(&chan->kref, relay_destroy_channel);
 536	mutex_unlock(&relay_channels_mutex);
 537	return NULL;
 538}
 539EXPORT_SYMBOL_GPL(relay_open);
 540
 541struct rchan_percpu_buf_dispatcher {
 542	struct rchan_buf *buf;
 543	struct dentry *dentry;
 544};
 545
 546/**
 547 *	relay_switch_subbuf - switch to a new sub-buffer
 548 *	@buf: channel buffer
 549 *	@length: size of current event
 550 *
 551 *	Returns either the length passed in or 0 if full.
 552 *
 553 *	Performs sub-buffer-switch tasks such as invoking callbacks,
 554 *	updating padding counts, waking up readers, etc.
 555 */
 556size_t relay_switch_subbuf(struct rchan_buf *buf, size_t length)
 557{
 558	void *old, *new;
 559	size_t old_subbuf, new_subbuf;
 560
 561	if (unlikely(length > buf->chan->subbuf_size))
 562		goto toobig;
 563
 564	if (buf->offset != buf->chan->subbuf_size + 1) {
 565		size_t prev_padding;
 566
 567		prev_padding = buf->chan->subbuf_size - buf->offset;
 568		old_subbuf = buf->subbufs_produced % buf->chan->n_subbufs;
 569		buf->padding[old_subbuf] = prev_padding;
 570		buf->subbufs_produced++;
 571		if (buf->dentry)
 572			d_inode(buf->dentry)->i_size +=
 573				buf->chan->subbuf_size -
 574				buf->padding[old_subbuf];
 575		else
 576			buf->early_bytes += buf->chan->subbuf_size -
 577					    buf->padding[old_subbuf];
 578		smp_mb();
 579		if (waitqueue_active(&buf->read_wait)) {
 580			/*
 581			 * Calling wake_up_interruptible() from here
 582			 * will deadlock if we happen to be logging
 583			 * from the scheduler (trying to re-grab
 584			 * rq->lock), so defer it.
 585			 */
 586			irq_work_queue(&buf->wakeup_work);
 587		}
 588	}
 589
 590	old = buf->data;
 591	new_subbuf = buf->subbufs_produced % buf->chan->n_subbufs;
 592	new = buf->start + new_subbuf * buf->chan->subbuf_size;
 593	buf->offset = 0;
 594	if (!relay_subbuf_start(buf, new, old)) {
 595		buf->offset = buf->chan->subbuf_size + 1;
 596		return 0;
 597	}
 598	buf->data = new;
 599	buf->padding[new_subbuf] = 0;
 600
 601	if (unlikely(length + buf->offset > buf->chan->subbuf_size))
 602		goto toobig;
 603
 604	return length;
 605
 606toobig:
 607	buf->stats.big_count++;
 608	return 0;
 609}
 610EXPORT_SYMBOL_GPL(relay_switch_subbuf);
 611
 612/**
 613 *	relay_subbufs_consumed - update the buffer's sub-buffers-consumed count
 614 *	@chan: the channel
 615 *	@cpu: the cpu associated with the channel buffer to update
 616 *	@subbufs_consumed: number of sub-buffers to add to current buf's count
 617 *
 618 *	Adds to the channel buffer's consumed sub-buffer count.
 619 *	subbufs_consumed should be the number of sub-buffers newly consumed,
 620 *	not the total consumed.
 621 *
 622 *	NOTE. Kernel clients don't need to call this function if the channel
 623 *	mode is 'overwrite'.
 624 */
 625void relay_subbufs_consumed(struct rchan *chan,
 626			    unsigned int cpu,
 627			    size_t subbufs_consumed)
 628{
 629	struct rchan_buf *buf;
 630
 631	if (!chan || cpu >= NR_CPUS)
 632		return;
 633
 634	buf = *per_cpu_ptr(chan->buf, cpu);
 635	if (!buf || subbufs_consumed > chan->n_subbufs)
 636		return;
 637
 638	if (subbufs_consumed > buf->subbufs_produced - buf->subbufs_consumed)
 639		buf->subbufs_consumed = buf->subbufs_produced;
 640	else
 641		buf->subbufs_consumed += subbufs_consumed;
 642}
 643EXPORT_SYMBOL_GPL(relay_subbufs_consumed);
 644
 645/**
 646 *	relay_close - close the channel
 647 *	@chan: the channel
 648 *
 649 *	Closes all channel buffers and frees the channel.
 650 */
 651void relay_close(struct rchan *chan)
 652{
 653	struct rchan_buf *buf;
 654	unsigned int i;
 655
 656	if (!chan)
 657		return;
 658
 659	mutex_lock(&relay_channels_mutex);
 660	if (chan->is_global && (buf = *per_cpu_ptr(chan->buf, 0)))
 661		relay_close_buf(buf);
 662	else
 663		for_each_possible_cpu(i)
 664			if ((buf = *per_cpu_ptr(chan->buf, i)))
 665				relay_close_buf(buf);
 666
 667	list_del(&chan->list);
 668	kref_put(&chan->kref, relay_destroy_channel);
 669	mutex_unlock(&relay_channels_mutex);
 670}
 671EXPORT_SYMBOL_GPL(relay_close);
 672
 673/**
 674 *	relay_flush - close the channel
 675 *	@chan: the channel
 676 *
 677 *	Flushes all channel buffers, i.e. forces buffer switch.
 678 */
 679void relay_flush(struct rchan *chan)
 680{
 681	struct rchan_buf *buf;
 682	unsigned int i;
 683
 684	if (!chan)
 685		return;
 686
 687	if (chan->is_global && (buf = *per_cpu_ptr(chan->buf, 0))) {
 688		relay_switch_subbuf(buf, 0);
 689		return;
 690	}
 691
 692	mutex_lock(&relay_channels_mutex);
 693	for_each_possible_cpu(i)
 694		if ((buf = *per_cpu_ptr(chan->buf, i)))
 695			relay_switch_subbuf(buf, 0);
 696	mutex_unlock(&relay_channels_mutex);
 697}
 698EXPORT_SYMBOL_GPL(relay_flush);
 699
 700/**
 701 *	relay_stats - get channel buffer statistics
 702 *	@chan: the channel
 703 *	@flags: select particular information to get
 704 *
 705 *	Returns the count of certain field that caller specifies.
 706 */
 707size_t relay_stats(struct rchan *chan, int flags)
 708{
 709	unsigned int i, count = 0;
 710	struct rchan_buf *rbuf;
 711
 712	if (!chan || flags > RELAY_STATS_LAST)
 713		return 0;
 714
 715	if (chan->is_global) {
 716		rbuf = *per_cpu_ptr(chan->buf, 0);
 717		if (flags & RELAY_STATS_BUF_FULL)
 718			count = rbuf->stats.full_count;
 719		else if (flags & RELAY_STATS_WRT_BIG)
 720			count = rbuf->stats.big_count;
 721	} else {
 722		for_each_online_cpu(i) {
 723			rbuf = *per_cpu_ptr(chan->buf, i);
 724			if (rbuf) {
 725				if (flags & RELAY_STATS_BUF_FULL)
 726					count += rbuf->stats.full_count;
 727				else if (flags & RELAY_STATS_WRT_BIG)
 728					count += rbuf->stats.big_count;
 729			}
 730		}
 731	}
 732
 733	return count;
 734}
 735
 736/**
 737 *	relay_file_open - open file op for relay files
 738 *	@inode: the inode
 739 *	@filp: the file
 740 *
 741 *	Increments the channel buffer refcount.
 742 */
 743static int relay_file_open(struct inode *inode, struct file *filp)
 744{
 745	struct rchan_buf *buf = inode->i_private;
 746	kref_get(&buf->kref);
 747	filp->private_data = buf;
 748
 749	return nonseekable_open(inode, filp);
 750}
 751
 752/**
 753 *	relay_file_mmap_prepare - mmap file op for relay files
 754 *	@desc: describing what to map
 755 *
 756 *	Calls upon relay_mmap_prepare_buf() to map the file into user space.
 757 */
 758static int relay_file_mmap_prepare(struct vm_area_desc *desc)
 759{
 760	struct rchan_buf *buf = desc->file->private_data;
 761
 762	return relay_mmap_prepare_buf(buf, desc);
 763}
 764
 765/**
 766 *	relay_file_poll - poll file op for relay files
 767 *	@filp: the file
 768 *	@wait: poll table
 769 *
 770 *	Poll implemention.
 771 */
 772static __poll_t relay_file_poll(struct file *filp, poll_table *wait)
 773{
 774	__poll_t mask = 0;
 775	struct rchan_buf *buf = filp->private_data;
 776
 777	if (buf->finalized)
 778		return EPOLLERR;
 779
 780	if (filp->f_mode & FMODE_READ) {
 781		poll_wait(filp, &buf->read_wait, wait);
 782		if (!relay_buf_empty(buf))
 783			mask |= EPOLLIN | EPOLLRDNORM;
 784	}
 785
 786	return mask;
 787}
 788
 789/**
 790 *	relay_file_release - release file op for relay files
 791 *	@inode: the inode
 792 *	@filp: the file
 793 *
 794 *	Decrements the channel refcount, as the filesystem is
 795 *	no longer using it.
 796 */
 797static int relay_file_release(struct inode *inode, struct file *filp)
 798{
 799	struct rchan_buf *buf = filp->private_data;
 800	kref_put(&buf->kref, relay_remove_buf);
 801
 802	return 0;
 803}
 804
 805/*
 806 *	relay_file_read_consume - update the consumed count for the buffer
 807 */
 808static void relay_file_read_consume(struct rchan_buf *buf,
 809				    size_t read_pos,
 810				    size_t bytes_consumed)
 811{
 812	size_t subbuf_size = buf->chan->subbuf_size;
 813	size_t n_subbufs = buf->chan->n_subbufs;
 814	size_t read_subbuf;
 815
 816	if (buf->subbufs_produced == buf->subbufs_consumed &&
 817	    buf->offset == buf->bytes_consumed)
 818		return;
 819
 820	if (buf->bytes_consumed + bytes_consumed > subbuf_size) {
 821		relay_subbufs_consumed(buf->chan, buf->cpu, 1);
 822		buf->bytes_consumed = 0;
 823	}
 824
 825	buf->bytes_consumed += bytes_consumed;
 826	if (!read_pos)
 827		read_subbuf = buf->subbufs_consumed % n_subbufs;
 828	else
 829		read_subbuf = read_pos / buf->chan->subbuf_size;
 830	if (buf->bytes_consumed + buf->padding[read_subbuf] == subbuf_size) {
 831		if ((read_subbuf == buf->subbufs_produced % n_subbufs) &&
 832		    (buf->offset == subbuf_size))
 833			return;
 834		relay_subbufs_consumed(buf->chan, buf->cpu, 1);
 835		buf->bytes_consumed = 0;
 836	}
 837}
 838
 839/*
 840 *	relay_file_read_avail - boolean, are there unconsumed bytes available?
 841 */
 842static int relay_file_read_avail(struct rchan_buf *buf)
 843{
 844	size_t subbuf_size = buf->chan->subbuf_size;
 845	size_t n_subbufs = buf->chan->n_subbufs;
 846	size_t produced = buf->subbufs_produced;
 847	size_t consumed;
 848
 849	relay_file_read_consume(buf, 0, 0);
 850
 851	consumed = buf->subbufs_consumed;
 852
 853	if (unlikely(buf->offset > subbuf_size)) {
 854		if (produced == consumed)
 855			return 0;
 856		return 1;
 857	}
 858
 859	if (unlikely(produced - consumed >= n_subbufs)) {
 860		consumed = produced - n_subbufs + 1;
 861		buf->subbufs_consumed = consumed;
 862		buf->bytes_consumed = 0;
 863	}
 864
 865	produced = (produced % n_subbufs) * subbuf_size + buf->offset;
 866	consumed = (consumed % n_subbufs) * subbuf_size + buf->bytes_consumed;
 867
 868	if (consumed > produced)
 869		produced += n_subbufs * subbuf_size;
 870
 871	if (consumed == produced) {
 872		if (buf->offset == subbuf_size &&
 873		    buf->subbufs_produced > buf->subbufs_consumed)
 874			return 1;
 875		return 0;
 876	}
 877
 878	return 1;
 879}
 880
 881/**
 882 *	relay_file_read_subbuf_avail - return bytes available in sub-buffer
 883 *	@read_pos: file read position
 884 *	@buf: relay channel buffer
 885 */
 886static size_t relay_file_read_subbuf_avail(size_t read_pos,
 887					   struct rchan_buf *buf)
 888{
 889	size_t padding, avail = 0;
 890	size_t read_subbuf, read_offset, write_subbuf, write_offset;
 891	size_t subbuf_size = buf->chan->subbuf_size;
 892
 893	write_subbuf = (buf->data - buf->start) / subbuf_size;
 894	write_offset = buf->offset > subbuf_size ? subbuf_size : buf->offset;
 895	read_subbuf = read_pos / subbuf_size;
 896	read_offset = read_pos % subbuf_size;
 897	padding = buf->padding[read_subbuf];
 898
 899	if (read_subbuf == write_subbuf) {
 900		if (read_offset + padding < write_offset)
 901			avail = write_offset - (read_offset + padding);
 902	} else
 903		avail = (subbuf_size - padding) - read_offset;
 904
 905	return avail;
 906}
 907
 908/**
 909 *	relay_file_read_start_pos - find the first available byte to read
 910 *	@buf: relay channel buffer
 911 *
 912 *	If the read_pos is in the middle of padding, return the
 913 *	position of the first actually available byte, otherwise
 914 *	return the original value.
 915 */
 916static size_t relay_file_read_start_pos(struct rchan_buf *buf)
 917{
 918	size_t read_subbuf, padding, padding_start, padding_end;
 919	size_t subbuf_size = buf->chan->subbuf_size;
 920	size_t n_subbufs = buf->chan->n_subbufs;
 921	size_t consumed = buf->subbufs_consumed % n_subbufs;
 922	size_t read_pos = (consumed * subbuf_size + buf->bytes_consumed)
 923			% (n_subbufs * subbuf_size);
 924
 925	read_subbuf = read_pos / subbuf_size;
 926	padding = buf->padding[read_subbuf];
 927	padding_start = (read_subbuf + 1) * subbuf_size - padding;
 928	padding_end = (read_subbuf + 1) * subbuf_size;
 929	if (read_pos >= padding_start && read_pos < padding_end) {
 930		read_subbuf = (read_subbuf + 1) % n_subbufs;
 931		read_pos = read_subbuf * subbuf_size;
 932	}
 933
 934	return read_pos;
 935}
 936
 937/**
 938 *	relay_file_read_end_pos - return the new read position
 939 *	@read_pos: file read position
 940 *	@buf: relay channel buffer
 941 *	@count: number of bytes to be read
 942 */
 943static size_t relay_file_read_end_pos(struct rchan_buf *buf,
 944				      size_t read_pos,
 945				      size_t count)
 946{
 947	size_t read_subbuf, padding, end_pos;
 948	size_t subbuf_size = buf->chan->subbuf_size;
 949	size_t n_subbufs = buf->chan->n_subbufs;
 950
 951	read_subbuf = read_pos / subbuf_size;
 952	padding = buf->padding[read_subbuf];
 953	if (read_pos % subbuf_size + count + padding == subbuf_size)
 954		end_pos = (read_subbuf + 1) * subbuf_size;
 955	else
 956		end_pos = read_pos + count;
 957	if (end_pos >= subbuf_size * n_subbufs)
 958		end_pos = 0;
 959
 960	return end_pos;
 961}
 962
 963static ssize_t relay_file_read(struct file *filp,
 964			       char __user *buffer,
 965			       size_t count,
 966			       loff_t *ppos)
 967{
 968	struct rchan_buf *buf = filp->private_data;
 969	size_t read_start, avail;
 970	size_t written = 0;
 971	int ret;
 972
 973	if (!count)
 974		return 0;
 975
 976	inode_lock(file_inode(filp));
 977	do {
 978		void *from;
 979
 980		if (!relay_file_read_avail(buf))
 981			break;
 982
 983		read_start = relay_file_read_start_pos(buf);
 984		avail = relay_file_read_subbuf_avail(read_start, buf);
 985		if (!avail)
 986			break;
 987
 988		avail = min(count, avail);
 989		from = buf->start + read_start;
 990		ret = avail;
 991		if (copy_to_user(buffer, from, avail))
 992			break;
 993
 994		buffer += ret;
 995		written += ret;
 996		count -= ret;
 997
 998		relay_file_read_consume(buf, read_start, ret);
 999		*ppos = relay_file_read_end_pos(buf, read_start, ret);
1000	} while (count);
1001	inode_unlock(file_inode(filp));
1002
1003	return written;
1004}
1005
1006
1007const struct file_operations relay_file_operations = {
1008	.open		= relay_file_open,
1009	.poll		= relay_file_poll,
1010	.mmap_prepare	= relay_file_mmap_prepare,
1011	.read		= relay_file_read,
1012	.release	= relay_file_release,
1013};
1014EXPORT_SYMBOL_GPL(relay_file_operations);