X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=lib%2Fringbuffer%2Fring_buffer_backend.c;h=c7f2fe9d7b6226b33738d2fbabb379b42999ef93;hb=769b62a02f62ff8db93f109d9f63a1df6ea9990b;hp=8c9d2b7c54e7b583e896fec159349b2336a0fb21;hpb=61eb4c393f32dbea95e9125c6ec4d8cdad535f70;p=lttng-modules.git diff --git a/lib/ringbuffer/ring_buffer_backend.c b/lib/ringbuffer/ring_buffer_backend.c index 8c9d2b7c..c7f2fe9d 100644 --- a/lib/ringbuffer/ring_buffer_backend.c +++ b/lib/ringbuffer/ring_buffer_backend.c @@ -28,10 +28,10 @@ #include #include -#include "../../wrapper/vmalloc.h" /* for wrapper_vmalloc_sync_all() */ -#include "../../wrapper/ringbuffer/config.h" -#include "../../wrapper/ringbuffer/backend.h" -#include "../../wrapper/ringbuffer/frontend.h" +#include /* for wrapper_vmalloc_sync_all() */ +#include +#include +#include /** * lib_ring_buffer_backend_allocate - allocate a channel buffer @@ -52,7 +52,6 @@ int lib_ring_buffer_backend_allocate(const struct lib_ring_buffer_config *config unsigned long subbuf_size, mmap_offset = 0; unsigned long num_subbuf_alloc; struct page **pages; - void **virt; unsigned long i; num_pages = size >> PAGE_SHIFT; @@ -71,12 +70,6 @@ int lib_ring_buffer_backend_allocate(const struct lib_ring_buffer_config *config if (unlikely(!pages)) goto pages_error; - virt = kmalloc_node(ALIGN(sizeof(*virt) * num_pages, - 1 << INTERNODE_CACHE_SHIFT), - GFP_KERNEL, cpu_to_node(max(bufb->cpu, 0))); - if (unlikely(!virt)) - goto virt_error; - bufb->array = kmalloc_node(ALIGN(sizeof(*bufb->array) * num_subbuf_alloc, 1 << INTERNODE_CACHE_SHIFT), @@ -89,7 +82,6 @@ int lib_ring_buffer_backend_allocate(const struct lib_ring_buffer_config *config GFP_KERNEL | __GFP_ZERO, 0); if (unlikely(!pages[i])) goto depopulate; - virt[i] = page_address(pages[i]); } bufb->num_pages_per_subbuf = num_pages_per_subbuf; @@ -125,12 +117,21 @@ int lib_ring_buffer_backend_allocate(const struct lib_ring_buffer_config *config else bufb->buf_rsb.id = subbuffer_id(config, 0, 1, 0); + /* Allocate subbuffer packet counter table */ + bufb->buf_cnt = kzalloc_node(ALIGN( + sizeof(struct lib_ring_buffer_backend_counts) + * num_subbuf, + 1 << INTERNODE_CACHE_SHIFT), + GFP_KERNEL, cpu_to_node(max(bufb->cpu, 0))); + if (unlikely(!bufb->buf_cnt)) + goto free_wsb; + /* Assign pages to page index */ for (i = 0; i < num_subbuf_alloc; i++) { for (j = 0; j < num_pages_per_subbuf; j++) { CHAN_WARN_ON(chanb, page_idx > num_pages); - bufb->array[i]->p[j].virt = virt[page_idx]; - bufb->array[i]->p[j].page = pages[page_idx]; + bufb->array[i]->p[j].virt = page_address(pages[page_idx]); + bufb->array[i]->p[j].pfn = page_to_pfn(pages[page_idx]); page_idx++; } if (config->output == RING_BUFFER_MMAP) { @@ -144,10 +145,11 @@ int lib_ring_buffer_backend_allocate(const struct lib_ring_buffer_config *config * will not fault. */ wrapper_vmalloc_sync_all(); - kfree(virt); kfree(pages); return 0; +free_wsb: + kfree(bufb->buf_wsb); free_array: for (i = 0; (i < num_subbuf_alloc && bufb->array[i]); i++) kfree(bufb->array[i]); @@ -157,8 +159,6 @@ depopulate: __free_page(pages[i]); kfree(bufb->array); array_error: - kfree(virt); -virt_error: kfree(pages); pages_error: return -ENOMEM; @@ -187,9 +187,10 @@ void lib_ring_buffer_backend_free(struct lib_ring_buffer_backend *bufb) num_subbuf_alloc++; kfree(bufb->buf_wsb); + kfree(bufb->buf_cnt); for (i = 0; i < num_subbuf_alloc; i++) { for (j = 0; j < bufb->num_pages_per_subbuf; j++) - __free_page(bufb->array[i]->p[j].page); + __free_page(pfn_to_page(bufb->array[i]->p[j].pfn)); kfree(bufb->array[i]); } kfree(bufb->array); @@ -331,6 +332,12 @@ int channel_backend_init(struct channel_backend *chanb, return -EINVAL; if (!num_subbuf || (num_subbuf & (num_subbuf - 1))) return -EINVAL; + /* + * Overwrite mode buffers require at least 2 subbuffers per + * buffer. + */ + if (config->mode == RING_BUFFER_OVERWRITE && num_subbuf < 2) + return -EINVAL; ret = subbuffer_id_check_index(config, num_subbuf); if (ret) @@ -415,6 +422,7 @@ free_bufs: } #ifdef CONFIG_HOTPLUG_CPU put_online_cpus(); + unregister_hotcpu_notifier(&chanb->cpu_hp_notifier); #endif free_percpu(chanb->buf); } else @@ -478,7 +486,7 @@ void channel_backend_free(struct channel_backend *chanb) * @pagecpy : page size copied so far */ void _lib_ring_buffer_write(struct lib_ring_buffer_backend *bufb, size_t offset, - const void *src, size_t len, ssize_t pagecpy) + const void *src, size_t len, size_t pagecpy) { struct channel_backend *chanb = &bufb->chan->backend; const struct lib_ring_buffer_config *config = &chanb->config; @@ -524,7 +532,7 @@ EXPORT_SYMBOL_GPL(_lib_ring_buffer_write); */ void _lib_ring_buffer_memset(struct lib_ring_buffer_backend *bufb, size_t offset, - int c, size_t len, ssize_t pagecpy) + int c, size_t len, size_t pagecpy) { struct channel_backend *chanb = &bufb->chan->backend; const struct lib_ring_buffer_config *config = &chanb->config; @@ -557,6 +565,87 @@ void _lib_ring_buffer_memset(struct lib_ring_buffer_backend *bufb, } EXPORT_SYMBOL_GPL(_lib_ring_buffer_memset); +/** + * lib_ring_buffer_strcpy - write string data to a ring_buffer buffer. + * @bufb : buffer backend + * @offset : offset within the buffer + * @src : source address + * @len : length to write + * @pagecpy : page size copied so far + * @pad : character to use for padding + */ +void _lib_ring_buffer_strcpy(struct lib_ring_buffer_backend *bufb, + size_t offset, const char *src, size_t len, + size_t pagecpy, int pad) +{ + struct channel_backend *chanb = &bufb->chan->backend; + const struct lib_ring_buffer_config *config = &chanb->config; + size_t sbidx, index; + struct lib_ring_buffer_backend_pages *rpages; + unsigned long sb_bindex, id; + int src_terminated = 0; + + CHAN_WARN_ON(chanb, !len); + offset += pagecpy; + do { + len -= pagecpy; + if (!src_terminated) + src += pagecpy; + sbidx = offset >> chanb->subbuf_size_order; + index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT; + + /* + * Underlying layer should never ask for writes across + * subbuffers. + */ + CHAN_WARN_ON(chanb, offset >= chanb->buf_size); + + pagecpy = min_t(size_t, len, PAGE_SIZE - (offset & ~PAGE_MASK)); + id = bufb->buf_wsb[sbidx].id; + sb_bindex = subbuffer_id_get_index(config, id); + rpages = bufb->array[sb_bindex]; + CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE + && subbuffer_id_is_noref(config, id)); + + if (likely(!src_terminated)) { + size_t count, to_copy; + + to_copy = pagecpy; + if (pagecpy == len) + to_copy--; /* Final '\0' */ + count = lib_ring_buffer_do_strcpy(config, + rpages->p[index].virt + + (offset & ~PAGE_MASK), + src, to_copy); + offset += count; + /* Padding */ + if (unlikely(count < to_copy)) { + size_t pad_len = to_copy - count; + + /* Next pages will have padding */ + src_terminated = 1; + lib_ring_buffer_do_memset(rpages->p[index].virt + + (offset & ~PAGE_MASK), + pad, pad_len); + offset += pad_len; + } + } else { + size_t pad_len; + + pad_len = pagecpy; + if (pagecpy == len) + pad_len--; /* Final '\0' */ + lib_ring_buffer_do_memset(rpages->p[index].virt + + (offset & ~PAGE_MASK), + pad, pad_len); + offset += pad_len; + } + } while (unlikely(len != pagecpy)); + /* Ending '\0' */ + lib_ring_buffer_do_memset(rpages->p[index].virt + (offset & ~PAGE_MASK), + '\0', 1); +} +EXPORT_SYMBOL_GPL(_lib_ring_buffer_strcpy); /** * lib_ring_buffer_copy_from_user_inatomic - write user data to a ring_buffer buffer. @@ -573,7 +662,7 @@ EXPORT_SYMBOL_GPL(_lib_ring_buffer_memset); void _lib_ring_buffer_copy_from_user_inatomic(struct lib_ring_buffer_backend *bufb, size_t offset, const void __user *src, size_t len, - ssize_t pagecpy) + size_t pagecpy) { struct channel_backend *chanb = &bufb->chan->backend; const struct lib_ring_buffer_config *config = &chanb->config; @@ -605,8 +694,7 @@ void _lib_ring_buffer_copy_from_user_inatomic(struct lib_ring_buffer_backend *bu + (offset & ~PAGE_MASK), src, pagecpy) != 0; if (ret > 0) { - offset += (pagecpy - ret); - len -= (pagecpy - ret); + /* Copy failed. */ _lib_ring_buffer_memset(bufb, offset, 0, len, 0); break; /* stop copy */ } @@ -614,6 +702,91 @@ void _lib_ring_buffer_copy_from_user_inatomic(struct lib_ring_buffer_backend *bu } EXPORT_SYMBOL_GPL(_lib_ring_buffer_copy_from_user_inatomic); +/** + * lib_ring_buffer_strcpy_from_user_inatomic - write userspace string data to a ring_buffer buffer. + * @bufb : buffer backend + * @offset : offset within the buffer + * @src : source address + * @len : length to write + * @pagecpy : page size copied so far + * @pad : character to use for padding + * + * This function deals with userspace pointers, it should never be called + * directly without having the src pointer checked with access_ok() + * previously. + */ +void _lib_ring_buffer_strcpy_from_user_inatomic(struct lib_ring_buffer_backend *bufb, + size_t offset, const char __user *src, size_t len, + size_t pagecpy, int pad) +{ + struct channel_backend *chanb = &bufb->chan->backend; + const struct lib_ring_buffer_config *config = &chanb->config; + size_t sbidx, index; + struct lib_ring_buffer_backend_pages *rpages; + unsigned long sb_bindex, id; + int src_terminated = 0; + + offset += pagecpy; + do { + len -= pagecpy; + if (!src_terminated) + src += pagecpy; + sbidx = offset >> chanb->subbuf_size_order; + index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT; + + /* + * Underlying layer should never ask for writes across + * subbuffers. + */ + CHAN_WARN_ON(chanb, offset >= chanb->buf_size); + + pagecpy = min_t(size_t, len, PAGE_SIZE - (offset & ~PAGE_MASK)); + id = bufb->buf_wsb[sbidx].id; + sb_bindex = subbuffer_id_get_index(config, id); + rpages = bufb->array[sb_bindex]; + CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE + && subbuffer_id_is_noref(config, id)); + + if (likely(!src_terminated)) { + size_t count, to_copy; + + to_copy = pagecpy; + if (pagecpy == len) + to_copy--; /* Final '\0' */ + count = lib_ring_buffer_do_strcpy_from_user_inatomic(config, + rpages->p[index].virt + + (offset & ~PAGE_MASK), + src, to_copy); + offset += count; + /* Padding */ + if (unlikely(count < to_copy)) { + size_t pad_len = to_copy - count; + + /* Next pages will have padding */ + src_terminated = 1; + lib_ring_buffer_do_memset(rpages->p[index].virt + + (offset & ~PAGE_MASK), + pad, pad_len); + offset += pad_len; + } + } else { + size_t pad_len; + + pad_len = pagecpy; + if (pagecpy == len) + pad_len--; /* Final '\0' */ + lib_ring_buffer_do_memset(rpages->p[index].virt + + (offset & ~PAGE_MASK), + pad, pad_len); + offset += pad_len; + } + } while (unlikely(len != pagecpy)); + /* Ending '\0' */ + lib_ring_buffer_do_memset(rpages->p[index].virt + (offset & ~PAGE_MASK), + '\0', 1); +} +EXPORT_SYMBOL_GPL(_lib_ring_buffer_strcpy_from_user_inatomic); + /** * lib_ring_buffer_read - read data from ring_buffer_buffer. * @bufb : buffer backend @@ -629,8 +802,7 @@ size_t lib_ring_buffer_read(struct lib_ring_buffer_backend *bufb, size_t offset, { struct channel_backend *chanb = &bufb->chan->backend; const struct lib_ring_buffer_config *config = &chanb->config; - size_t index; - ssize_t pagecpy, orig_len; + size_t index, pagecpy, orig_len; struct lib_ring_buffer_backend_pages *rpages; unsigned long sb_bindex, id; @@ -778,15 +950,15 @@ int lib_ring_buffer_read_cstr(struct lib_ring_buffer_backend *bufb, size_t offse EXPORT_SYMBOL_GPL(lib_ring_buffer_read_cstr); /** - * lib_ring_buffer_read_get_page - Get a whole page to read from + * lib_ring_buffer_read_get_pfn - Get a page frame number to read from * @bufb : buffer backend * @offset : offset within the buffer * @virt : pointer to page address (output) * * Should be protected by get_subbuf/put_subbuf. - * Returns the pointer to the page struct pointer. + * Returns the pointer to the page frame number unsigned long. */ -struct page **lib_ring_buffer_read_get_page(struct lib_ring_buffer_backend *bufb, +unsigned long *lib_ring_buffer_read_get_pfn(struct lib_ring_buffer_backend *bufb, size_t offset, void ***virt) { size_t index; @@ -803,9 +975,9 @@ struct page **lib_ring_buffer_read_get_page(struct lib_ring_buffer_backend *bufb CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE && subbuffer_id_is_noref(config, id)); *virt = &rpages->p[index].virt; - return &rpages->p[index].page; + return &rpages->p[index].pfn; } -EXPORT_SYMBOL_GPL(lib_ring_buffer_read_get_page); +EXPORT_SYMBOL_GPL(lib_ring_buffer_read_get_pfn); /** * lib_ring_buffer_read_offset_address - get address of a buffer location @@ -814,8 +986,9 @@ EXPORT_SYMBOL_GPL(lib_ring_buffer_read_get_page); * * Return the address where a given offset is located (for read). * Should be used to get the current subbuffer header pointer. Given we know - * it's never on a page boundary, it's safe to write directly to this address, - * as long as the write is never bigger than a page size. + * it's never on a page boundary, it's safe to read/write directly + * from/to this address, as long as the read/write is never bigger than a + * page size. */ void *lib_ring_buffer_read_offset_address(struct lib_ring_buffer_backend *bufb, size_t offset)