1/*
   2 * Public API and common code for kernel->userspace relay file support.
   3 *
   4 * See Documentation/filesystems/relay.txt 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 * close() vm_op implementation for relay file mapping.
  32 */
  33static void relay_file_mmap_close(struct vm_area_struct *vma)
  34{
  35	struct rchan_buf *buf = vma->vm_private_data;
  36	buf->chan->cb->buf_unmapped(buf, vma->vm_file);
  37}
  38
  39/*
  40 * fault() vm_op implementation for relay file mapping.
  41 */
  42static int relay_buf_fault(struct vm_area_struct *vma, struct vm_fault *vmf)
  43{
  44	struct page *page;
  45	struct rchan_buf *buf = vma->vm_private_data;
  46	pgoff_t pgoff = vmf->pgoff;
  47
  48	if (!buf)
  49		return VM_FAULT_OOM;
  50
  51	page = vmalloc_to_page(buf->start + (pgoff << PAGE_SHIFT));
  52	if (!page)
  53		return VM_FAULT_SIGBUS;
  54	get_page(page);
  55	vmf->page = page;
  56
  57	return 0;
  58}
  59
  60/*
  61 * vm_ops for relay file mappings.
  62 */
  63static const struct vm_operations_struct relay_file_mmap_ops = {
  64	.fault = relay_buf_fault,
  65	.close = relay_file_mmap_close,
  66};
  67
  68/*
  69 * allocate an array of pointers of struct page
  70 */
  71static struct page **relay_alloc_page_array(unsigned int n_pages)
  72{
  73	const size_t pa_size = n_pages * sizeof(struct page *);
  74	if (pa_size > PAGE_SIZE)
  75		return vzalloc(pa_size);
  76	return kzalloc(pa_size, GFP_KERNEL);
  77}
  78
  79/*
  80 * free an array of pointers of struct page
  81 */
  82static void relay_free_page_array(struct page **array)
  83{
  84	kvfree(array);
  85}
  86
  87/**
  88 *	relay_mmap_buf: - mmap channel buffer to process address space
  89 *	@buf: relay channel buffer
  90 *	@vma: vm_area_struct describing memory to be mapped
  91 *
  92 *	Returns 0 if ok, negative on error
  93 *
  94 *	Caller should already have grabbed mmap_sem.
  95 */
  96static int relay_mmap_buf(struct rchan_buf *buf, struct vm_area_struct *vma)
  97{
  98	unsigned long length = vma->vm_end - vma->vm_start;
  99	struct file *filp = vma->vm_file;
 100
 101	if (!buf)
 102		return -EBADF;
 103
 104	if (length != (unsigned long)buf->chan->alloc_size)
 105		return -EINVAL;
 106
 107	vma->vm_ops = &relay_file_mmap_ops;
 108	vma->vm_flags |= VM_DONTEXPAND;
 109	vma->vm_private_data = buf;
 110	buf->chan->cb->buf_mapped(buf, filp);
 111
 112	return 0;
 113}
 114
 115/**
 116 *	relay_alloc_buf - allocate a channel buffer
 117 *	@buf: the buffer struct
 118 *	@size: total size of the buffer
 119 *
 120 *	Returns a pointer to the resulting buffer, %NULL if unsuccessful. The
 121 *	passed in size will get page aligned, if it isn't already.
 122 */
 123static void *relay_alloc_buf(struct rchan_buf *buf, size_t *size)
 124{
 125	void *mem;
 126	unsigned int i, j, n_pages;
 127
 128	*size = PAGE_ALIGN(*size);
 129	n_pages = *size >> PAGE_SHIFT;
 130
 131	buf->page_array = relay_alloc_page_array(n_pages);
 132	if (!buf->page_array)
 133		return NULL;
 134
 135	for (i = 0; i < n_pages; i++) {
 136		buf->page_array[i] = alloc_page(GFP_KERNEL);
 137		if (unlikely(!buf->page_array[i]))
 138			goto depopulate;
 139		set_page_private(buf->page_array[i], (unsigned long)buf);
 140	}
 141	mem = vmap(buf->page_array, n_pages, VM_MAP, PAGE_KERNEL);
 142	if (!mem)
 143		goto depopulate;
 144
 145	memset(mem, 0, *size);
 146	buf->page_count = n_pages;
 147	return mem;
 148
 149depopulate:
 150	for (j = 0; j < i; j++)
 151		__free_page(buf->page_array[j]);
 152	relay_free_page_array(buf->page_array);
 153	return NULL;
 154}
 155
 156/**
 157 *	relay_create_buf - allocate and initialize a channel buffer
 158 *	@chan: the relay channel
 159 *
 160 *	Returns channel buffer if successful, %NULL otherwise.
 161 */
 162static struct rchan_buf *relay_create_buf(struct rchan *chan)
 163{
 164	struct rchan_buf *buf;
 165
 166	if (chan->n_subbufs > UINT_MAX / sizeof(size_t *))
 167		return NULL;
 168
 169	buf = kzalloc(sizeof(struct rchan_buf), GFP_KERNEL);
 170	if (!buf)
 171		return NULL;
 172	buf->padding = kmalloc(chan->n_subbufs * sizeof(size_t *), GFP_KERNEL);
 173	if (!buf->padding)
 174		goto free_buf;
 175
 176	buf->start = relay_alloc_buf(buf, &chan->alloc_size);
 177	if (!buf->start)
 178		goto free_buf;
 179
 180	buf->chan = chan;
 181	kref_get(&buf->chan->kref);
 182	return buf;
 183
 184free_buf:
 185	kfree(buf->padding);
 186	kfree(buf);
 187	return NULL;
 188}
 189
 190/**
 191 *	relay_destroy_channel - free the channel struct
 192 *	@kref: target kernel reference that contains the relay channel
 193 *
 194 *	Should only be called from kref_put().
 195 */
 196static void relay_destroy_channel(struct kref *kref)
 197{
 198	struct rchan *chan = container_of(kref, struct rchan, kref);
 199	kfree(chan);
 200}
 201
 202/**
 203 *	relay_destroy_buf - destroy an rchan_buf struct and associated buffer
 204 *	@buf: the buffer struct
 205 */
 206static void relay_destroy_buf(struct rchan_buf *buf)
 207{
 208	struct rchan *chan = buf->chan;
 209	unsigned int i;
 210
 211	if (likely(buf->start)) {
 212		vunmap(buf->start);
 213		for (i = 0; i < buf->page_count; i++)
 214			__free_page(buf->page_array[i]);
 215		relay_free_page_array(buf->page_array);
 216	}
 217	*per_cpu_ptr(chan->buf, buf->cpu) = NULL;
 218	kfree(buf->padding);
 219	kfree(buf);
 220	kref_put(&chan->kref, relay_destroy_channel);
 221}
 222
 223/**
 224 *	relay_remove_buf - remove a channel buffer
 225 *	@kref: target kernel reference that contains the relay buffer
 226 *
 227 *	Removes the file from the filesystem, which also frees the
 228 *	rchan_buf_struct and the channel buffer.  Should only be called from
 229 *	kref_put().
 230 */
 231static void relay_remove_buf(struct kref *kref)
 232{
 233	struct rchan_buf *buf = container_of(kref, struct rchan_buf, kref);
 234	relay_destroy_buf(buf);
 235}
 236
 237/**
 238 *	relay_buf_empty - boolean, is the channel buffer empty?
 239 *	@buf: channel buffer
 240 *
 241 *	Returns 1 if the buffer is empty, 0 otherwise.
 242 */
 243static int relay_buf_empty(struct rchan_buf *buf)
 244{
 245	return (buf->subbufs_produced - buf->subbufs_consumed) ? 0 : 1;
 246}
 247
 248/**
 249 *	relay_buf_full - boolean, is the channel buffer full?
 250 *	@buf: channel buffer
 251 *
 252 *	Returns 1 if the buffer is full, 0 otherwise.
 253 */
 254int relay_buf_full(struct rchan_buf *buf)
 255{
 256	size_t ready = buf->subbufs_produced - buf->subbufs_consumed;
 257	return (ready >= buf->chan->n_subbufs) ? 1 : 0;
 258}
 259EXPORT_SYMBOL_GPL(relay_buf_full);
 260
 261/*
 262 * High-level relay kernel API and associated functions.
 263 */
 264
 265/*
 266 * rchan_callback implementations defining default channel behavior.  Used
 267 * in place of corresponding NULL values in client callback struct.
 268 */
 269
 270/*
 271 * subbuf_start() default callback.  Does nothing.
 272 */
 273static int subbuf_start_default_callback (struct rchan_buf *buf,
 274					  void *subbuf,
 275					  void *prev_subbuf,
 276					  size_t prev_padding)
 277{
 278	if (relay_buf_full(buf))
 279		return 0;
 280
 281	return 1;
 282}
 283
 284/*
 285 * buf_mapped() default callback.  Does nothing.
 286 */
 287static void buf_mapped_default_callback(struct rchan_buf *buf,
 288					struct file *filp)
 289{
 290}
 291
 292/*
 293 * buf_unmapped() default callback.  Does nothing.
 294 */
 295static void buf_unmapped_default_callback(struct rchan_buf *buf,
 296					  struct file *filp)
 297{
 298}
 299
 300/*
 301 * create_buf_file_create() default callback.  Does nothing.
 302 */
 303static struct dentry *create_buf_file_default_callback(const char *filename,
 304						       struct dentry *parent,
 305						       umode_t mode,
 306						       struct rchan_buf *buf,
 307						       int *is_global)
 308{
 309	return NULL;
 310}
 311
 312/*
 313 * remove_buf_file() default callback.  Does nothing.
 314 */
 315static int remove_buf_file_default_callback(struct dentry *dentry)
 316{
 317	return -EINVAL;
 318}
 319
 320/* relay channel default callbacks */
 321static struct rchan_callbacks default_channel_callbacks = {
 322	.subbuf_start = subbuf_start_default_callback,
 323	.buf_mapped = buf_mapped_default_callback,
 324	.buf_unmapped = buf_unmapped_default_callback,
 325	.create_buf_file = create_buf_file_default_callback,
 326	.remove_buf_file = remove_buf_file_default_callback,
 327};
 328
 329/**
 330 *	wakeup_readers - wake up readers waiting on a channel
 331 *	@work: contains the channel buffer
 332 *
 333 *	This is the function used to defer reader waking
 334 */
 335static void wakeup_readers(struct irq_work *work)
 336{
 337	struct rchan_buf *buf;
 338
 339	buf = container_of(work, struct rchan_buf, wakeup_work);
 340	wake_up_interruptible(&buf->read_wait);
 341}
 342
 343/**
 344 *	__relay_reset - reset a channel buffer
 345 *	@buf: the channel buffer
 346 *	@init: 1 if this is a first-time initialization
 347 *
 348 *	See relay_reset() for description of effect.
 349 */
 350static void __relay_reset(struct rchan_buf *buf, unsigned int init)
 351{
 352	size_t i;
 353
 354	if (init) {
 355		init_waitqueue_head(&buf->read_wait);
 356		kref_init(&buf->kref);
 357		init_irq_work(&buf->wakeup_work, wakeup_readers);
 358	} else {
 359		irq_work_sync(&buf->wakeup_work);
 360	}
 361
 362	buf->subbufs_produced = 0;
 363	buf->subbufs_consumed = 0;
 364	buf->bytes_consumed = 0;
 365	buf->finalized = 0;
 366	buf->data = buf->start;
 367	buf->offset = 0;
 368
 369	for (i = 0; i < buf->chan->n_subbufs; i++)
 370		buf->padding[i] = 0;
 371
 372	buf->chan->cb->subbuf_start(buf, buf->data, NULL, 0);
 373}
 374
 375/**
 376 *	relay_reset - reset the channel
 377 *	@chan: the channel
 378 *
 379 *	This has the effect of erasing all data from all channel buffers
 380 *	and restarting the channel in its initial state.  The buffers
 381 *	are not freed, so any mappings are still in effect.
 382 *
 383 *	NOTE. Care should be taken that the channel isn't actually
 384 *	being used by anything when this call is made.
 385 */
 386void relay_reset(struct rchan *chan)
 387{
 388	struct rchan_buf *buf;
 389	unsigned int i;
 390
 391	if (!chan)
 392		return;
 393
 394	if (chan->is_global && (buf = *per_cpu_ptr(chan->buf, 0))) {
 395		__relay_reset(buf, 0);
 396		return;
 397	}
 398
 399	mutex_lock(&relay_channels_mutex);
 400	for_each_possible_cpu(i)
 401		if ((buf = *per_cpu_ptr(chan->buf, i)))
 402			__relay_reset(buf, 0);
 403	mutex_unlock(&relay_channels_mutex);
 404}
 405EXPORT_SYMBOL_GPL(relay_reset);
 406
 407static inline void relay_set_buf_dentry(struct rchan_buf *buf,
 408					struct dentry *dentry)
 409{
 410	buf->dentry = dentry;
 411	d_inode(buf->dentry)->i_size = buf->early_bytes;
 412}
 413
 414static struct dentry *relay_create_buf_file(struct rchan *chan,
 415					    struct rchan_buf *buf,
 416					    unsigned int cpu)
 417{
 418	struct dentry *dentry;
 419	char *tmpname;
 420
 421	tmpname = kzalloc(NAME_MAX + 1, GFP_KERNEL);
 422	if (!tmpname)
 423		return NULL;
 424	snprintf(tmpname, NAME_MAX, "%s%d", chan->base_filename, cpu);
 425
 426	/* Create file in fs */
 427	dentry = chan->cb->create_buf_file(tmpname, chan->parent,
 428					   S_IRUSR, buf,
 429					   &chan->is_global);
 430
 431	kfree(tmpname);
 432
 433	return dentry;
 434}
 435
 436/*
 437 *	relay_open_buf - create a new relay channel buffer
 438 *
 439 *	used by relay_open() and CPU hotplug.
 440 */
 441static struct rchan_buf *relay_open_buf(struct rchan *chan, unsigned int cpu)
 442{
 443 	struct rchan_buf *buf = NULL;
 444	struct dentry *dentry;
 445
 446 	if (chan->is_global)
 447		return *per_cpu_ptr(chan->buf, 0);
 448
 449	buf = relay_create_buf(chan);
 450	if (!buf)
 451		return NULL;
 452
 453	if (chan->has_base_filename) {
 454		dentry = relay_create_buf_file(chan, buf, cpu);
 455		if (!dentry)
 456			goto free_buf;
 457		relay_set_buf_dentry(buf, dentry);
 458	} else {
 459		/* Only retrieve global info, nothing more, nothing less */
 460		dentry = chan->cb->create_buf_file(NULL, NULL,
 461						   S_IRUSR, buf,
 462						   &chan->is_global);
 463		if (WARN_ON(dentry))
 464			goto free_buf;
 465	}
 466
 467 	buf->cpu = cpu;
 468 	__relay_reset(buf, 1);
 469
 470 	if(chan->is_global) {
 471		*per_cpu_ptr(chan->buf, 0) = buf;
 472 		buf->cpu = 0;
 473  	}
 474
 475	return buf;
 476
 477free_buf:
 478 	relay_destroy_buf(buf);
 479	return NULL;
 480}
 481
 482/**
 483 *	relay_close_buf - close a channel buffer
 484 *	@buf: channel buffer
 485 *
 486 *	Marks the buffer finalized and restores the default callbacks.
 487 *	The channel buffer and channel buffer data structure are then freed
 488 *	automatically when the last reference is given up.
 489 */
 490static void relay_close_buf(struct rchan_buf *buf)
 491{
 492	buf->finalized = 1;
 493	irq_work_sync(&buf->wakeup_work);
 494	buf->chan->cb->remove_buf_file(buf->dentry);
 495	kref_put(&buf->kref, relay_remove_buf);
 496}
 497
 498static void setup_callbacks(struct rchan *chan,
 499				   struct rchan_callbacks *cb)
 500{
 501	if (!cb) {
 502		chan->cb = &default_channel_callbacks;
 503		return;
 504	}
 505
 506	if (!cb->subbuf_start)
 507		cb->subbuf_start = subbuf_start_default_callback;
 508	if (!cb->buf_mapped)
 509		cb->buf_mapped = buf_mapped_default_callback;
 510	if (!cb->buf_unmapped)
 511		cb->buf_unmapped = buf_unmapped_default_callback;
 512	if (!cb->create_buf_file)
 513		cb->create_buf_file = create_buf_file_default_callback;
 514	if (!cb->remove_buf_file)
 515		cb->remove_buf_file = remove_buf_file_default_callback;
 516	chan->cb = cb;
 517}
 518
 519int relay_prepare_cpu(unsigned int cpu)
 520{
 521	struct rchan *chan;
 522	struct rchan_buf *buf;
 523
 524	mutex_lock(&relay_channels_mutex);
 525	list_for_each_entry(chan, &relay_channels, list) {
 526		if ((buf = *per_cpu_ptr(chan->buf, cpu)))
 527			continue;
 528		buf = relay_open_buf(chan, cpu);
 529		if (!buf) {
 530			pr_err("relay: cpu %d buffer creation failed\n", cpu);
 531			mutex_unlock(&relay_channels_mutex);
 532			return -ENOMEM;
 533		}
 534		*per_cpu_ptr(chan->buf, cpu) = buf;
 535	}
 536	mutex_unlock(&relay_channels_mutex);
 537	return 0;
 538}
 539
 540/**
 541 *	relay_open - create a new relay channel
 542 *	@base_filename: base name of files to create, %NULL for buffering only
 543 *	@parent: dentry of parent directory, %NULL for root directory or buffer
 544 *	@subbuf_size: size of sub-buffers
 545 *	@n_subbufs: number of sub-buffers
 546 *	@cb: client callback functions
 547 *	@private_data: user-defined data
 548 *
 549 *	Returns channel pointer if successful, %NULL otherwise.
 550 *
 551 *	Creates a channel buffer for each cpu using the sizes and
 552 *	attributes specified.  The created channel buffer files
 553 *	will be named base_filename0...base_filenameN-1.  File
 554 *	permissions will be %S_IRUSR.
 555 *
 556 *	If opening a buffer (@parent = NULL) that you later wish to register
 557 *	in a filesystem, call relay_late_setup_files() once the @parent dentry
 558 *	is available.
 559 */
 560struct rchan *relay_open(const char *base_filename,
 561			 struct dentry *parent,
 562			 size_t subbuf_size,
 563			 size_t n_subbufs,
 564			 struct rchan_callbacks *cb,
 565			 void *private_data)
 566{
 567	unsigned int i;
 568	struct rchan *chan;
 569	struct rchan_buf *buf;
 570
 571	if (!(subbuf_size && n_subbufs))
 572		return NULL;
 573	if (subbuf_size > UINT_MAX / n_subbufs)
 574		return NULL;
 575
 576	chan = kzalloc(sizeof(struct rchan), GFP_KERNEL);
 577	if (!chan)
 578		return NULL;
 579
 580	chan->buf = alloc_percpu(struct rchan_buf *);
 581	chan->version = RELAYFS_CHANNEL_VERSION;
 582	chan->n_subbufs = n_subbufs;
 583	chan->subbuf_size = subbuf_size;
 584	chan->alloc_size = PAGE_ALIGN(subbuf_size * n_subbufs);
 585	chan->parent = parent;
 586	chan->private_data = private_data;
 587	if (base_filename) {
 588		chan->has_base_filename = 1;
 589		strlcpy(chan->base_filename, base_filename, NAME_MAX);
 590	}
 591	setup_callbacks(chan, cb);
 592	kref_init(&chan->kref);
 593
 594	mutex_lock(&relay_channels_mutex);
 595	for_each_online_cpu(i) {
 596		buf = relay_open_buf(chan, i);
 597		if (!buf)
 598			goto free_bufs;
 599		*per_cpu_ptr(chan->buf, i) = buf;
 600	}
 601	list_add(&chan->list, &relay_channels);
 602	mutex_unlock(&relay_channels_mutex);
 603
 604	return chan;
 605
 606free_bufs:
 607	for_each_possible_cpu(i) {
 608		if ((buf = *per_cpu_ptr(chan->buf, i)))
 609			relay_close_buf(buf);
 610	}
 611
 612	kref_put(&chan->kref, relay_destroy_channel);
 613	mutex_unlock(&relay_channels_mutex);
 614	kfree(chan);
 615	return NULL;
 616}
 617EXPORT_SYMBOL_GPL(relay_open);
 618
 619struct rchan_percpu_buf_dispatcher {
 620	struct rchan_buf *buf;
 621	struct dentry *dentry;
 622};
 623
 624/* Called in atomic context. */
 625static void __relay_set_buf_dentry(void *info)
 626{
 627	struct rchan_percpu_buf_dispatcher *p = info;
 628
 629	relay_set_buf_dentry(p->buf, p->dentry);
 630}
 631
 632/**
 633 *	relay_late_setup_files - triggers file creation
 634 *	@chan: channel to operate on
 635 *	@base_filename: base name of files to create
 636 *	@parent: dentry of parent directory, %NULL for root directory
 637 *
 638 *	Returns 0 if successful, non-zero otherwise.
 639 *
 640 *	Use to setup files for a previously buffer-only channel created
 641 *	by relay_open() with a NULL parent dentry.
 642 *
 643 *	For example, this is useful for perfomring early tracing in kernel,
 644 *	before VFS is up and then exposing the early results once the dentry
 645 *	is available.
 646 */
 647int relay_late_setup_files(struct rchan *chan,
 648			   const char *base_filename,
 649			   struct dentry *parent)
 650{
 651	int err = 0;
 652	unsigned int i, curr_cpu;
 653	unsigned long flags;
 654	struct dentry *dentry;
 655	struct rchan_buf *buf;
 656	struct rchan_percpu_buf_dispatcher disp;
 657
 658	if (!chan || !base_filename)
 659		return -EINVAL;
 660
 661	strlcpy(chan->base_filename, base_filename, NAME_MAX);
 662
 663	mutex_lock(&relay_channels_mutex);
 664	/* Is chan already set up? */
 665	if (unlikely(chan->has_base_filename)) {
 666		mutex_unlock(&relay_channels_mutex);
 667		return -EEXIST;
 668	}
 669	chan->has_base_filename = 1;
 670	chan->parent = parent;
 671
 672	if (chan->is_global) {
 673		err = -EINVAL;
 674		buf = *per_cpu_ptr(chan->buf, 0);
 675		if (!WARN_ON_ONCE(!buf)) {
 676			dentry = relay_create_buf_file(chan, buf, 0);
 677			if (dentry && !WARN_ON_ONCE(!chan->is_global)) {
 678				relay_set_buf_dentry(buf, dentry);
 679				err = 0;
 680			}
 681		}
 682		mutex_unlock(&relay_channels_mutex);
 683		return err;
 684	}
 685
 686	curr_cpu = get_cpu();
 687	/*
 688	 * The CPU hotplug notifier ran before us and created buffers with
 689	 * no files associated. So it's safe to call relay_setup_buf_file()
 690	 * on all currently online CPUs.
 691	 */
 692	for_each_online_cpu(i) {
 693		buf = *per_cpu_ptr(chan->buf, i);
 694		if (unlikely(!buf)) {
 695			WARN_ONCE(1, KERN_ERR "CPU has no buffer!\n");
 696			err = -EINVAL;
 697			break;
 698		}
 699
 700		dentry = relay_create_buf_file(chan, buf, i);
 701		if (unlikely(!dentry)) {
 702			err = -EINVAL;
 703			break;
 704		}
 705
 706		if (curr_cpu == i) {
 707			local_irq_save(flags);
 708			relay_set_buf_dentry(buf, dentry);
 709			local_irq_restore(flags);
 710		} else {
 711			disp.buf = buf;
 712			disp.dentry = dentry;
 713			smp_mb();
 714			/* relay_channels_mutex must be held, so wait. */
 715			err = smp_call_function_single(i,
 716						       __relay_set_buf_dentry,
 717						       &disp, 1);
 718		}
 719		if (unlikely(err))
 720			break;
 721	}
 722	put_cpu();
 723	mutex_unlock(&relay_channels_mutex);
 724
 725	return err;
 726}
 727EXPORT_SYMBOL_GPL(relay_late_setup_files);
 728
 729/**
 730 *	relay_switch_subbuf - switch to a new sub-buffer
 731 *	@buf: channel buffer
 732 *	@length: size of current event
 733 *
 734 *	Returns either the length passed in or 0 if full.
 735 *
 736 *	Performs sub-buffer-switch tasks such as invoking callbacks,
 737 *	updating padding counts, waking up readers, etc.
 738 */
 739size_t relay_switch_subbuf(struct rchan_buf *buf, size_t length)
 740{
 741	void *old, *new;
 742	size_t old_subbuf, new_subbuf;
 743
 744	if (unlikely(length > buf->chan->subbuf_size))
 745		goto toobig;
 746
 747	if (buf->offset != buf->chan->subbuf_size + 1) {
 748		buf->prev_padding = buf->chan->subbuf_size - buf->offset;
 749		old_subbuf = buf->subbufs_produced % buf->chan->n_subbufs;
 750		buf->padding[old_subbuf] = buf->prev_padding;
 751		buf->subbufs_produced++;
 752		if (buf->dentry)
 753			d_inode(buf->dentry)->i_size +=
 754				buf->chan->subbuf_size -
 755				buf->padding[old_subbuf];
 756		else
 757			buf->early_bytes += buf->chan->subbuf_size -
 758					    buf->padding[old_subbuf];
 759		smp_mb();
 760		if (waitqueue_active(&buf->read_wait)) {
 761			/*
 762			 * Calling wake_up_interruptible() from here
 763			 * will deadlock if we happen to be logging
 764			 * from the scheduler (trying to re-grab
 765			 * rq->lock), so defer it.
 766			 */
 767			irq_work_queue(&buf->wakeup_work);
 768		}
 769	}
 770
 771	old = buf->data;
 772	new_subbuf = buf->subbufs_produced % buf->chan->n_subbufs;
 773	new = buf->start + new_subbuf * buf->chan->subbuf_size;
 774	buf->offset = 0;
 775	if (!buf->chan->cb->subbuf_start(buf, new, old, buf->prev_padding)) {
 776		buf->offset = buf->chan->subbuf_size + 1;
 777		return 0;
 778	}
 779	buf->data = new;
 780	buf->padding[new_subbuf] = 0;
 781
 782	if (unlikely(length + buf->offset > buf->chan->subbuf_size))
 783		goto toobig;
 784
 785	return length;
 786
 787toobig:
 788	buf->chan->last_toobig = length;
 789	return 0;
 790}
 791EXPORT_SYMBOL_GPL(relay_switch_subbuf);
 792
 793/**
 794 *	relay_subbufs_consumed - update the buffer's sub-buffers-consumed count
 795 *	@chan: the channel
 796 *	@cpu: the cpu associated with the channel buffer to update
 797 *	@subbufs_consumed: number of sub-buffers to add to current buf's count
 798 *
 799 *	Adds to the channel buffer's consumed sub-buffer count.
 800 *	subbufs_consumed should be the number of sub-buffers newly consumed,
 801 *	not the total consumed.
 802 *
 803 *	NOTE. Kernel clients don't need to call this function if the channel
 804 *	mode is 'overwrite'.
 805 */
 806void relay_subbufs_consumed(struct rchan *chan,
 807			    unsigned int cpu,
 808			    size_t subbufs_consumed)
 809{
 810	struct rchan_buf *buf;
 811
 812	if (!chan || cpu >= NR_CPUS)
 813		return;
 814
 815	buf = *per_cpu_ptr(chan->buf, cpu);
 816	if (!buf || subbufs_consumed > chan->n_subbufs)
 817		return;
 818
 819	if (subbufs_consumed > buf->subbufs_produced - buf->subbufs_consumed)
 820		buf->subbufs_consumed = buf->subbufs_produced;
 821	else
 822		buf->subbufs_consumed += subbufs_consumed;
 823}
 824EXPORT_SYMBOL_GPL(relay_subbufs_consumed);
 825
 826/**
 827 *	relay_close - close the channel
 828 *	@chan: the channel
 829 *
 830 *	Closes all channel buffers and frees the channel.
 831 */
 832void relay_close(struct rchan *chan)
 833{
 834	struct rchan_buf *buf;
 835	unsigned int i;
 836
 837	if (!chan)
 838		return;
 839
 840	mutex_lock(&relay_channels_mutex);
 841	if (chan->is_global && (buf = *per_cpu_ptr(chan->buf, 0)))
 842		relay_close_buf(buf);
 843	else
 844		for_each_possible_cpu(i)
 845			if ((buf = *per_cpu_ptr(chan->buf, i)))
 846				relay_close_buf(buf);
 847
 848	if (chan->last_toobig)
 849		printk(KERN_WARNING "relay: one or more items not logged "
 850		       "[item size (%Zd) > sub-buffer size (%Zd)]\n",
 851		       chan->last_toobig, chan->subbuf_size);
 852
 853	list_del(&chan->list);
 854	kref_put(&chan->kref, relay_destroy_channel);
 855	mutex_unlock(&relay_channels_mutex);
 856}
 857EXPORT_SYMBOL_GPL(relay_close);
 858
 859/**
 860 *	relay_flush - close the channel
 861 *	@chan: the channel
 862 *
 863 *	Flushes all channel buffers, i.e. forces buffer switch.
 864 */
 865void relay_flush(struct rchan *chan)
 866{
 867	struct rchan_buf *buf;
 868	unsigned int i;
 869
 870	if (!chan)
 871		return;
 872
 873	if (chan->is_global && (buf = *per_cpu_ptr(chan->buf, 0))) {
 874		relay_switch_subbuf(buf, 0);
 875		return;
 876	}
 877
 878	mutex_lock(&relay_channels_mutex);
 879	for_each_possible_cpu(i)
 880		if ((buf = *per_cpu_ptr(chan->buf, i)))
 881			relay_switch_subbuf(buf, 0);
 882	mutex_unlock(&relay_channels_mutex);
 883}
 884EXPORT_SYMBOL_GPL(relay_flush);
 885
 886/**
 887 *	relay_file_open - open file op for relay files
 888 *	@inode: the inode
 889 *	@filp: the file
 890 *
 891 *	Increments the channel buffer refcount.
 892 */
 893static int relay_file_open(struct inode *inode, struct file *filp)
 894{
 895	struct rchan_buf *buf = inode->i_private;
 896	kref_get(&buf->kref);
 897	filp->private_data = buf;
 898
 899	return nonseekable_open(inode, filp);
 900}
 901
 902/**
 903 *	relay_file_mmap - mmap file op for relay files
 904 *	@filp: the file
 905 *	@vma: the vma describing what to map
 906 *
 907 *	Calls upon relay_mmap_buf() to map the file into user space.
 908 */
 909static int relay_file_mmap(struct file *filp, struct vm_area_struct *vma)
 910{
 911	struct rchan_buf *buf = filp->private_data;
 912	return relay_mmap_buf(buf, vma);
 913}
 914
 915/**
 916 *	relay_file_poll - poll file op for relay files
 917 *	@filp: the file
 918 *	@wait: poll table
 919 *
 920 *	Poll implemention.
 921 */
 922static unsigned int relay_file_poll(struct file *filp, poll_table *wait)
 923{
 924	unsigned int mask = 0;
 925	struct rchan_buf *buf = filp->private_data;
 926
 927	if (buf->finalized)
 928		return POLLERR;
 929
 930	if (filp->f_mode & FMODE_READ) {
 931		poll_wait(filp, &buf->read_wait, wait);
 932		if (!relay_buf_empty(buf))
 933			mask |= POLLIN | POLLRDNORM;
 934	}
 935
 936	return mask;
 937}
 938
 939/**
 940 *	relay_file_release - release file op for relay files
 941 *	@inode: the inode
 942 *	@filp: the file
 943 *
 944 *	Decrements the channel refcount, as the filesystem is
 945 *	no longer using it.
 946 */
 947static int relay_file_release(struct inode *inode, struct file *filp)
 948{
 949	struct rchan_buf *buf = filp->private_data;
 950	kref_put(&buf->kref, relay_remove_buf);
 951
 952	return 0;
 953}
 954
 955/*
 956 *	relay_file_read_consume - update the consumed count for the buffer
 957 */
 958static void relay_file_read_consume(struct rchan_buf *buf,
 959				    size_t read_pos,
 960				    size_t bytes_consumed)
 961{
 962	size_t subbuf_size = buf->chan->subbuf_size;
 963	size_t n_subbufs = buf->chan->n_subbufs;
 964	size_t read_subbuf;
 965
 966	if (buf->subbufs_produced == buf->subbufs_consumed &&
 967	    buf->offset == buf->bytes_consumed)
 968		return;
 969
 970	if (buf->bytes_consumed + bytes_consumed > subbuf_size) {
 971		relay_subbufs_consumed(buf->chan, buf->cpu, 1);
 972		buf->bytes_consumed = 0;
 973	}
 974
 975	buf->bytes_consumed += bytes_consumed;
 976	if (!read_pos)
 977		read_subbuf = buf->subbufs_consumed % n_subbufs;
 978	else
 979		read_subbuf = read_pos / buf->chan->subbuf_size;
 980	if (buf->bytes_consumed + buf->padding[read_subbuf] == subbuf_size) {
 981		if ((read_subbuf == buf->subbufs_produced % n_subbufs) &&
 982		    (buf->offset == subbuf_size))
 983			return;
 984		relay_subbufs_consumed(buf->chan, buf->cpu, 1);
 985		buf->bytes_consumed = 0;
 986	}
 987}
 988
 989/*
 990 *	relay_file_read_avail - boolean, are there unconsumed bytes available?
 991 */
 992static int relay_file_read_avail(struct rchan_buf *buf, size_t read_pos)
 993{
 994	size_t subbuf_size = buf->chan->subbuf_size;
 995	size_t n_subbufs = buf->chan->n_subbufs;
 996	size_t produced = buf->subbufs_produced;
 997	size_t consumed = buf->subbufs_consumed;
 998
 999	relay_file_read_consume(buf, read_pos, 0);
1000
1001	consumed = buf->subbufs_consumed;
1002
1003	if (unlikely(buf->offset > subbuf_size)) {
1004		if (produced == consumed)
1005			return 0;
1006		return 1;
1007	}
1008
1009	if (unlikely(produced - consumed >= n_subbufs)) {
1010		consumed = produced - n_subbufs + 1;
1011		buf->subbufs_consumed = consumed;
1012		buf->bytes_consumed = 0;
1013	}
1014
1015	produced = (produced % n_subbufs) * subbuf_size + buf->offset;
1016	consumed = (consumed % n_subbufs) * subbuf_size + buf->bytes_consumed;
1017
1018	if (consumed > produced)
1019		produced += n_subbufs * subbuf_size;
1020
1021	if (consumed == produced) {
1022		if (buf->offset == subbuf_size &&
1023		    buf->subbufs_produced > buf->subbufs_consumed)
1024			return 1;
1025		return 0;
1026	}
1027
1028	return 1;
1029}
1030
1031/**
1032 *	relay_file_read_subbuf_avail - return bytes available in sub-buffer
1033 *	@read_pos: file read position
1034 *	@buf: relay channel buffer
1035 */
1036static size_t relay_file_read_subbuf_avail(size_t read_pos,
1037					   struct rchan_buf *buf)
1038{
1039	size_t padding, avail = 0;
1040	size_t read_subbuf, read_offset, write_subbuf, write_offset;
1041	size_t subbuf_size = buf->chan->subbuf_size;
1042
1043	write_subbuf = (buf->data - buf->start) / subbuf_size;
1044	write_offset = buf->offset > subbuf_size ? subbuf_size : buf->offset;
1045	read_subbuf = read_pos / subbuf_size;
1046	read_offset = read_pos % subbuf_size;
1047	padding = buf->padding[read_subbuf];
1048
1049	if (read_subbuf == write_subbuf) {
1050		if (read_offset + padding < write_offset)
1051			avail = write_offset - (read_offset + padding);
1052	} else
1053		avail = (subbuf_size - padding) - read_offset;
1054
1055	return avail;
1056}
1057
1058/**
1059 *	relay_file_read_start_pos - find the first available byte to read
1060 *	@read_pos: file read position
1061 *	@buf: relay channel buffer
1062 *
1063 *	If the @read_pos is in the middle of padding, return the
1064 *	position of the first actually available byte, otherwise
1065 *	return the original value.
1066 */
1067static size_t relay_file_read_start_pos(size_t read_pos,
1068					struct rchan_buf *buf)
1069{
1070	size_t read_subbuf, padding, padding_start, padding_end;
1071	size_t subbuf_size = buf->chan->subbuf_size;
1072	size_t n_subbufs = buf->chan->n_subbufs;
1073	size_t consumed = buf->subbufs_consumed % n_subbufs;
1074
1075	if (!read_pos)
1076		read_pos = consumed * subbuf_size + buf->bytes_consumed;
1077	read_subbuf = read_pos / subbuf_size;
1078	padding = buf->padding[read_subbuf];
1079	padding_start = (read_subbuf + 1) * subbuf_size - padding;
1080	padding_end = (read_subbuf + 1) * subbuf_size;
1081	if (read_pos >= padding_start && read_pos < padding_end) {
1082		read_subbuf = (read_subbuf + 1) % n_subbufs;
1083		read_pos = read_subbuf * subbuf_size;
1084	}
1085
1086	return read_pos;
1087}
1088
1089/**
1090 *	relay_file_read_end_pos - return the new read position
1091 *	@read_pos: file read position
1092 *	@buf: relay channel buffer
1093 *	@count: number of bytes to be read
1094 */
1095static size_t relay_file_read_end_pos(struct rchan_buf *buf,
1096				      size_t read_pos,
1097				      size_t count)
1098{
1099	size_t read_subbuf, padding, end_pos;
1100	size_t subbuf_size = buf->chan->subbuf_size;
1101	size_t n_subbufs = buf->chan->n_subbufs;
1102
1103	read_subbuf = read_pos / subbuf_size;
1104	padding = buf->padding[read_subbuf];
1105	if (read_pos % subbuf_size + count + padding == subbuf_size)
1106		end_pos = (read_subbuf + 1) * subbuf_size;
1107	else
1108		end_pos = read_pos + count;
1109	if (end_pos >= subbuf_size * n_subbufs)
1110		end_pos = 0;
1111
1112	return end_pos;
1113}
1114
1115static ssize_t relay_file_read(struct file *filp,
1116			       char __user *buffer,
1117			       size_t count,
1118			       loff_t *ppos)
1119{
1120	struct rchan_buf *buf = filp->private_data;
1121	size_t read_start, avail;
1122	size_t written = 0;
1123	int ret;
1124
1125	if (!count)
1126		return 0;
1127
1128	inode_lock(file_inode(filp));
1129	do {
1130		void *from;
1131
1132		if (!relay_file_read_avail(buf, *ppos))
1133			break;
1134
1135		read_start = relay_file_read_start_pos(*ppos, buf);
1136		avail = relay_file_read_subbuf_avail(read_start, buf);
1137		if (!avail)
1138			break;
1139
1140		avail = min(count, avail);
1141		from = buf->start + read_start;
1142		ret = avail;
1143		if (copy_to_user(buffer, from, avail))
1144			break;
1145
1146		buffer += ret;
1147		written += ret;
1148		count -= ret;
1149
1150		relay_file_read_consume(buf, read_start, ret);
1151		*ppos = relay_file_read_end_pos(buf, read_start, ret);
1152	} while (count);
1153	inode_unlock(file_inode(filp));
1154
1155	return written;
1156}
1157
1158static void relay_consume_bytes(struct rchan_buf *rbuf, int bytes_consumed)
1159{
1160	rbuf->bytes_consumed += bytes_consumed;
1161
1162	if (rbuf->bytes_consumed >= rbuf->chan->subbuf_size) {
1163		relay_subbufs_consumed(rbuf->chan, rbuf->cpu, 1);
1164		rbuf->bytes_consumed %= rbuf->chan->subbuf_size;
1165	}
1166}
1167
1168static void relay_pipe_buf_release(struct pipe_inode_info *pipe,
1169				   struct pipe_buffer *buf)
1170{
1171	struct rchan_buf *rbuf;
1172
1173	rbuf = (struct rchan_buf *)page_private(buf->page);
1174	relay_consume_bytes(rbuf, buf->private);
1175}
1176
1177static const struct pipe_buf_operations relay_pipe_buf_ops = {
1178	.can_merge = 0,
1179	.confirm = generic_pipe_buf_confirm,
1180	.release = relay_pipe_buf_release,
1181	.steal = generic_pipe_buf_steal,
1182	.get = generic_pipe_buf_get,
1183};
1184
1185static void relay_page_release(struct splice_pipe_desc *spd, unsigned int i)
1186{
1187}
1188
1189/*
1190 *	subbuf_splice_actor - splice up to one subbuf's worth of data
1191 */
1192static ssize_t subbuf_splice_actor(struct file *in,
1193			       loff_t *ppos,
1194			       struct pipe_inode_info *pipe,
1195			       size_t len,
1196			       unsigned int flags,
1197			       int *nonpad_ret)
1198{
1199	unsigned int pidx, poff, total_len, subbuf_pages, nr_pages;
1200	struct rchan_buf *rbuf = in->private_data;
1201	unsigned int subbuf_size = rbuf->chan->subbuf_size;
1202	uint64_t pos = (uint64_t) *ppos;
1203	uint32_t alloc_size = (uint32_t) rbuf->chan->alloc_size;
1204	size_t read_start = (size_t) do_div(pos, alloc_size);
1205	size_t read_subbuf = read_start / subbuf_size;
1206	size_t padding = rbuf->padding[read_subbuf];
1207	size_t nonpad_end = read_subbuf * subbuf_size + subbuf_size - padding;
1208	struct page *pages[PIPE_DEF_BUFFERS];
1209	struct partial_page partial[PIPE_DEF_BUFFERS];
1210	struct splice_pipe_desc spd = {
1211		.pages = pages,
1212		.nr_pages = 0,
1213		.nr_pages_max = PIPE_DEF_BUFFERS,
1214		.partial = partial,
1215		.flags = flags,
1216		.ops = &relay_pipe_buf_ops,
1217		.spd_release = relay_page_release,
1218	};
1219	ssize_t ret;
1220
1221	if (rbuf->subbufs_produced == rbuf->subbufs_consumed)
1222		return 0;
1223	if (splice_grow_spd(pipe, &spd))
1224		return -ENOMEM;
1225
1226	/*
1227	 * Adjust read len, if longer than what is available
1228	 */
1229	if (len > (subbuf_size - read_start % subbuf_size))
1230		len = subbuf_size - read_start % subbuf_size;
1231
1232	subbuf_pages = rbuf->chan->alloc_size >> PAGE_SHIFT;
1233	pidx = (read_start / PAGE_SIZE) % subbuf_pages;
1234	poff = read_start & ~PAGE_MASK;
1235	nr_pages = min_t(unsigned int, subbuf_pages, spd.nr_pages_max);
1236
1237	for (total_len = 0; spd.nr_pages < nr_pages; spd.nr_pages++) {
1238		unsigned int this_len, this_end, private;
1239		unsigned int cur_pos = read_start + total_len;
1240
1241		if (!len)
1242			break;
1243
1244		this_len = min_t(unsigned long, len, PAGE_SIZE - poff);
1245		private = this_len;
1246
1247		spd.pages[spd.nr_pages] = rbuf->page_array[pidx];
1248		spd.partial[spd.nr_pages].offset = poff;
1249
1250		this_end = cur_pos + this_len;
1251		if (this_end >= nonpad_end) {
1252			this_len = nonpad_end - cur_pos;
1253			private = this_len + padding;
1254		}
1255		spd.partial[spd.nr_pages].len = this_len;
1256		spd.partial[spd.nr_pages].private = private;
1257
1258		len -= this_len;
1259		total_len += this_len;
1260		poff = 0;
1261		pidx = (pidx + 1) % subbuf_pages;
1262
1263		if (this_end >= nonpad_end) {
1264			spd.nr_pages++;
1265			break;
1266		}
1267	}
1268
1269	ret = 0;
1270	if (!spd.nr_pages)
1271		goto out;
1272
1273	ret = *nonpad_ret = splice_to_pipe(pipe, &spd);
1274	if (ret < 0 || ret < total_len)
1275		goto out;
1276
1277        if (read_start + ret == nonpad_end)
1278                ret += padding;
1279
1280out:
1281	splice_shrink_spd(&spd);
1282	return ret;
1283}
1284
1285static ssize_t relay_file_splice_read(struct file *in,
1286				      loff_t *ppos,
1287				      struct pipe_inode_info *pipe,
1288				      size_t len,
1289				      unsigned int flags)
1290{
1291	ssize_t spliced;
1292	int ret;
1293	int nonpad_ret = 0;
1294
1295	ret = 0;
1296	spliced = 0;
1297
1298	while (len && !spliced) {
1299		ret = subbuf_splice_actor(in, ppos, pipe, len, flags, &nonpad_ret);
1300		if (ret < 0)
1301			break;
1302		else if (!ret) {
1303			if (flags & SPLICE_F_NONBLOCK)
1304				ret = -EAGAIN;
1305			break;
1306		}
1307
1308		*ppos += ret;
1309		if (ret > len)
1310			len = 0;
1311		else
1312			len -= ret;
1313		spliced += nonpad_ret;
1314		nonpad_ret = 0;
1315	}
1316
1317	if (spliced)
1318		return spliced;
1319
1320	return ret;
1321}
1322
1323const struct file_operations relay_file_operations = {
1324	.open		= relay_file_open,
1325	.poll		= relay_file_poll,
1326	.mmap		= relay_file_mmap,
1327	.read		= relay_file_read,
1328	.llseek		= no_llseek,
1329	.release	= relay_file_release,
1330	.splice_read	= relay_file_splice_read,
1331};
1332EXPORT_SYMBOL_GPL(relay_file_operations);