lib/ringbuffer/ring_buffer_backend.c

   1 /*
   2  * ring_buffer_backend.c
   3  *
   4  * Copyright (C) 2005-2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   5  *
   6  * Dual LGPL v2.1/GPL v2 license.
   7  */
   8
   9 #include <linux/stddef.h>
  10 #include <linux/module.h>
  11 #include <linux/string.h>
  12 #include <linux/bitops.h>
  13 #include <linux/delay.h>
  14 #include <linux/errno.h>
  15 #include <linux/slab.h>
  16 #include <linux/cpu.h>
  17 #include <linux/mm.h>
  18
  19 #include "../../wrapper/vmalloc.h"      /* for wrapper_vmalloc_sync_all() */
  20 #include "../../wrapper/ringbuffer/config.h"
  21 #include "../../wrapper/ringbuffer/backend.h"
  22 #include "../../wrapper/ringbuffer/frontend.h"
  23
  24 /**
  25  * lib_ring_buffer_backend_allocate - allocate a channel buffer
  26  * @config: ring buffer instance configuration
  27  * @buf: the buffer struct
  28  * @size: total size of the buffer
  29  * @num_subbuf: number of subbuffers
  30  * @extra_reader_sb: need extra subbuffer for reader
  31  */
  32 static
  33 int lib_ring_buffer_backend_allocate(const struct lib_ring_buffer_config *config,
  34                                      struct lib_ring_buffer_backend *bufb,
  35                                      size_t size, size_t num_subbuf,
  36                                      int extra_reader_sb)
  37 {
  38         struct channel_backend *chanb = &bufb->chan->backend;
  39         unsigned long j, num_pages, num_pages_per_subbuf, page_idx = 0;
  40         unsigned long subbuf_size, mmap_offset = 0;
  41         unsigned long num_subbuf_alloc;
  42         struct page **pages;
  43         void **virt;
  44         unsigned long i;
  45
  46         num_pages = size >> PAGE_SHIFT;
  47         num_pages_per_subbuf = num_pages >> get_count_order(num_subbuf);
  48         subbuf_size = chanb->subbuf_size;
  49         num_subbuf_alloc = num_subbuf;
  50
  51         if (extra_reader_sb) {
  52                 num_pages += num_pages_per_subbuf; /* Add pages for reader */
  53                 num_subbuf_alloc++;
  54         }
  55
  56         pages = kmalloc_node(ALIGN(sizeof(*pages) * num_pages,
  57                                    1 << INTERNODE_CACHE_SHIFT),
  58                         GFP_KERNEL, cpu_to_node(max(bufb->cpu, 0)));
  59         if (unlikely(!pages))
  60                 goto pages_error;
  61
  62         virt = kmalloc_node(ALIGN(sizeof(*virt) * num_pages,
  63                                   1 << INTERNODE_CACHE_SHIFT),
  64                         GFP_KERNEL, cpu_to_node(max(bufb->cpu, 0)));
  65         if (unlikely(!virt))
  66                 goto virt_error;
  67
  68         bufb->array = kmalloc_node(ALIGN(sizeof(*bufb->array)
  69                                          * num_subbuf_alloc,
  70                                   1 << INTERNODE_CACHE_SHIFT),
  71                         GFP_KERNEL, cpu_to_node(max(bufb->cpu, 0)));
  72         if (unlikely(!bufb->array))
  73                 goto array_error;
  74
  75         for (i = 0; i < num_pages; i++) {
  76                 pages[i] = alloc_pages_node(cpu_to_node(max(bufb->cpu, 0)),
  77                                             GFP_KERNEL | __GFP_ZERO, 0);
  78                 if (unlikely(!pages[i]))
  79                         goto depopulate;
  80                 virt[i] = page_address(pages[i]);
  81         }
  82         bufb->num_pages_per_subbuf = num_pages_per_subbuf;
  83
  84         /* Allocate backend pages array elements */
  85         for (i = 0; i < num_subbuf_alloc; i++) {
  86                 bufb->array[i] =
  87                         kzalloc_node(ALIGN(
  88                                 sizeof(struct lib_ring_buffer_backend_pages) +
  89                                 sizeof(struct lib_ring_buffer_backend_page)
  90                                 * num_pages_per_subbuf,
  91                                 1 << INTERNODE_CACHE_SHIFT),
  92                                 GFP_KERNEL, cpu_to_node(max(bufb->cpu, 0)));
  93                 if (!bufb->array[i])
  94                         goto free_array;
  95         }
  96
  97         /* Allocate write-side subbuffer table */
  98         bufb->buf_wsb = kzalloc_node(ALIGN(
  99                                 sizeof(struct lib_ring_buffer_backend_subbuffer)
 100                                 * num_subbuf,
 101                                 1 << INTERNODE_CACHE_SHIFT),
 102                                 GFP_KERNEL, cpu_to_node(max(bufb->cpu, 0)));
 103         if (unlikely(!bufb->buf_wsb))
 104                 goto free_array;
 105
 106         for (i = 0; i < num_subbuf; i++)
 107                 bufb->buf_wsb[i].id = subbuffer_id(config, 0, 1, i);
 108
 109         /* Assign read-side subbuffer table */
 110         if (extra_reader_sb)
 111                 bufb->buf_rsb.id = subbuffer_id(config, 0, 1,
 112                                                 num_subbuf_alloc - 1);
 113         else
 114                 bufb->buf_rsb.id = subbuffer_id(config, 0, 1, 0);
 115
 116         /* Assign pages to page index */
 117         for (i = 0; i < num_subbuf_alloc; i++) {
 118                 for (j = 0; j < num_pages_per_subbuf; j++) {
 119                         CHAN_WARN_ON(chanb, page_idx > num_pages);
 120                         bufb->array[i]->p[j].virt = virt[page_idx];
 121                         bufb->array[i]->p[j].page = pages[page_idx];
 122                         page_idx++;
 123                 }
 124                 if (config->output == RING_BUFFER_MMAP) {
 125                         bufb->array[i]->mmap_offset = mmap_offset;
 126                         mmap_offset += subbuf_size;
 127                 }
 128         }
 129
 130         /*
 131          * If kmalloc ever uses vmalloc underneath, make sure the buffer pages
 132          * will not fault.
 133          */
 134         wrapper_vmalloc_sync_all();
 135         kfree(virt);
 136         kfree(pages);
 137         return 0;
 138
 139 free_array:
 140         for (i = 0; (i < num_subbuf_alloc && bufb->array[i]); i++)
 141                 kfree(bufb->array[i]);
 142 depopulate:
 143         /* Free all allocated pages */
 144         for (i = 0; (i < num_pages && pages[i]); i++)
 145                 __free_page(pages[i]);
 146         kfree(bufb->array);
 147 array_error:
 148         kfree(virt);
 149 virt_error:
 150         kfree(pages);
 151 pages_error:
 152         return -ENOMEM;
 153 }
 154
 155 int lib_ring_buffer_backend_create(struct lib_ring_buffer_backend *bufb,
 156                                    struct channel_backend *chanb, int cpu)
 157 {
 158         const struct lib_ring_buffer_config *config = &chanb->config;
 159
 160         bufb->chan = container_of(chanb, struct channel, backend);
 161         bufb->cpu = cpu;
 162
 163         return lib_ring_buffer_backend_allocate(config, bufb, chanb->buf_size,
 164                                                 chanb->num_subbuf,
 165                                                 chanb->extra_reader_sb);
 166 }
 167
 168 void lib_ring_buffer_backend_free(struct lib_ring_buffer_backend *bufb)
 169 {
 170         struct channel_backend *chanb = &bufb->chan->backend;
 171         unsigned long i, j, num_subbuf_alloc;
 172
 173         num_subbuf_alloc = chanb->num_subbuf;
 174         if (chanb->extra_reader_sb)
 175                 num_subbuf_alloc++;
 176
 177         kfree(bufb->buf_wsb);
 178         for (i = 0; i < num_subbuf_alloc; i++) {
 179                 for (j = 0; j < bufb->num_pages_per_subbuf; j++)
 180                         __free_page(bufb->array[i]->p[j].page);
 181                 kfree(bufb->array[i]);
 182         }
 183         kfree(bufb->array);
 184         bufb->allocated = 0;
 185 }
 186
 187 void lib_ring_buffer_backend_reset(struct lib_ring_buffer_backend *bufb)
 188 {
 189         struct channel_backend *chanb = &bufb->chan->backend;
 190         const struct lib_ring_buffer_config *config = &chanb->config;
 191         unsigned long num_subbuf_alloc;
 192         unsigned int i;
 193
 194         num_subbuf_alloc = chanb->num_subbuf;
 195         if (chanb->extra_reader_sb)
 196                 num_subbuf_alloc++;
 197
 198         for (i = 0; i < chanb->num_subbuf; i++)
 199                 bufb->buf_wsb[i].id = subbuffer_id(config, 0, 1, i);
 200         if (chanb->extra_reader_sb)
 201                 bufb->buf_rsb.id = subbuffer_id(config, 0, 1,
 202                                                 num_subbuf_alloc - 1);
 203         else
 204                 bufb->buf_rsb.id = subbuffer_id(config, 0, 1, 0);
 205
 206         for (i = 0; i < num_subbuf_alloc; i++) {
 207                 /* Don't reset mmap_offset */
 208                 v_set(config, &bufb->array[i]->records_commit, 0);
 209                 v_set(config, &bufb->array[i]->records_unread, 0);
 210                 bufb->array[i]->data_size = 0;
 211                 /* Don't reset backend page and virt addresses */
 212         }
 213         /* Don't reset num_pages_per_subbuf, cpu, allocated */
 214         v_set(config, &bufb->records_read, 0);
 215 }
 216
 217 /*
 218  * The frontend is responsible for also calling ring_buffer_backend_reset for
 219  * each buffer when calling channel_backend_reset.
 220  */
 221 void channel_backend_reset(struct channel_backend *chanb)
 222 {
 223         struct channel *chan = container_of(chanb, struct channel, backend);
 224         const struct lib_ring_buffer_config *config = &chanb->config;
 225
 226         /*
 227          * Don't reset buf_size, subbuf_size, subbuf_size_order,
 228          * num_subbuf_order, buf_size_order, extra_reader_sb, num_subbuf,
 229          * priv, notifiers, config, cpumask and name.
 230          */
 231         chanb->start_tsc = config->cb.ring_buffer_clock_read(chan);
 232 }
 233
 234 #ifdef CONFIG_HOTPLUG_CPU
 235 /**
 236  *      lib_ring_buffer_cpu_hp_callback - CPU hotplug callback
 237  *      @nb: notifier block
 238  *      @action: hotplug action to take
 239  *      @hcpu: CPU number
 240  *
 241  *      Returns the success/failure of the operation. (%NOTIFY_OK, %NOTIFY_BAD)
 242  */
 243 static
 244 int __cpuinit lib_ring_buffer_cpu_hp_callback(struct notifier_block *nb,
 245                                               unsigned long action,
 246                                               void *hcpu)
 247 {
 248         unsigned int cpu = (unsigned long)hcpu;
 249         struct channel_backend *chanb = container_of(nb, struct channel_backend,
 250                                                      cpu_hp_notifier);
 251         const struct lib_ring_buffer_config *config = &chanb->config;
 252         struct lib_ring_buffer *buf;
 253         int ret;
 254
 255         CHAN_WARN_ON(chanb, config->alloc == RING_BUFFER_ALLOC_GLOBAL);
 256
 257         switch (action) {
 258         case CPU_UP_PREPARE:
 259         case CPU_UP_PREPARE_FROZEN:
 260                 buf = per_cpu_ptr(chanb->buf, cpu);
 261                 ret = lib_ring_buffer_create(buf, chanb, cpu);
 262                 if (ret) {
 263                         printk(KERN_ERR
 264                           "ring_buffer_cpu_hp_callback: cpu %d "
 265                           "buffer creation failed\n", cpu);
 266                         return NOTIFY_BAD;
 267                 }
 268                 break;
 269         case CPU_DEAD:
 270         case CPU_DEAD_FROZEN:
 271                 /* No need to do a buffer switch here, because it will happen
 272                  * when tracing is stopped, or will be done by switch timer CPU
 273                  * DEAD callback. */
 274                 break;
 275         }
 276         return NOTIFY_OK;
 277 }
 278 #endif
 279
 280 /**
 281  * channel_backend_init - initialize a channel backend
 282  * @chanb: channel backend
 283  * @name: channel name
 284  * @config: client ring buffer configuration
 285  * @priv: client private data
 286  * @parent: dentry of parent directory, %NULL for root directory
 287  * @subbuf_size: size of sub-buffers (> PAGE_SIZE, power of 2)
 288  * @num_subbuf: number of sub-buffers (power of 2)
 289  *
 290  * Returns channel pointer if successful, %NULL otherwise.
 291  *
 292  * Creates per-cpu channel buffers using the sizes and attributes
 293  * specified.  The created channel buffer files will be named
 294  * name_0...name_N-1.  File permissions will be %S_IRUSR.
 295  *
 296  * Called with CPU hotplug disabled.
 297  */
 298 int channel_backend_init(struct channel_backend *chanb,
 299                          const char *name,
 300                          const struct lib_ring_buffer_config *config,
 301                          void *priv, size_t subbuf_size, size_t num_subbuf)
 302 {
 303         struct channel *chan = container_of(chanb, struct channel, backend);
 304         unsigned int i;
 305         int ret;
 306
 307         if (!name)
 308                 return -EPERM;
 309
 310         if (!(subbuf_size && num_subbuf))
 311                 return -EPERM;
 312
 313         /* Check that the subbuffer size is larger than a page. */
 314         if (subbuf_size < PAGE_SIZE)
 315                 return -EINVAL;
 316
 317         /*
 318          * Make sure the number of subbuffers and subbuffer size are power of 2.
 319          */
 320         if (hweight32(subbuf_size) != 1)
 321                 return -EINVAL;
 322         if (hweight32(num_subbuf) != 1)
 323                 return -EINVAL;
 324
 325         ret = subbuffer_id_check_index(config, num_subbuf);
 326         if (ret)
 327                 return ret;
 328
 329         chanb->priv = priv;
 330         chanb->buf_size = num_subbuf * subbuf_size;
 331         chanb->subbuf_size = subbuf_size;
 332         chanb->buf_size_order = get_count_order(chanb->buf_size);
 333         chanb->subbuf_size_order = get_count_order(subbuf_size);
 334         chanb->num_subbuf_order = get_count_order(num_subbuf);
 335         chanb->extra_reader_sb =
 336                         (config->mode == RING_BUFFER_OVERWRITE) ? 1 : 0;
 337         chanb->num_subbuf = num_subbuf;
 338         strlcpy(chanb->name, name, NAME_MAX);
 339         memcpy(&chanb->config, config, sizeof(chanb->config));
 340
 341         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 342                 if (!zalloc_cpumask_var(&chanb->cpumask, GFP_KERNEL))
 343                         return -ENOMEM;
 344         }
 345
 346         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 347                 /* Allocating the buffer per-cpu structures */
 348                 chanb->buf = alloc_percpu(struct lib_ring_buffer);
 349                 if (!chanb->buf)
 350                         goto free_cpumask;
 351
 352                 /*
 353                  * In case of non-hotplug cpu, if the ring-buffer is allocated
 354                  * in early initcall, it will not be notified of secondary cpus.
 355                  * In that off case, we need to allocate for all possible cpus.
 356                  */
 357 #ifdef CONFIG_HOTPLUG_CPU
 358                 /*
 359                  * buf->backend.allocated test takes care of concurrent CPU
 360                  * hotplug.
 361                  * Priority higher than frontend, so we create the ring buffer
 362                  * before we start the timer.
 363                  */
 364                 chanb->cpu_hp_notifier.notifier_call =
 365                                 lib_ring_buffer_cpu_hp_callback;
 366                 chanb->cpu_hp_notifier.priority = 5;
 367                 register_hotcpu_notifier(&chanb->cpu_hp_notifier);
 368
 369                 get_online_cpus();
 370                 for_each_online_cpu(i) {
 371                         ret = lib_ring_buffer_create(per_cpu_ptr(chanb->buf, i),
 372                                                  chanb, i);
 373                         if (ret)
 374                                 goto free_bufs; /* cpu hotplug locked */
 375                 }
 376                 put_online_cpus();
 377 #else
 378                 for_each_possible_cpu(i) {
 379                         ret = lib_ring_buffer_create(per_cpu_ptr(chanb->buf, i),
 380                                                  chanb, i);
 381                         if (ret)
 382                                 goto free_bufs; /* cpu hotplug locked */
 383                 }
 384 #endif
 385         } else {
 386                 chanb->buf = kzalloc(sizeof(struct lib_ring_buffer), GFP_KERNEL);
 387                 if (!chanb->buf)
 388                         goto free_cpumask;
 389                 ret = lib_ring_buffer_create(chanb->buf, chanb, -1);
 390                 if (ret)
 391                         goto free_bufs;
 392         }
 393         chanb->start_tsc = config->cb.ring_buffer_clock_read(chan);
 394
 395         return 0;
 396
 397 free_bufs:
 398         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 399                 for_each_possible_cpu(i) {
 400                         struct lib_ring_buffer *buf = per_cpu_ptr(chanb->buf, i);
 401
 402                         if (!buf->backend.allocated)
 403                                 continue;
 404                         lib_ring_buffer_free(buf);
 405                 }
 406 #ifdef CONFIG_HOTPLUG_CPU
 407                 put_online_cpus();
 408 #endif
 409                 free_percpu(chanb->buf);
 410         } else
 411                 kfree(chanb->buf);
 412 free_cpumask:
 413         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
 414                 free_cpumask_var(chanb->cpumask);
 415         return -ENOMEM;
 416 }
 417
 418 /**
 419  * channel_backend_unregister_notifiers - unregister notifiers
 420  * @chan: the channel
 421  *
 422  * Holds CPU hotplug.
 423  */
 424 void channel_backend_unregister_notifiers(struct channel_backend *chanb)
 425 {
 426         const struct lib_ring_buffer_config *config = &chanb->config;
 427
 428         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
 429                 unregister_hotcpu_notifier(&chanb->cpu_hp_notifier);
 430 }
 431
 432 /**
 433  * channel_backend_free - destroy the channel
 434  * @chan: the channel
 435  *
 436  * Destroy all channel buffers and frees the channel.
 437  */
 438 void channel_backend_free(struct channel_backend *chanb)
 439 {
 440         const struct lib_ring_buffer_config *config = &chanb->config;
 441         unsigned int i;
 442
 443         if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 444                 for_each_possible_cpu(i) {
 445                         struct lib_ring_buffer *buf = per_cpu_ptr(chanb->buf, i);
 446
 447                         if (!buf->backend.allocated)
 448                                 continue;
 449                         lib_ring_buffer_free(buf);
 450                 }
 451                 free_cpumask_var(chanb->cpumask);
 452                 free_percpu(chanb->buf);
 453         } else {
 454                 struct lib_ring_buffer *buf = chanb->buf;
 455
 456                 CHAN_WARN_ON(chanb, !buf->backend.allocated);
 457                 lib_ring_buffer_free(buf);
 458                 kfree(buf);
 459         }
 460 }
 461
 462 /**
 463  * lib_ring_buffer_write - write data to a ring_buffer buffer.
 464  * @bufb : buffer backend
 465  * @offset : offset within the buffer
 466  * @src : source address
 467  * @len : length to write
 468  * @pagecpy : page size copied so far
 469  */
 470 void _lib_ring_buffer_write(struct lib_ring_buffer_backend *bufb, size_t offset,
 471                             const void *src, size_t len, ssize_t pagecpy)
 472 {
 473         struct channel_backend *chanb = &bufb->chan->backend;
 474         const struct lib_ring_buffer_config *config = &chanb->config;
 475         size_t sbidx, index;
 476         struct lib_ring_buffer_backend_pages *rpages;
 477         unsigned long sb_bindex, id;
 478
 479         do {
 480                 len -= pagecpy;
 481                 src += pagecpy;
 482                 offset += pagecpy;
 483                 sbidx = offset >> chanb->subbuf_size_order;
 484                 index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
 485
 486                 /*
 487                  * Underlying layer should never ask for writes across
 488                  * subbuffers.
 489                  */
 490                 CHAN_WARN_ON(chanb, offset >= chanb->buf_size);
 491
 492                 pagecpy = min_t(size_t, len, PAGE_SIZE - (offset & ~PAGE_MASK));
 493                 id = bufb->buf_wsb[sbidx].id;
 494                 sb_bindex = subbuffer_id_get_index(config, id);
 495                 rpages = bufb->array[sb_bindex];
 496                 CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 497                              && subbuffer_id_is_noref(config, id));
 498                 lib_ring_buffer_do_copy(config,
 499                                         rpages->p[index].virt
 500                                                 + (offset & ~PAGE_MASK),
 501                                         src, pagecpy);
 502         } while (unlikely(len != pagecpy));
 503 }
 504 EXPORT_SYMBOL_GPL(_lib_ring_buffer_write);
 505
 506
 507 /**
 508  * lib_ring_buffer_memset - write len bytes of c to a ring_buffer buffer.
 509  * @bufb : buffer backend
 510  * @offset : offset within the buffer
 511  * @c : the byte to write
 512  * @len : length to write
 513  * @pagecpy : page size copied so far
 514  */
 515 void _lib_ring_buffer_memset(struct lib_ring_buffer_backend *bufb,
 516                              size_t offset,
 517                              int c, size_t len, ssize_t pagecpy)
 518 {
 519         struct channel_backend *chanb = &bufb->chan->backend;
 520         const struct lib_ring_buffer_config *config = &chanb->config;
 521         size_t sbidx, index;
 522         struct lib_ring_buffer_backend_pages *rpages;
 523         unsigned long sb_bindex, id;
 524
 525         do {
 526                 len -= pagecpy;
 527                 offset += pagecpy;
 528                 sbidx = offset >> chanb->subbuf_size_order;
 529                 index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
 530
 531                 /*
 532                  * Underlying layer should never ask for writes across
 533                  * subbuffers.
 534                  */
 535                 CHAN_WARN_ON(chanb, offset >= chanb->buf_size);
 536
 537                 pagecpy = min_t(size_t, len, PAGE_SIZE - (offset & ~PAGE_MASK));
 538                 id = bufb->buf_wsb[sbidx].id;
 539                 sb_bindex = subbuffer_id_get_index(config, id);
 540                 rpages = bufb->array[sb_bindex];
 541                 CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 542                              && subbuffer_id_is_noref(config, id));
 543                 lib_ring_buffer_do_memset(rpages->p[index].virt
 544                                           + (offset & ~PAGE_MASK),
 545                                           c, pagecpy);
 546         } while (unlikely(len != pagecpy));
 547 }
 548 EXPORT_SYMBOL_GPL(_lib_ring_buffer_memset);
 549
 550
 551 /**
 552  * lib_ring_buffer_copy_from_user - write user data to a ring_buffer buffer.
 553  * @bufb : buffer backend
 554  * @offset : offset within the buffer
 555  * @src : source address
 556  * @len : length to write
 557  * @pagecpy : page size copied so far
 558  *
 559  * This function deals with userspace pointers, it should never be called
 560  * directly without having the src pointer checked with access_ok()
 561  * previously.
 562  */
 563 void _lib_ring_buffer_copy_from_user(struct lib_ring_buffer_backend *bufb,
 564                                       size_t offset,
 565                                       const void __user *src, size_t len,
 566                                       ssize_t pagecpy)
 567 {
 568         struct channel_backend *chanb = &bufb->chan->backend;
 569         const struct lib_ring_buffer_config *config = &chanb->config;
 570         size_t sbidx, index;
 571         struct lib_ring_buffer_backend_pages *rpages;
 572         unsigned long sb_bindex, id;
 573         int ret;
 574
 575         do {
 576                 len -= pagecpy;
 577                 src += pagecpy;
 578                 offset += pagecpy;
 579                 sbidx = offset >> chanb->subbuf_size_order;
 580                 index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
 581
 582                 /*
 583                  * Underlying layer should never ask for writes across
 584                  * subbuffers.
 585                  */
 586                 CHAN_WARN_ON(chanb, offset >= chanb->buf_size);
 587
 588                 pagecpy = min_t(size_t, len, PAGE_SIZE - (offset & ~PAGE_MASK));
 589                 id = bufb->buf_wsb[sbidx].id;
 590                 sb_bindex = subbuffer_id_get_index(config, id);
 591                 rpages = bufb->array[sb_bindex];
 592                 CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 593                                 && subbuffer_id_is_noref(config, id));
 594                 ret = lib_ring_buffer_do_copy_from_user(rpages->p[index].virt
 595                                                         + (offset & ~PAGE_MASK),
 596                                                         src, pagecpy) != 0;
 597                 if (ret > 0) {
 598                         offset += (pagecpy - ret);
 599                         len -= (pagecpy - ret);
 600                         _lib_ring_buffer_memset(bufb, offset, 0, len, 0);
 601                         break; /* stop copy */
 602                 }
 603         } while (unlikely(len != pagecpy));
 604 }
 605 EXPORT_SYMBOL_GPL(_lib_ring_buffer_copy_from_user);
 606
 607 /**
 608  * lib_ring_buffer_read - read data from ring_buffer_buffer.
 609  * @bufb : buffer backend
 610  * @offset : offset within the buffer
 611  * @dest : destination address
 612  * @len : length to copy to destination
 613  *
 614  * Should be protected by get_subbuf/put_subbuf.
 615  * Returns the length copied.
 616  */
 617 size_t lib_ring_buffer_read(struct lib_ring_buffer_backend *bufb, size_t offset,
 618                             void *dest, size_t len)
 619 {
 620         struct channel_backend *chanb = &bufb->chan->backend;
 621         const struct lib_ring_buffer_config *config = &chanb->config;
 622         size_t index;
 623         ssize_t pagecpy, orig_len;
 624         struct lib_ring_buffer_backend_pages *rpages;
 625         unsigned long sb_bindex, id;
 626
 627         orig_len = len;
 628         offset &= chanb->buf_size - 1;
 629         index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
 630         if (unlikely(!len))
 631                 return 0;
 632         for (;;) {
 633                 pagecpy = min_t(size_t, len, PAGE_SIZE - (offset & ~PAGE_MASK));
 634                 id = bufb->buf_rsb.id;
 635                 sb_bindex = subbuffer_id_get_index(config, id);
 636                 rpages = bufb->array[sb_bindex];
 637                 CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 638                              && subbuffer_id_is_noref(config, id));
 639                 memcpy(dest, rpages->p[index].virt + (offset & ~PAGE_MASK),
 640                        pagecpy);
 641                 len -= pagecpy;
 642                 if (likely(!len))
 643                         break;
 644                 dest += pagecpy;
 645                 offset += pagecpy;
 646                 index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
 647                 /*
 648                  * Underlying layer should never ask for reads across
 649                  * subbuffers.
 650                  */
 651                 CHAN_WARN_ON(chanb, offset >= chanb->buf_size);
 652         }
 653         return orig_len;
 654 }
 655 EXPORT_SYMBOL_GPL(lib_ring_buffer_read);
 656
 657 /**
 658  * __lib_ring_buffer_copy_to_user - read data from ring_buffer to userspace
 659  * @bufb : buffer backend
 660  * @offset : offset within the buffer
 661  * @dest : destination userspace address
 662  * @len : length to copy to destination
 663  *
 664  * Should be protected by get_subbuf/put_subbuf.
 665  * access_ok() must have been performed on dest addresses prior to call this
 666  * function.
 667  * Returns -EFAULT on error, 0 if ok.
 668  */
 669 int __lib_ring_buffer_copy_to_user(struct lib_ring_buffer_backend *bufb,
 670                                    size_t offset, void __user *dest, size_t len)
 671 {
 672         struct channel_backend *chanb = &bufb->chan->backend;
 673         const struct lib_ring_buffer_config *config = &chanb->config;
 674         size_t index;
 675         ssize_t pagecpy;
 676         struct lib_ring_buffer_backend_pages *rpages;
 677         unsigned long sb_bindex, id;
 678
 679         offset &= chanb->buf_size - 1;
 680         index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
 681         if (unlikely(!len))
 682                 return 0;
 683         for (;;) {
 684                 pagecpy = min_t(size_t, len, PAGE_SIZE - (offset & ~PAGE_MASK));
 685                 id = bufb->buf_rsb.id;
 686                 sb_bindex = subbuffer_id_get_index(config, id);
 687                 rpages = bufb->array[sb_bindex];
 688                 CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 689                              && subbuffer_id_is_noref(config, id));
 690                 if (__copy_to_user(dest,
 691                                rpages->p[index].virt + (offset & ~PAGE_MASK),
 692                                pagecpy))
 693                         return -EFAULT;
 694                 len -= pagecpy;
 695                 if (likely(!len))
 696                         break;
 697                 dest += pagecpy;
 698                 offset += pagecpy;
 699                 index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
 700                 /*
 701                  * Underlying layer should never ask for reads across
 702                  * subbuffers.
 703                  */
 704                 CHAN_WARN_ON(chanb, offset >= chanb->buf_size);
 705         }
 706         return 0;
 707 }
 708 EXPORT_SYMBOL_GPL(__lib_ring_buffer_copy_to_user);
 709
 710 /**
 711  * lib_ring_buffer_read_cstr - read a C-style string from ring_buffer.
 712  * @bufb : buffer backend
 713  * @offset : offset within the buffer
 714  * @dest : destination address
 715  * @len : destination's length
 716  *
 717  * return string's length
 718  * Should be protected by get_subbuf/put_subbuf.
 719  */
 720 int lib_ring_buffer_read_cstr(struct lib_ring_buffer_backend *bufb, size_t offset,
 721                               void *dest, size_t len)
 722 {
 723         struct channel_backend *chanb = &bufb->chan->backend;
 724         const struct lib_ring_buffer_config *config = &chanb->config;
 725         size_t index;
 726         ssize_t pagecpy, pagelen, strpagelen, orig_offset;
 727         char *str;
 728         struct lib_ring_buffer_backend_pages *rpages;
 729         unsigned long sb_bindex, id;
 730
 731         offset &= chanb->buf_size - 1;
 732         index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
 733         orig_offset = offset;
 734         for (;;) {
 735                 id = bufb->buf_rsb.id;
 736                 sb_bindex = subbuffer_id_get_index(config, id);
 737                 rpages = bufb->array[sb_bindex];
 738                 CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 739                              && subbuffer_id_is_noref(config, id));
 740                 str = (char *)rpages->p[index].virt + (offset & ~PAGE_MASK);
 741                 pagelen = PAGE_SIZE - (offset & ~PAGE_MASK);
 742                 strpagelen = strnlen(str, pagelen);
 743                 if (len) {
 744                         pagecpy = min_t(size_t, len, strpagelen);
 745                         if (dest) {
 746                                 memcpy(dest, str, pagecpy);
 747                                 dest += pagecpy;
 748                         }
 749                         len -= pagecpy;
 750                 }
 751                 offset += strpagelen;
 752                 index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
 753                 if (strpagelen < pagelen)
 754                         break;
 755                 /*
 756                  * Underlying layer should never ask for reads across
 757                  * subbuffers.
 758                  */
 759                 CHAN_WARN_ON(chanb, offset >= chanb->buf_size);
 760         }
 761         if (dest && len)
 762                 ((char *)dest)[0] = 0;
 763         return offset - orig_offset;
 764 }
 765 EXPORT_SYMBOL_GPL(lib_ring_buffer_read_cstr);
 766
 767 /**
 768  * lib_ring_buffer_read_get_page - Get a whole page to read from
 769  * @bufb : buffer backend
 770  * @offset : offset within the buffer
 771  * @virt : pointer to page address (output)
 772  *
 773  * Should be protected by get_subbuf/put_subbuf.
 774  * Returns the pointer to the page struct pointer.
 775  */
 776 struct page **lib_ring_buffer_read_get_page(struct lib_ring_buffer_backend *bufb,
 777                                             size_t offset, void ***virt)
 778 {
 779         size_t index;
 780         struct lib_ring_buffer_backend_pages *rpages;
 781         struct channel_backend *chanb = &bufb->chan->backend;
 782         const struct lib_ring_buffer_config *config = &chanb->config;
 783         unsigned long sb_bindex, id;
 784
 785         offset &= chanb->buf_size - 1;
 786         index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
 787         id = bufb->buf_rsb.id;
 788         sb_bindex = subbuffer_id_get_index(config, id);
 789         rpages = bufb->array[sb_bindex];
 790         CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 791                      && subbuffer_id_is_noref(config, id));
 792         *virt = &rpages->p[index].virt;
 793         return &rpages->p[index].page;
 794 }
 795 EXPORT_SYMBOL_GPL(lib_ring_buffer_read_get_page);
 796
 797 /**
 798  * lib_ring_buffer_read_offset_address - get address of a buffer location
 799  * @bufb : buffer backend
 800  * @offset : offset within the buffer.
 801  *
 802  * Return the address where a given offset is located (for read).
 803  * Should be used to get the current subbuffer header pointer. Given we know
 804  * it's never on a page boundary, it's safe to write directly to this address,
 805  * as long as the write is never bigger than a page size.
 806  */
 807 void *lib_ring_buffer_read_offset_address(struct lib_ring_buffer_backend *bufb,
 808                                           size_t offset)
 809 {
 810         size_t index;
 811         struct lib_ring_buffer_backend_pages *rpages;
 812         struct channel_backend *chanb = &bufb->chan->backend;
 813         const struct lib_ring_buffer_config *config = &chanb->config;
 814         unsigned long sb_bindex, id;
 815
 816         offset &= chanb->buf_size - 1;
 817         index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
 818         id = bufb->buf_rsb.id;
 819         sb_bindex = subbuffer_id_get_index(config, id);
 820         rpages = bufb->array[sb_bindex];
 821         CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 822                      && subbuffer_id_is_noref(config, id));
 823         return rpages->p[index].virt + (offset & ~PAGE_MASK);
 824 }
 825 EXPORT_SYMBOL_GPL(lib_ring_buffer_read_offset_address);
 826
 827 /**
 828  * lib_ring_buffer_offset_address - get address of a location within the buffer
 829  * @bufb : buffer backend
 830  * @offset : offset within the buffer.
 831  *
 832  * Return the address where a given offset is located.
 833  * Should be used to get the current subbuffer header pointer. Given we know
 834  * it's always at the beginning of a page, it's safe to write directly to this
 835  * address, as long as the write is never bigger than a page size.
 836  */
 837 void *lib_ring_buffer_offset_address(struct lib_ring_buffer_backend *bufb,
 838                                      size_t offset)
 839 {
 840         size_t sbidx, index;
 841         struct lib_ring_buffer_backend_pages *rpages;
 842         struct channel_backend *chanb = &bufb->chan->backend;
 843         const struct lib_ring_buffer_config *config = &chanb->config;
 844         unsigned long sb_bindex, id;
 845
 846         offset &= chanb->buf_size - 1;
 847         sbidx = offset >> chanb->subbuf_size_order;
 848         index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
 849         id = bufb->buf_wsb[sbidx].id;
 850         sb_bindex = subbuffer_id_get_index(config, id);
 851         rpages = bufb->array[sb_bindex];
 852         CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 853                      && subbuffer_id_is_noref(config, id));
 854         return rpages->p[index].virt + (offset & ~PAGE_MASK);
 855 }
 856 EXPORT_SYMBOL_GPL(lib_ring_buffer_offset_address);