start adding LGPL headers
[ust.git] / libust / relay.c
1 /*
2 * Public API and common code for kernel->userspace relay file support.
3 *
4 * Copyright (C) 2002-2005 - Tom Zanussi (zanussi@us.ibm.com), IBM Corp
5 * Copyright (C) 1999-2005 - Karim Yaghmour (karim@opersys.com)
6 * Copyright (C) 2008 - Mathieu Desnoyers (mathieu.desnoyers@polymtl.ca)
7 *
8 * Moved to kernel/relay.c by Paul Mundt, 2006.
9 * November 2006 - CPU hotplug support by Mathieu Desnoyers
10 * (mathieu.desnoyers@polymtl.ca)
11 *
12 * This file is released under the GPL.
13 */
14 //ust// #include <linux/errno.h>
15 //ust// #include <linux/stddef.h>
16 //ust// #include <linux/slab.h>
17 //ust// #include <linux/module.h>
18 //ust// #include <linux/string.h>
19 //ust// #include <linux/ltt-relay.h>
20 //ust// #include <linux/vmalloc.h>
21 //ust// #include <linux/mm.h>
22 //ust// #include <linux/cpu.h>
23 //ust// #include <linux/splice.h>
24 //ust// #include <linux/bitops.h>
25 #include "kernelcompat.h"
26 #include <sys/mman.h>
27 #include <sys/ipc.h>
28 #include <sys/shm.h>
29 //#include "list.h"
30 #include "relay.h"
31 #include "channels.h"
32 #include <kcompat/kref.h>
33 #include "tracer.h"
34 #include "tracercore.h"
35 #include "usterr.h"
36
37 /* list of open channels, for cpu hotplug */
38 static DEFINE_MUTEX(relay_channels_mutex);
39 static LIST_HEAD(relay_channels);
40
41
42 static struct dentry *ltt_create_buf_file_callback(struct rchan_buf *buf);
43
44 /**
45 * relay_alloc_buf - allocate a channel buffer
46 * @buf: the buffer struct
47 * @size: total size of the buffer
48 */
49 //ust// static int relay_alloc_buf(struct rchan_buf *buf, size_t *size)
50 //ust//{
51 //ust// unsigned int i, n_pages;
52 //ust// struct buf_page *buf_page, *n;
53 //ust//
54 //ust// *size = PAGE_ALIGN(*size);
55 //ust// n_pages = *size >> PAGE_SHIFT;
56 //ust//
57 //ust// INIT_LIST_HEAD(&buf->pages);
58 //ust//
59 //ust// for (i = 0; i < n_pages; i++) {
60 //ust// buf_page = kmalloc_node(sizeof(*buf_page), GFP_KERNEL,
61 //ust// cpu_to_node(buf->cpu));
62 //ust// if (unlikely(!buf_page))
63 //ust// goto depopulate;
64 //ust// buf_page->page = alloc_pages_node(cpu_to_node(buf->cpu),
65 //ust// GFP_KERNEL | __GFP_ZERO, 0);
66 //ust// if (unlikely(!buf_page->page)) {
67 //ust// kfree(buf_page);
68 //ust// goto depopulate;
69 //ust// }
70 //ust// list_add_tail(&buf_page->list, &buf->pages);
71 //ust// buf_page->offset = (size_t)i << PAGE_SHIFT;
72 //ust// buf_page->buf = buf;
73 //ust// set_page_private(buf_page->page, (unsigned long)buf_page);
74 //ust// if (i == 0) {
75 //ust// buf->wpage = buf_page;
76 //ust// buf->hpage[0] = buf_page;
77 //ust// buf->hpage[1] = buf_page;
78 //ust// buf->rpage = buf_page;
79 //ust// }
80 //ust// }
81 //ust// buf->page_count = n_pages;
82 //ust// return 0;
83 //ust//
84 //ust//depopulate:
85 //ust// list_for_each_entry_safe(buf_page, n, &buf->pages, list) {
86 //ust// list_del_init(&buf_page->list);
87 //ust// __free_page(buf_page->page);
88 //ust// kfree(buf_page);
89 //ust// }
90 //ust// return -ENOMEM;
91 //ust//}
92
93 static int relay_alloc_buf(struct rchan_buf *buf, size_t *size)
94 {
95 //ust// unsigned int n_pages;
96 //ust// struct buf_page *buf_page, *n;
97
98 void *ptr;
99 int result;
100
101 *size = PAGE_ALIGN(*size);
102
103 result = buf->shmid = shmget(getpid(), *size, IPC_CREAT | IPC_EXCL | 0700);
104 if(buf->shmid == -1) {
105 PERROR("shmget");
106 return -1;
107 }
108
109 ptr = shmat(buf->shmid, NULL, 0);
110 if(ptr == (void *) -1) {
111 perror("shmat");
112 goto destroy_shmem;
113 }
114
115 /* Already mark the shared memory for destruction. This will occur only
116 * when all users have detached.
117 */
118 result = shmctl(buf->shmid, IPC_RMID, NULL);
119 if(result == -1) {
120 perror("shmctl");
121 return -1;
122 }
123
124 buf->buf_data = ptr;
125 buf->buf_size = *size;
126
127 return 0;
128
129 destroy_shmem:
130 result = shmctl(buf->shmid, IPC_RMID, NULL);
131 if(result == -1) {
132 perror("shmctl");
133 }
134
135 return -1;
136 }
137
138 /**
139 * relay_create_buf - allocate and initialize a channel buffer
140 * @chan: the relay channel
141 * @cpu: cpu the buffer belongs to
142 *
143 * Returns channel buffer if successful, %NULL otherwise.
144 */
145 static struct rchan_buf *relay_create_buf(struct rchan *chan)
146 {
147 int ret;
148 struct rchan_buf *buf = kzalloc(sizeof(struct rchan_buf), GFP_KERNEL);
149 if (!buf)
150 return NULL;
151
152 // buf->cpu = cpu;
153 ret = relay_alloc_buf(buf, &chan->alloc_size);
154 if (ret)
155 goto free_buf;
156
157 buf->chan = chan;
158 kref_get(&buf->chan->kref);
159 return buf;
160
161 free_buf:
162 kfree(buf);
163 return NULL;
164 }
165
166 /**
167 * relay_destroy_channel - free the channel struct
168 * @kref: target kernel reference that contains the relay channel
169 *
170 * Should only be called from kref_put().
171 */
172 static void relay_destroy_channel(struct kref *kref)
173 {
174 struct rchan *chan = container_of(kref, struct rchan, kref);
175 kfree(chan);
176 }
177
178 /**
179 * relay_destroy_buf - destroy an rchan_buf struct and associated buffer
180 * @buf: the buffer struct
181 */
182 static void relay_destroy_buf(struct rchan_buf *buf)
183 {
184 struct rchan *chan = buf->chan;
185 struct buf_page *buf_page, *n;
186 int result;
187
188 result = munmap(buf->buf_data, buf->buf_size);
189 if(result == -1) {
190 PERROR("munmap");
191 }
192
193 //ust// chan->buf[buf->cpu] = NULL;
194 kfree(buf);
195 kref_put(&chan->kref, relay_destroy_channel);
196 }
197
198 /**
199 * relay_remove_buf - remove a channel buffer
200 * @kref: target kernel reference that contains the relay buffer
201 *
202 * Removes the file from the fileystem, which also frees the
203 * rchan_buf_struct and the channel buffer. Should only be called from
204 * kref_put().
205 */
206 static void relay_remove_buf(struct kref *kref)
207 {
208 struct rchan_buf *buf = container_of(kref, struct rchan_buf, kref);
209 //ust// buf->chan->cb->remove_buf_file(buf);
210 relay_destroy_buf(buf);
211 }
212
213 /*
214 * High-level relay kernel API and associated functions.
215 */
216
217 /*
218 * rchan_callback implementations defining default channel behavior. Used
219 * in place of corresponding NULL values in client callback struct.
220 */
221
222 /*
223 * create_buf_file_create() default callback. Does nothing.
224 */
225 static struct dentry *create_buf_file_default_callback(const char *filename,
226 struct dentry *parent,
227 int mode,
228 struct rchan_buf *buf)
229 {
230 return NULL;
231 }
232
233 /*
234 * remove_buf_file() default callback. Does nothing.
235 */
236 static int remove_buf_file_default_callback(struct dentry *dentry)
237 {
238 return -EINVAL;
239 }
240
241 /**
242 * wakeup_readers - wake up readers waiting on a channel
243 * @data: contains the channel buffer
244 *
245 * This is the timer function used to defer reader waking.
246 */
247 //ust// static void wakeup_readers(unsigned long data)
248 //ust// {
249 //ust// struct rchan_buf *buf = (struct rchan_buf *)data;
250 //ust// wake_up_interruptible(&buf->read_wait);
251 //ust// }
252
253 /**
254 * __relay_reset - reset a channel buffer
255 * @buf: the channel buffer
256 * @init: 1 if this is a first-time initialization
257 *
258 * See relay_reset() for description of effect.
259 */
260 static void __relay_reset(struct rchan_buf *buf, unsigned int init)
261 {
262 if (init) {
263 //ust// init_waitqueue_head(&buf->read_wait);
264 kref_init(&buf->kref);
265 //ust// setup_timer(&buf->timer, wakeup_readers, (unsigned long)buf);
266 } else
267 //ust// del_timer_sync(&buf->timer);
268
269 buf->finalized = 0;
270 }
271
272 /*
273 * relay_open_buf - create a new relay channel buffer
274 *
275 * used by relay_open() and CPU hotplug.
276 */
277 static struct rchan_buf *relay_open_buf(struct rchan *chan)
278 {
279 struct rchan_buf *buf = NULL;
280 struct dentry *dentry;
281 //ust// char *tmpname;
282
283 //ust// tmpname = kzalloc(NAME_MAX + 1, GFP_KERNEL);
284 //ust// if (!tmpname)
285 //ust// goto end;
286 //ust// snprintf(tmpname, NAME_MAX, "%s%d", chan->base_filename, cpu);
287
288 buf = relay_create_buf(chan);
289 if (!buf)
290 goto free_name;
291
292 __relay_reset(buf, 1);
293
294 /* Create file in fs */
295 //ust// dentry = chan->cb->create_buf_file(tmpname, chan->parent, S_IRUSR,
296 //ust// buf);
297
298 ltt_create_buf_file_callback(buf); // ust //
299
300 //ust// if (!dentry)
301 //ust// goto free_buf;
302 //ust//
303 //ust// buf->dentry = dentry;
304
305 goto free_name;
306
307 free_buf:
308 relay_destroy_buf(buf);
309 buf = NULL;
310 free_name:
311 //ust// kfree(tmpname);
312 end:
313 return buf;
314 }
315
316 /**
317 * relay_close_buf - close a channel buffer
318 * @buf: channel buffer
319 *
320 * Marks the buffer finalized and restores the default callbacks.
321 * The channel buffer and channel buffer data structure are then freed
322 * automatically when the last reference is given up.
323 */
324 static void relay_close_buf(struct rchan_buf *buf)
325 {
326 //ust// del_timer_sync(&buf->timer);
327 kref_put(&buf->kref, relay_remove_buf);
328 }
329
330 //ust// static void setup_callbacks(struct rchan *chan,
331 //ust// struct rchan_callbacks *cb)
332 //ust// {
333 //ust// if (!cb) {
334 //ust// chan->cb = &default_channel_callbacks;
335 //ust// return;
336 //ust// }
337 //ust//
338 //ust// if (!cb->create_buf_file)
339 //ust// cb->create_buf_file = create_buf_file_default_callback;
340 //ust// if (!cb->remove_buf_file)
341 //ust// cb->remove_buf_file = remove_buf_file_default_callback;
342 //ust// chan->cb = cb;
343 //ust// }
344
345 /**
346 * relay_hotcpu_callback - CPU hotplug callback
347 * @nb: notifier block
348 * @action: hotplug action to take
349 * @hcpu: CPU number
350 *
351 * Returns the success/failure of the operation. (%NOTIFY_OK, %NOTIFY_BAD)
352 */
353 //ust// static int __cpuinit relay_hotcpu_callback(struct notifier_block *nb,
354 //ust// unsigned long action,
355 //ust// void *hcpu)
356 //ust// {
357 //ust// unsigned int hotcpu = (unsigned long)hcpu;
358 //ust// struct rchan *chan;
359 //ust//
360 //ust// switch (action) {
361 //ust// case CPU_UP_PREPARE:
362 //ust// case CPU_UP_PREPARE_FROZEN:
363 //ust// mutex_lock(&relay_channels_mutex);
364 //ust// list_for_each_entry(chan, &relay_channels, list) {
365 //ust// if (chan->buf[hotcpu])
366 //ust// continue;
367 //ust// chan->buf[hotcpu] = relay_open_buf(chan, hotcpu);
368 //ust// if (!chan->buf[hotcpu]) {
369 //ust// printk(KERN_ERR
370 //ust// "relay_hotcpu_callback: cpu %d buffer "
371 //ust// "creation failed\n", hotcpu);
372 //ust// mutex_unlock(&relay_channels_mutex);
373 //ust// return NOTIFY_BAD;
374 //ust// }
375 //ust// }
376 //ust// mutex_unlock(&relay_channels_mutex);
377 //ust// break;
378 //ust// case CPU_DEAD:
379 //ust// case CPU_DEAD_FROZEN:
380 //ust// /* No need to flush the cpu : will be flushed upon
381 //ust// * final relay_flush() call. */
382 //ust// break;
383 //ust// }
384 //ust// return NOTIFY_OK;
385 //ust// }
386
387 /**
388 * ltt_relay_open - create a new relay channel
389 * @base_filename: base name of files to create
390 * @parent: dentry of parent directory, %NULL for root directory
391 * @subbuf_size: size of sub-buffers
392 * @n_subbufs: number of sub-buffers
393 * @cb: client callback functions
394 * @private_data: user-defined data
395 *
396 * Returns channel pointer if successful, %NULL otherwise.
397 *
398 * Creates a channel buffer for each cpu using the sizes and
399 * attributes specified. The created channel buffer files
400 * will be named base_filename0...base_filenameN-1. File
401 * permissions will be %S_IRUSR.
402 */
403 struct rchan *ltt_relay_open(const char *base_filename,
404 struct dentry *parent,
405 size_t subbuf_size,
406 size_t n_subbufs,
407 void *private_data)
408 {
409 unsigned int i;
410 struct rchan *chan;
411 //ust// if (!base_filename)
412 //ust// return NULL;
413
414 if (!(subbuf_size && n_subbufs))
415 return NULL;
416
417 chan = kzalloc(sizeof(struct rchan), GFP_KERNEL);
418 if (!chan)
419 return NULL;
420
421 chan->version = LTT_RELAY_CHANNEL_VERSION;
422 chan->n_subbufs = n_subbufs;
423 chan->subbuf_size = subbuf_size;
424 chan->subbuf_size_order = get_count_order(subbuf_size);
425 chan->alloc_size = FIX_SIZE(subbuf_size * n_subbufs);
426 chan->parent = parent;
427 chan->private_data = private_data;
428 //ust// strlcpy(chan->base_filename, base_filename, NAME_MAX);
429 //ust// setup_callbacks(chan, cb);
430 kref_init(&chan->kref);
431
432 mutex_lock(&relay_channels_mutex);
433 //ust// for_each_online_cpu(i) {
434 chan->buf = relay_open_buf(chan);
435 if (!chan->buf)
436 goto error;
437 //ust// }
438 list_add(&chan->list, &relay_channels);
439 mutex_unlock(&relay_channels_mutex);
440
441 return chan;
442
443 //ust//free_bufs:
444 //ust// for_each_possible_cpu(i) {
445 //ust// if (!chan->buf[i])
446 //ust// break;
447 //ust// relay_close_buf(chan->buf[i]);
448 //ust// }
449
450 error:
451 kref_put(&chan->kref, relay_destroy_channel);
452 mutex_unlock(&relay_channels_mutex);
453 return NULL;
454 }
455 //ust// EXPORT_SYMBOL_GPL(ltt_relay_open);
456
457 /**
458 * ltt_relay_close - close the channel
459 * @chan: the channel
460 *
461 * Closes all channel buffers and frees the channel.
462 */
463 void ltt_relay_close(struct rchan *chan)
464 {
465 unsigned int i;
466
467 if (!chan)
468 return;
469
470 mutex_lock(&relay_channels_mutex);
471 //ust// for_each_possible_cpu(i)
472 if (chan->buf)
473 relay_close_buf(chan->buf);
474
475 list_del(&chan->list);
476 kref_put(&chan->kref, relay_destroy_channel);
477 mutex_unlock(&relay_channels_mutex);
478 }
479 //ust// EXPORT_SYMBOL_GPL(ltt_relay_close);
480
481 /*
482 * Start iteration at the previous element. Skip the real list head.
483 */
484 //ust// struct buf_page *ltt_relay_find_prev_page(struct rchan_buf *buf,
485 //ust// struct buf_page *page, size_t offset, ssize_t diff_offset)
486 //ust// {
487 //ust// struct buf_page *iter;
488 //ust// size_t orig_iter_off;
489 //ust// unsigned int i = 0;
490 //ust//
491 //ust// orig_iter_off = page->offset;
492 //ust// list_for_each_entry_reverse(iter, &page->list, list) {
493 //ust// /*
494 //ust// * Skip the real list head.
495 //ust// */
496 //ust// if (&iter->list == &buf->pages)
497 //ust// continue;
498 //ust// i++;
499 //ust// if (offset >= iter->offset
500 //ust// && offset < iter->offset + PAGE_SIZE) {
501 //ust// #ifdef CONFIG_LTT_RELAY_CHECK_RANDOM_ACCESS
502 //ust// if (i > 1) {
503 //ust// printk(KERN_WARNING
504 //ust// "Backward random access detected in "
505 //ust// "ltt_relay. Iterations %u, "
506 //ust// "offset %zu, orig iter->off %zu, "
507 //ust// "iter->off %zu diff_offset %zd.\n", i,
508 //ust// offset, orig_iter_off, iter->offset,
509 //ust// diff_offset);
510 //ust// WARN_ON(1);
511 //ust// }
512 //ust// #endif
513 //ust// return iter;
514 //ust// }
515 //ust// }
516 //ust// WARN_ON(1);
517 //ust// return NULL;
518 //ust// }
519 //ust// EXPORT_SYMBOL_GPL(ltt_relay_find_prev_page);
520
521 /*
522 * Start iteration at the next element. Skip the real list head.
523 */
524 //ust// struct buf_page *ltt_relay_find_next_page(struct rchan_buf *buf,
525 //ust// struct buf_page *page, size_t offset, ssize_t diff_offset)
526 //ust// {
527 //ust// struct buf_page *iter;
528 //ust// unsigned int i = 0;
529 //ust// size_t orig_iter_off;
530 //ust//
531 //ust// orig_iter_off = page->offset;
532 //ust// list_for_each_entry(iter, &page->list, list) {
533 //ust// /*
534 //ust// * Skip the real list head.
535 //ust// */
536 //ust// if (&iter->list == &buf->pages)
537 //ust// continue;
538 //ust// i++;
539 //ust// if (offset >= iter->offset
540 //ust// && offset < iter->offset + PAGE_SIZE) {
541 //ust// #ifdef CONFIG_LTT_RELAY_CHECK_RANDOM_ACCESS
542 //ust// if (i > 1) {
543 //ust// printk(KERN_WARNING
544 //ust// "Forward random access detected in "
545 //ust// "ltt_relay. Iterations %u, "
546 //ust// "offset %zu, orig iter->off %zu, "
547 //ust// "iter->off %zu diff_offset %zd.\n", i,
548 //ust// offset, orig_iter_off, iter->offset,
549 //ust// diff_offset);
550 //ust// WARN_ON(1);
551 //ust// }
552 //ust// #endif
553 //ust// return iter;
554 //ust// }
555 //ust// }
556 //ust// WARN_ON(1);
557 //ust// return NULL;
558 //ust// }
559 //ust// EXPORT_SYMBOL_GPL(ltt_relay_find_next_page);
560
561 /**
562 * ltt_relay_write - write data to a ltt_relay buffer.
563 * @buf : buffer
564 * @offset : offset within the buffer
565 * @src : source address
566 * @len : length to write
567 * @page : cached buffer page
568 * @pagecpy : page size copied so far
569 */
570 void _ltt_relay_write(struct rchan_buf *buf, size_t offset,
571 const void *src, size_t len, ssize_t cpy)
572 {
573 do {
574 len -= cpy;
575 src += cpy;
576 offset += cpy;
577 /*
578 * Underlying layer should never ask for writes across
579 * subbuffers.
580 */
581 WARN_ON(offset >= buf->buf_size);
582
583 cpy = min_t(size_t, len, buf->buf_size - offset);
584 ltt_relay_do_copy(buf->buf_data + offset, src, cpy);
585 } while (unlikely(len != cpy));
586 }
587 //ust// EXPORT_SYMBOL_GPL(_ltt_relay_write);
588
589 /**
590 * ltt_relay_read - read data from ltt_relay_buffer.
591 * @buf : buffer
592 * @offset : offset within the buffer
593 * @dest : destination address
594 * @len : length to write
595 */
596 //ust// int ltt_relay_read(struct rchan_buf *buf, size_t offset,
597 //ust// void *dest, size_t len)
598 //ust// {
599 //ust// struct buf_page *page;
600 //ust// ssize_t pagecpy, orig_len;
601 //ust//
602 //ust// orig_len = len;
603 //ust// offset &= buf->chan->alloc_size - 1;
604 //ust// page = buf->rpage;
605 //ust// if (unlikely(!len))
606 //ust// return 0;
607 //ust// for (;;) {
608 //ust// page = ltt_relay_cache_page(buf, &buf->rpage, page, offset);
609 //ust// pagecpy = min_t(size_t, len, PAGE_SIZE - (offset & ~PAGE_MASK));
610 //ust// memcpy(dest, page_address(page->page) + (offset & ~PAGE_MASK),
611 //ust// pagecpy);
612 //ust// len -= pagecpy;
613 //ust// if (likely(!len))
614 //ust// break;
615 //ust// dest += pagecpy;
616 //ust// offset += pagecpy;
617 //ust// /*
618 //ust// * Underlying layer should never ask for reads across
619 //ust// * subbuffers.
620 //ust// */
621 //ust// WARN_ON(offset >= buf->chan->alloc_size);
622 //ust// }
623 //ust// return orig_len;
624 //ust// }
625 //ust// EXPORT_SYMBOL_GPL(ltt_relay_read);
626
627 /**
628 * ltt_relay_read_get_page - Get a whole page to read from
629 * @buf : buffer
630 * @offset : offset within the buffer
631 */
632 //ust// struct buf_page *ltt_relay_read_get_page(struct rchan_buf *buf, size_t offset)
633 //ust// {
634 //ust// struct buf_page *page;
635
636 //ust// offset &= buf->chan->alloc_size - 1;
637 //ust// page = buf->rpage;
638 //ust// page = ltt_relay_cache_page(buf, &buf->rpage, page, offset);
639 //ust// return page;
640 //ust// }
641 //ust// EXPORT_SYMBOL_GPL(ltt_relay_read_get_page);
642
643 /**
644 * ltt_relay_offset_address - get address of a location within the buffer
645 * @buf : buffer
646 * @offset : offset within the buffer.
647 *
648 * Return the address where a given offset is located.
649 * Should be used to get the current subbuffer header pointer. Given we know
650 * it's never on a page boundary, it's safe to write directly to this address,
651 * as long as the write is never bigger than a page size.
652 */
653 void *ltt_relay_offset_address(struct rchan_buf *buf, size_t offset)
654 {
655 //ust// struct buf_page *page;
656 //ust// unsigned int odd;
657 //ust//
658 //ust// offset &= buf->chan->alloc_size - 1;
659 //ust// odd = !!(offset & buf->chan->subbuf_size);
660 //ust// page = buf->hpage[odd];
661 //ust// if (offset < page->offset || offset >= page->offset + PAGE_SIZE)
662 //ust// buf->hpage[odd] = page = buf->wpage;
663 //ust// page = ltt_relay_cache_page(buf, &buf->hpage[odd], page, offset);
664 //ust// return page_address(page->page) + (offset & ~PAGE_MASK);
665 return ((char *)buf->buf_data)+offset;
666 return NULL;
667 }
668 //ust// EXPORT_SYMBOL_GPL(ltt_relay_offset_address);
669
670 /**
671 * relay_file_open - open file op for relay files
672 * @inode: the inode
673 * @filp: the file
674 *
675 * Increments the channel buffer refcount.
676 */
677 //ust// static int relay_file_open(struct inode *inode, struct file *filp)
678 //ust// {
679 //ust// struct rchan_buf *buf = inode->i_private;
680 //ust// kref_get(&buf->kref);
681 //ust// filp->private_data = buf;
682 //ust//
683 //ust// return nonseekable_open(inode, filp);
684 //ust// }
685
686 /**
687 * relay_file_release - release file op for relay files
688 * @inode: the inode
689 * @filp: the file
690 *
691 * Decrements the channel refcount, as the filesystem is
692 * no longer using it.
693 */
694 //ust// static int relay_file_release(struct inode *inode, struct file *filp)
695 //ust// {
696 //ust// struct rchan_buf *buf = filp->private_data;
697 //ust// kref_put(&buf->kref, relay_remove_buf);
698 //ust//
699 //ust// return 0;
700 //ust// }
701
702 //ust// const struct file_operations ltt_relay_file_operations = {
703 //ust// .open = relay_file_open,
704 //ust// .release = relay_file_release,
705 //ust// };
706 //ust// EXPORT_SYMBOL_GPL(ltt_relay_file_operations);
707
708 //ust// static __init int relay_init(void)
709 //ust// {
710 //ust// hotcpu_notifier(relay_hotcpu_callback, 5);
711 //ust// return 0;
712 //ust// }
713
714 //ust// module_init(relay_init);
715 /*
716 * ltt/ltt-relay.c
717 *
718 * (C) Copyright 2005-2008 - Mathieu Desnoyers (mathieu.desnoyers@polymtl.ca)
719 *
720 * LTTng lockless buffer space management (reader/writer).
721 *
722 * Author:
723 * Mathieu Desnoyers (mathieu.desnoyers@polymtl.ca)
724 *
725 * Inspired from LTT :
726 * Karim Yaghmour (karim@opersys.com)
727 * Tom Zanussi (zanussi@us.ibm.com)
728 * Bob Wisniewski (bob@watson.ibm.com)
729 * And from K42 :
730 * Bob Wisniewski (bob@watson.ibm.com)
731 *
732 * Changelog:
733 * 08/10/08, Cleanup.
734 * 19/10/05, Complete lockless mechanism.
735 * 27/05/05, Modular redesign and rewrite.
736 *
737 * Userspace reader semantic :
738 * while (poll fd != POLLHUP) {
739 * - ioctl RELAY_GET_SUBBUF_SIZE
740 * while (1) {
741 * - ioctl GET_SUBBUF
742 * - splice 1 subbuffer worth of data to a pipe
743 * - splice the data from pipe to disk/network
744 * - ioctl PUT_SUBBUF, check error value
745 * if err val < 0, previous subbuffer was corrupted.
746 * }
747 * }
748 */
749
750 //ust// #include <linux/time.h>
751 //ust// #include <linux/ltt-tracer.h>
752 //ust// #include <linux/ltt-relay.h>
753 //ust// #include <linux/module.h>
754 //ust// #include <linux/string.h>
755 //ust// #include <linux/slab.h>
756 //ust// #include <linux/init.h>
757 //ust// #include <linux/rcupdate.h>
758 //ust// #include <linux/sched.h>
759 //ust// #include <linux/bitops.h>
760 //ust// #include <linux/fs.h>
761 //ust// #include <linux/smp_lock.h>
762 //ust// #include <linux/debugfs.h>
763 //ust// #include <linux/stat.h>
764 //ust// #include <linux/cpu.h>
765 //ust// #include <linux/pipe_fs_i.h>
766 //ust// #include <linux/splice.h>
767 //ust// #include <asm/atomic.h>
768 //ust// #include <asm/local.h>
769
770 #if 0
771 #define printk_dbg(fmt, args...) printk(fmt, args)
772 #else
773 #define printk_dbg(fmt, args...)
774 #endif
775
776 /*
777 * Last TSC comparison functions. Check if the current TSC overflows
778 * LTT_TSC_BITS bits from the last TSC read. Reads and writes last_tsc
779 * atomically.
780 */
781
782 #if (BITS_PER_LONG == 32)
783 static inline void save_last_tsc(struct ltt_channel_buf_struct *ltt_buf,
784 u64 tsc)
785 {
786 ltt_buf->last_tsc = (unsigned long)(tsc >> LTT_TSC_BITS);
787 }
788
789 static inline int last_tsc_overflow(struct ltt_channel_buf_struct *ltt_buf,
790 u64 tsc)
791 {
792 unsigned long tsc_shifted = (unsigned long)(tsc >> LTT_TSC_BITS);
793
794 if (unlikely((tsc_shifted - ltt_buf->last_tsc)))
795 return 1;
796 else
797 return 0;
798 }
799 #else
800 static inline void save_last_tsc(struct ltt_channel_buf_struct *ltt_buf,
801 u64 tsc)
802 {
803 ltt_buf->last_tsc = (unsigned long)tsc;
804 }
805
806 static inline int last_tsc_overflow(struct ltt_channel_buf_struct *ltt_buf,
807 u64 tsc)
808 {
809 if (unlikely((tsc - ltt_buf->last_tsc) >> LTT_TSC_BITS))
810 return 1;
811 else
812 return 0;
813 }
814 #endif
815
816 //ust// static struct file_operations ltt_file_operations;
817
818 /*
819 * A switch is done during tracing or as a final flush after tracing (so it
820 * won't write in the new sub-buffer).
821 */
822 enum force_switch_mode { FORCE_ACTIVE, FORCE_FLUSH };
823
824 static int ltt_relay_create_buffer(struct ltt_trace_struct *trace,
825 struct ltt_channel_struct *ltt_chan,
826 struct rchan_buf *buf,
827 unsigned int n_subbufs);
828
829 static void ltt_relay_destroy_buffer(struct ltt_channel_struct *ltt_chan);
830
831 static void ltt_force_switch(struct rchan_buf *buf,
832 enum force_switch_mode mode);
833
834 /*
835 * Trace callbacks
836 */
837 static void ltt_buffer_begin_callback(struct rchan_buf *buf,
838 u64 tsc, unsigned int subbuf_idx)
839 {
840 struct ltt_channel_struct *channel =
841 (struct ltt_channel_struct *)buf->chan->private_data;
842 struct ltt_subbuffer_header *header =
843 (struct ltt_subbuffer_header *)
844 ltt_relay_offset_address(buf,
845 subbuf_idx * buf->chan->subbuf_size);
846
847 header->cycle_count_begin = tsc;
848 header->lost_size = 0xFFFFFFFF; /* for debugging */
849 header->buf_size = buf->chan->subbuf_size;
850 ltt_write_trace_header(channel->trace, header);
851 }
852
853 /*
854 * offset is assumed to never be 0 here : never deliver a completely empty
855 * subbuffer. The lost size is between 0 and subbuf_size-1.
856 */
857 static notrace void ltt_buffer_end_callback(struct rchan_buf *buf,
858 u64 tsc, unsigned int offset, unsigned int subbuf_idx)
859 {
860 struct ltt_channel_struct *channel =
861 (struct ltt_channel_struct *)buf->chan->private_data;
862 struct ltt_channel_buf_struct *ltt_buf = channel->buf;
863 struct ltt_subbuffer_header *header =
864 (struct ltt_subbuffer_header *)
865 ltt_relay_offset_address(buf,
866 subbuf_idx * buf->chan->subbuf_size);
867
868 header->lost_size = SUBBUF_OFFSET((buf->chan->subbuf_size - offset),
869 buf->chan);
870 header->cycle_count_end = tsc;
871 header->events_lost = local_read(&ltt_buf->events_lost);
872 header->subbuf_corrupt = local_read(&ltt_buf->corrupted_subbuffers);
873
874 }
875
876 void (*wake_consumer)(void *, int) = NULL;
877
878 void relay_set_wake_consumer(void (*wake)(void *, int))
879 {
880 wake_consumer = wake;
881 }
882
883 void relay_wake_consumer(void *arg, int finished)
884 {
885 if(wake_consumer)
886 wake_consumer(arg, finished);
887 }
888
889 static notrace void ltt_deliver(struct rchan_buf *buf, unsigned int subbuf_idx,
890 long commit_count)
891 {
892 struct ltt_channel_struct *channel =
893 (struct ltt_channel_struct *)buf->chan->private_data;
894 struct ltt_channel_buf_struct *ltt_buf = channel->buf;
895 int result;
896
897 //ust// #ifdef CONFIG_LTT_VMCORE
898 local_set(&ltt_buf->commit_seq[subbuf_idx], commit_count);
899 //ust// #endif
900
901 /* wakeup consumer */
902 result = write(ltt_buf->data_ready_fd_write, "1", 1);
903 if(result == -1) {
904 PERROR("write (in ltt_relay_buffer_flush)");
905 ERR("this should never happen!");
906 }
907 //ust// atomic_set(&ltt_buf->wakeup_readers, 1);
908 }
909
910 static struct dentry *ltt_create_buf_file_callback(struct rchan_buf *buf)
911 {
912 struct ltt_channel_struct *ltt_chan;
913 int err;
914 //ust// struct dentry *dentry;
915
916 ltt_chan = buf->chan->private_data;
917 err = ltt_relay_create_buffer(ltt_chan->trace, ltt_chan, buf, buf->chan->n_subbufs);
918 if (err)
919 return ERR_PTR(err);
920
921 //ust// dentry = debugfs_create_file(filename, mode, parent, buf,
922 //ust// &ltt_file_operations);
923 //ust// if (!dentry)
924 //ust// goto error;
925 //ust// return dentry;
926 return NULL; //ust//
927 //ust//error:
928 ltt_relay_destroy_buffer(ltt_chan);
929 return NULL;
930 }
931
932 static int ltt_remove_buf_file_callback(struct rchan_buf *buf)
933 {
934 //ust// struct rchan_buf *buf = dentry->d_inode->i_private;
935 struct ltt_channel_struct *ltt_chan = buf->chan->private_data;
936
937 //ust// debugfs_remove(dentry);
938 ltt_relay_destroy_buffer(ltt_chan);
939
940 return 0;
941 }
942
943 /*
944 * Wake writers :
945 *
946 * This must be done after the trace is removed from the RCU list so that there
947 * are no stalled writers.
948 */
949 //ust// static void ltt_relay_wake_writers(struct ltt_channel_buf_struct *ltt_buf)
950 //ust// {
951 //ust//
952 //ust// if (waitqueue_active(&ltt_buf->write_wait))
953 //ust// wake_up_interruptible(&ltt_buf->write_wait);
954 //ust// }
955
956 /*
957 * This function should not be called from NMI interrupt context
958 */
959 static notrace void ltt_buf_unfull(struct rchan_buf *buf,
960 unsigned int subbuf_idx,
961 long offset)
962 {
963 //ust// struct ltt_channel_struct *ltt_channel =
964 //ust// (struct ltt_channel_struct *)buf->chan->private_data;
965 //ust// struct ltt_channel_buf_struct *ltt_buf = ltt_channel->buf;
966 //ust//
967 //ust// ltt_relay_wake_writers(ltt_buf);
968 }
969
970 /**
971 * ltt_open - open file op for ltt files
972 * @inode: opened inode
973 * @file: opened file
974 *
975 * Open implementation. Makes sure only one open instance of a buffer is
976 * done at a given moment.
977 */
978 //ust// static int ltt_open(struct inode *inode, struct file *file)
979 //ust// {
980 //ust// struct rchan_buf *buf = inode->i_private;
981 //ust// struct ltt_channel_struct *ltt_channel =
982 //ust// (struct ltt_channel_struct *)buf->chan->private_data;
983 //ust// struct ltt_channel_buf_struct *ltt_buf =
984 //ust// percpu_ptr(ltt_channel->buf, buf->cpu);
985 //ust//
986 //ust// if (!atomic_long_add_unless(&ltt_buf->active_readers, 1, 1))
987 //ust// return -EBUSY;
988 //ust// return ltt_relay_file_operations.open(inode, file);
989 //ust// }
990
991 /**
992 * ltt_release - release file op for ltt files
993 * @inode: opened inode
994 * @file: opened file
995 *
996 * Release implementation.
997 */
998 //ust// static int ltt_release(struct inode *inode, struct file *file)
999 //ust// {
1000 //ust// struct rchan_buf *buf = inode->i_private;
1001 //ust// struct ltt_channel_struct *ltt_channel =
1002 //ust// (struct ltt_channel_struct *)buf->chan->private_data;
1003 //ust// struct ltt_channel_buf_struct *ltt_buf =
1004 //ust// percpu_ptr(ltt_channel->buf, buf->cpu);
1005 //ust// int ret;
1006 //ust//
1007 //ust// WARN_ON(atomic_long_read(&ltt_buf->active_readers) != 1);
1008 //ust// atomic_long_dec(&ltt_buf->active_readers);
1009 //ust// ret = ltt_relay_file_operations.release(inode, file);
1010 //ust// WARN_ON(ret);
1011 //ust// return ret;
1012 //ust// }
1013
1014 /**
1015 * ltt_poll - file op for ltt files
1016 * @filp: the file
1017 * @wait: poll table
1018 *
1019 * Poll implementation.
1020 */
1021 //ust// static unsigned int ltt_poll(struct file *filp, poll_table *wait)
1022 //ust// {
1023 //ust// unsigned int mask = 0;
1024 //ust// struct inode *inode = filp->f_dentry->d_inode;
1025 //ust// struct rchan_buf *buf = inode->i_private;
1026 //ust// struct ltt_channel_struct *ltt_channel =
1027 //ust// (struct ltt_channel_struct *)buf->chan->private_data;
1028 //ust// struct ltt_channel_buf_struct *ltt_buf =
1029 //ust// percpu_ptr(ltt_channel->buf, buf->cpu);
1030 //ust//
1031 //ust// if (filp->f_mode & FMODE_READ) {
1032 //ust// poll_wait_set_exclusive(wait);
1033 //ust// poll_wait(filp, &buf->read_wait, wait);
1034 //ust//
1035 //ust// WARN_ON(atomic_long_read(&ltt_buf->active_readers) != 1);
1036 //ust// if (SUBBUF_TRUNC(local_read(&ltt_buf->offset),
1037 //ust// buf->chan)
1038 //ust// - SUBBUF_TRUNC(atomic_long_read(&ltt_buf->consumed),
1039 //ust// buf->chan)
1040 //ust// == 0) {
1041 //ust// if (buf->finalized)
1042 //ust// return POLLHUP;
1043 //ust// else
1044 //ust// return 0;
1045 //ust// } else {
1046 //ust// struct rchan *rchan =
1047 //ust// ltt_channel->trans_channel_data;
1048 //ust// if (SUBBUF_TRUNC(local_read(&ltt_buf->offset),
1049 //ust// buf->chan)
1050 //ust// - SUBBUF_TRUNC(atomic_long_read(
1051 //ust// &ltt_buf->consumed),
1052 //ust// buf->chan)
1053 //ust// >= rchan->alloc_size)
1054 //ust// return POLLPRI | POLLRDBAND;
1055 //ust// else
1056 //ust// return POLLIN | POLLRDNORM;
1057 //ust// }
1058 //ust// }
1059 //ust// return mask;
1060 //ust// }
1061
1062 int ltt_do_get_subbuf(struct rchan_buf *buf, struct ltt_channel_buf_struct *ltt_buf, long *pconsumed_old)
1063 {
1064 struct ltt_channel_struct *ltt_channel = (struct ltt_channel_struct *)buf->chan->private_data;
1065 long consumed_old, consumed_idx, commit_count, write_offset;
1066 consumed_old = atomic_long_read(&ltt_buf->consumed);
1067 consumed_idx = SUBBUF_INDEX(consumed_old, buf->chan);
1068 commit_count = local_read(&ltt_buf->commit_count[consumed_idx]);
1069 /*
1070 * Make sure we read the commit count before reading the buffer
1071 * data and the write offset. Correct consumed offset ordering
1072 * wrt commit count is insured by the use of cmpxchg to update
1073 * the consumed offset.
1074 */
1075 smp_rmb();
1076 write_offset = local_read(&ltt_buf->offset);
1077 /*
1078 * Check that the subbuffer we are trying to consume has been
1079 * already fully committed.
1080 */
1081 if (((commit_count - buf->chan->subbuf_size)
1082 & ltt_channel->commit_count_mask)
1083 - (BUFFER_TRUNC(consumed_old, buf->chan)
1084 >> ltt_channel->n_subbufs_order)
1085 != 0) {
1086 return -EAGAIN;
1087 }
1088 /*
1089 * Check that we are not about to read the same subbuffer in
1090 * which the writer head is.
1091 */
1092 if ((SUBBUF_TRUNC(write_offset, buf->chan)
1093 - SUBBUF_TRUNC(consumed_old, buf->chan))
1094 == 0) {
1095 return -EAGAIN;
1096 }
1097
1098 *pconsumed_old = consumed_old;
1099 return 0;
1100 }
1101
1102 int ltt_do_put_subbuf(struct rchan_buf *buf, struct ltt_channel_buf_struct *ltt_buf, u32 uconsumed_old)
1103 {
1104 long consumed_new, consumed_old;
1105
1106 consumed_old = atomic_long_read(&ltt_buf->consumed);
1107 consumed_old = consumed_old & (~0xFFFFFFFFL);
1108 consumed_old = consumed_old | uconsumed_old;
1109 consumed_new = SUBBUF_ALIGN(consumed_old, buf->chan);
1110
1111 //ust// spin_lock(&ltt_buf->full_lock);
1112 if (atomic_long_cmpxchg(&ltt_buf->consumed, consumed_old,
1113 consumed_new)
1114 != consumed_old) {
1115 /* We have been pushed by the writer : the last
1116 * buffer read _is_ corrupted! It can also
1117 * happen if this is a buffer we never got. */
1118 //ust// spin_unlock(&ltt_buf->full_lock);
1119 return -EIO;
1120 } else {
1121 /* tell the client that buffer is now unfull */
1122 int index;
1123 long data;
1124 index = SUBBUF_INDEX(consumed_old, buf->chan);
1125 data = BUFFER_OFFSET(consumed_old, buf->chan);
1126 ltt_buf_unfull(buf, index, data);
1127 //ust// spin_unlock(&ltt_buf->full_lock);
1128 }
1129 return 0;
1130 }
1131
1132 /**
1133 * ltt_ioctl - control on the debugfs file
1134 *
1135 * @inode: the inode
1136 * @filp: the file
1137 * @cmd: the command
1138 * @arg: command arg
1139 *
1140 * This ioctl implements three commands necessary for a minimal
1141 * producer/consumer implementation :
1142 * RELAY_GET_SUBBUF
1143 * Get the next sub buffer that can be read. It never blocks.
1144 * RELAY_PUT_SUBBUF
1145 * Release the currently read sub-buffer. Parameter is the last
1146 * put subbuffer (returned by GET_SUBBUF).
1147 * RELAY_GET_N_BUBBUFS
1148 * returns the number of sub buffers in the per cpu channel.
1149 * RELAY_GET_SUBBUF_SIZE
1150 * returns the size of the sub buffers.
1151 */
1152 //ust// static int ltt_ioctl(struct inode *inode, struct file *filp,
1153 //ust// unsigned int cmd, unsigned long arg)
1154 //ust// {
1155 //ust// struct rchan_buf *buf = inode->i_private;
1156 //ust// struct ltt_channel_struct *ltt_channel =
1157 //ust// (struct ltt_channel_struct *)buf->chan->private_data;
1158 //ust// struct ltt_channel_buf_struct *ltt_buf =
1159 //ust// percpu_ptr(ltt_channel->buf, buf->cpu);
1160 //ust// u32 __user *argp = (u32 __user *)arg;
1161 //ust//
1162 //ust// WARN_ON(atomic_long_read(&ltt_buf->active_readers) != 1);
1163 //ust// switch (cmd) {
1164 //ust// case RELAY_GET_SUBBUF:
1165 //ust// {
1166 //ust// int ret;
1167 //ust// ret = ltt_do_get_subbuf(buf, ltt_buf, &consumed_old);
1168 //ust// if(ret < 0)
1169 //ust// return ret;
1170 //ust// return put_user((u32)consumed_old, argp);
1171 //ust// }
1172 //ust// case RELAY_PUT_SUBBUF:
1173 //ust// {
1174 //ust// int ret;
1175 //ust// u32 uconsumed_old;
1176 //ust// ret = get_user(uconsumed_old, argp);
1177 //ust// if (ret)
1178 //ust// return ret; /* will return -EFAULT */
1179 //ust// return ltt_do_put_subbuf(buf, ltt_buf, uconsumed_old);
1180 //ust// }
1181 //ust// case RELAY_GET_N_SUBBUFS:
1182 //ust// return put_user((u32)buf->chan->n_subbufs, argp);
1183 //ust// break;
1184 //ust// case RELAY_GET_SUBBUF_SIZE:
1185 //ust// return put_user((u32)buf->chan->subbuf_size, argp);
1186 //ust// break;
1187 //ust// default:
1188 //ust// return -ENOIOCTLCMD;
1189 //ust// }
1190 //ust// return 0;
1191 //ust// }
1192
1193 //ust// #ifdef CONFIG_COMPAT
1194 //ust// static long ltt_compat_ioctl(struct file *file, unsigned int cmd,
1195 //ust// unsigned long arg)
1196 //ust// {
1197 //ust// long ret = -ENOIOCTLCMD;
1198 //ust//
1199 //ust// lock_kernel();
1200 //ust// ret = ltt_ioctl(file->f_dentry->d_inode, file, cmd, arg);
1201 //ust// unlock_kernel();
1202 //ust//
1203 //ust// return ret;
1204 //ust// }
1205 //ust// #endif
1206
1207 //ust// static void ltt_relay_pipe_buf_release(struct pipe_inode_info *pipe,
1208 //ust// struct pipe_buffer *pbuf)
1209 //ust// {
1210 //ust// }
1211 //ust//
1212 //ust// static struct pipe_buf_operations ltt_relay_pipe_buf_ops = {
1213 //ust// .can_merge = 0,
1214 //ust// .map = generic_pipe_buf_map,
1215 //ust// .unmap = generic_pipe_buf_unmap,
1216 //ust// .confirm = generic_pipe_buf_confirm,
1217 //ust// .release = ltt_relay_pipe_buf_release,
1218 //ust// .steal = generic_pipe_buf_steal,
1219 //ust// .get = generic_pipe_buf_get,
1220 //ust// };
1221
1222 //ust// static void ltt_relay_page_release(struct splice_pipe_desc *spd, unsigned int i)
1223 //ust// {
1224 //ust// }
1225
1226 /*
1227 * subbuf_splice_actor - splice up to one subbuf's worth of data
1228 */
1229 //ust// static int subbuf_splice_actor(struct file *in,
1230 //ust// loff_t *ppos,
1231 //ust// struct pipe_inode_info *pipe,
1232 //ust// size_t len,
1233 //ust// unsigned int flags)
1234 //ust// {
1235 //ust// struct rchan_buf *buf = in->private_data;
1236 //ust// struct ltt_channel_struct *ltt_channel =
1237 //ust// (struct ltt_channel_struct *)buf->chan->private_data;
1238 //ust// struct ltt_channel_buf_struct *ltt_buf =
1239 //ust// percpu_ptr(ltt_channel->buf, buf->cpu);
1240 //ust// unsigned int poff, subbuf_pages, nr_pages;
1241 //ust// struct page *pages[PIPE_BUFFERS];
1242 //ust// struct partial_page partial[PIPE_BUFFERS];
1243 //ust// struct splice_pipe_desc spd = {
1244 //ust// .pages = pages,
1245 //ust// .nr_pages = 0,
1246 //ust// .partial = partial,
1247 //ust// .flags = flags,
1248 //ust// .ops = &ltt_relay_pipe_buf_ops,
1249 //ust// .spd_release = ltt_relay_page_release,
1250 //ust// };
1251 //ust// long consumed_old, consumed_idx, roffset;
1252 //ust// unsigned long bytes_avail;
1253 //ust//
1254 //ust// /*
1255 //ust// * Check that a GET_SUBBUF ioctl has been done before.
1256 //ust// */
1257 //ust// WARN_ON(atomic_long_read(&ltt_buf->active_readers) != 1);
1258 //ust// consumed_old = atomic_long_read(&ltt_buf->consumed);
1259 //ust// consumed_old += *ppos;
1260 //ust// consumed_idx = SUBBUF_INDEX(consumed_old, buf->chan);
1261 //ust//
1262 //ust// /*
1263 //ust// * Adjust read len, if longer than what is available
1264 //ust// */
1265 //ust// bytes_avail = SUBBUF_TRUNC(local_read(&ltt_buf->offset), buf->chan)
1266 //ust// - consumed_old;
1267 //ust// WARN_ON(bytes_avail > buf->chan->alloc_size);
1268 //ust// len = min_t(size_t, len, bytes_avail);
1269 //ust// subbuf_pages = bytes_avail >> PAGE_SHIFT;
1270 //ust// nr_pages = min_t(unsigned int, subbuf_pages, PIPE_BUFFERS);
1271 //ust// roffset = consumed_old & PAGE_MASK;
1272 //ust// poff = consumed_old & ~PAGE_MASK;
1273 //ust// printk_dbg(KERN_DEBUG "SPLICE actor len %zu pos %zd write_pos %ld\n",
1274 //ust// len, (ssize_t)*ppos, local_read(&ltt_buf->offset));
1275 //ust//
1276 //ust// for (; spd.nr_pages < nr_pages; spd.nr_pages++) {
1277 //ust// unsigned int this_len;
1278 //ust// struct buf_page *page;
1279 //ust//
1280 //ust// if (!len)
1281 //ust// break;
1282 //ust// printk_dbg(KERN_DEBUG "SPLICE actor loop len %zu roffset %ld\n",
1283 //ust// len, roffset);
1284 //ust//
1285 //ust// this_len = PAGE_SIZE - poff;
1286 //ust// page = ltt_relay_read_get_page(buf, roffset);
1287 //ust// spd.pages[spd.nr_pages] = page->page;
1288 //ust// spd.partial[spd.nr_pages].offset = poff;
1289 //ust// spd.partial[spd.nr_pages].len = this_len;
1290 //ust//
1291 //ust// poff = 0;
1292 //ust// roffset += PAGE_SIZE;
1293 //ust// len -= this_len;
1294 //ust// }
1295 //ust//
1296 //ust// if (!spd.nr_pages)
1297 //ust// return 0;
1298 //ust//
1299 //ust// return splice_to_pipe(pipe, &spd);
1300 //ust// }
1301
1302 //ust// static ssize_t ltt_relay_file_splice_read(struct file *in,
1303 //ust// loff_t *ppos,
1304 //ust// struct pipe_inode_info *pipe,
1305 //ust// size_t len,
1306 //ust// unsigned int flags)
1307 //ust// {
1308 //ust// ssize_t spliced;
1309 //ust// int ret;
1310 //ust//
1311 //ust// ret = 0;
1312 //ust// spliced = 0;
1313 //ust//
1314 //ust// printk_dbg(KERN_DEBUG "SPLICE read len %zu pos %zd\n",
1315 //ust// len, (ssize_t)*ppos);
1316 //ust// while (len && !spliced) {
1317 //ust// ret = subbuf_splice_actor(in, ppos, pipe, len, flags);
1318 //ust// printk_dbg(KERN_DEBUG "SPLICE read loop ret %d\n", ret);
1319 //ust// if (ret < 0)
1320 //ust// break;
1321 //ust// else if (!ret) {
1322 //ust// if (flags & SPLICE_F_NONBLOCK)
1323 //ust// ret = -EAGAIN;
1324 //ust// break;
1325 //ust// }
1326 //ust//
1327 //ust// *ppos += ret;
1328 //ust// if (ret > len)
1329 //ust// len = 0;
1330 //ust// else
1331 //ust// len -= ret;
1332 //ust// spliced += ret;
1333 //ust// }
1334 //ust//
1335 //ust// if (spliced)
1336 //ust// return spliced;
1337 //ust//
1338 //ust// return ret;
1339 //ust// }
1340
1341 static void ltt_relay_print_subbuffer_errors(
1342 struct ltt_channel_struct *ltt_chan,
1343 long cons_off)
1344 {
1345 struct rchan *rchan = ltt_chan->trans_channel_data;
1346 struct ltt_channel_buf_struct *ltt_buf = ltt_chan->buf;
1347 long cons_idx, commit_count, write_offset;
1348
1349 cons_idx = SUBBUF_INDEX(cons_off, rchan);
1350 commit_count = local_read(&ltt_buf->commit_count[cons_idx]);
1351 /*
1352 * No need to order commit_count and write_offset reads because we
1353 * execute after trace is stopped when there are no readers left.
1354 */
1355 write_offset = local_read(&ltt_buf->offset);
1356 printk(KERN_WARNING
1357 "LTT : unread channel %s offset is %ld "
1358 "and cons_off : %ld\n",
1359 ltt_chan->channel_name, write_offset, cons_off);
1360 /* Check each sub-buffer for non filled commit count */
1361 if (((commit_count - rchan->subbuf_size) & ltt_chan->commit_count_mask)
1362 - (BUFFER_TRUNC(cons_off, rchan) >> ltt_chan->n_subbufs_order)
1363 != 0)
1364 printk(KERN_ALERT
1365 "LTT : %s : subbuffer %lu has non filled "
1366 "commit count %lu.\n",
1367 ltt_chan->channel_name, cons_idx, commit_count);
1368 printk(KERN_ALERT "LTT : %s : commit count : %lu, subbuf size %zd\n",
1369 ltt_chan->channel_name, commit_count,
1370 rchan->subbuf_size);
1371 }
1372
1373 static void ltt_relay_print_errors(struct ltt_trace_struct *trace,
1374 struct ltt_channel_struct *ltt_chan)
1375 {
1376 struct rchan *rchan = ltt_chan->trans_channel_data;
1377 struct ltt_channel_buf_struct *ltt_buf = ltt_chan->buf;
1378 long cons_off;
1379
1380 for (cons_off = atomic_long_read(&ltt_buf->consumed);
1381 (SUBBUF_TRUNC(local_read(&ltt_buf->offset),
1382 rchan)
1383 - cons_off) > 0;
1384 cons_off = SUBBUF_ALIGN(cons_off, rchan))
1385 ltt_relay_print_subbuffer_errors(ltt_chan, cons_off);
1386 }
1387
1388 static void ltt_relay_print_buffer_errors(struct ltt_channel_struct *ltt_chan)
1389 {
1390 struct ltt_trace_struct *trace = ltt_chan->trace;
1391 struct ltt_channel_buf_struct *ltt_buf = ltt_chan->buf;
1392
1393 if (local_read(&ltt_buf->events_lost))
1394 printk(KERN_ALERT
1395 "LTT : %s : %ld events lost "
1396 "in %s channel.\n",
1397 ltt_chan->channel_name,
1398 local_read(&ltt_buf->events_lost),
1399 ltt_chan->channel_name);
1400 if (local_read(&ltt_buf->corrupted_subbuffers))
1401 printk(KERN_ALERT
1402 "LTT : %s : %ld corrupted subbuffers "
1403 "in %s channel.\n",
1404 ltt_chan->channel_name,
1405 local_read(&ltt_buf->corrupted_subbuffers),
1406 ltt_chan->channel_name);
1407
1408 ltt_relay_print_errors(trace, ltt_chan);
1409 }
1410
1411 static void ltt_relay_remove_dirs(struct ltt_trace_struct *trace)
1412 {
1413 //ust// debugfs_remove(trace->dentry.trace_root);
1414 }
1415
1416 static void ltt_relay_release_channel(struct kref *kref)
1417 {
1418 struct ltt_channel_struct *ltt_chan = container_of(kref,
1419 struct ltt_channel_struct, kref);
1420 free(ltt_chan->buf);
1421 }
1422
1423 /*
1424 * Create ltt buffer.
1425 */
1426 //ust// static int ltt_relay_create_buffer(struct ltt_trace_struct *trace,
1427 //ust// struct ltt_channel_struct *ltt_chan, struct rchan_buf *buf,
1428 //ust// unsigned int cpu, unsigned int n_subbufs)
1429 //ust// {
1430 //ust// struct ltt_channel_buf_struct *ltt_buf =
1431 //ust// percpu_ptr(ltt_chan->buf, cpu);
1432 //ust// unsigned int j;
1433 //ust//
1434 //ust// ltt_buf->commit_count =
1435 //ust// kzalloc_node(sizeof(ltt_buf->commit_count) * n_subbufs,
1436 //ust// GFP_KERNEL, cpu_to_node(cpu));
1437 //ust// if (!ltt_buf->commit_count)
1438 //ust// return -ENOMEM;
1439 //ust// kref_get(&trace->kref);
1440 //ust// kref_get(&trace->ltt_transport_kref);
1441 //ust// kref_get(&ltt_chan->kref);
1442 //ust// local_set(&ltt_buf->offset, ltt_subbuffer_header_size());
1443 //ust// atomic_long_set(&ltt_buf->consumed, 0);
1444 //ust// atomic_long_set(&ltt_buf->active_readers, 0);
1445 //ust// for (j = 0; j < n_subbufs; j++)
1446 //ust// local_set(&ltt_buf->commit_count[j], 0);
1447 //ust// init_waitqueue_head(&ltt_buf->write_wait);
1448 //ust// atomic_set(&ltt_buf->wakeup_readers, 0);
1449 //ust// spin_lock_init(&ltt_buf->full_lock);
1450 //ust//
1451 //ust// ltt_buffer_begin_callback(buf, trace->start_tsc, 0);
1452 //ust// /* atomic_add made on local variable on data that belongs to
1453 //ust// * various CPUs : ok because tracing not started (for this cpu). */
1454 //ust// local_add(ltt_subbuffer_header_size(), &ltt_buf->commit_count[0]);
1455 //ust//
1456 //ust// local_set(&ltt_buf->events_lost, 0);
1457 //ust// local_set(&ltt_buf->corrupted_subbuffers, 0);
1458 //ust//
1459 //ust// return 0;
1460 //ust// }
1461
1462 static int ltt_relay_create_buffer(struct ltt_trace_struct *trace,
1463 struct ltt_channel_struct *ltt_chan, struct rchan_buf *buf,
1464 unsigned int n_subbufs)
1465 {
1466 struct ltt_channel_buf_struct *ltt_buf = ltt_chan->buf;
1467 unsigned int j;
1468 int fds[2];
1469 int result;
1470
1471 ltt_buf->commit_count =
1472 zmalloc(sizeof(ltt_buf->commit_count) * n_subbufs);
1473 if (!ltt_buf->commit_count)
1474 return -ENOMEM;
1475 kref_get(&trace->kref);
1476 kref_get(&trace->ltt_transport_kref);
1477 kref_get(&ltt_chan->kref);
1478 local_set(&ltt_buf->offset, ltt_subbuffer_header_size());
1479 atomic_long_set(&ltt_buf->consumed, 0);
1480 atomic_long_set(&ltt_buf->active_readers, 0);
1481 for (j = 0; j < n_subbufs; j++)
1482 local_set(&ltt_buf->commit_count[j], 0);
1483 //ust// init_waitqueue_head(&ltt_buf->write_wait);
1484 //ust// atomic_set(&ltt_buf->wakeup_readers, 0);
1485 //ust// spin_lock_init(&ltt_buf->full_lock);
1486
1487 ltt_buffer_begin_callback(buf, trace->start_tsc, 0);
1488
1489 local_add(ltt_subbuffer_header_size(), &ltt_buf->commit_count[0]);
1490
1491 local_set(&ltt_buf->events_lost, 0);
1492 local_set(&ltt_buf->corrupted_subbuffers, 0);
1493
1494 result = pipe(fds);
1495 if(result == -1) {
1496 PERROR("pipe");
1497 return -1;
1498 }
1499 ltt_buf->data_ready_fd_read = fds[0];
1500 ltt_buf->data_ready_fd_write = fds[1];
1501
1502 //ust// ltt_buf->commit_seq = malloc(sizeof(ltt_buf->commit_seq) * n_subbufs);
1503 //ust// if(!ltt_buf->commit_seq) {
1504 //ust// return -1;
1505 //ust// }
1506
1507 /* FIXME: decrementally destroy on error */
1508
1509 return 0;
1510 }
1511
1512 static void ltt_relay_destroy_buffer(struct ltt_channel_struct *ltt_chan)
1513 {
1514 struct ltt_trace_struct *trace = ltt_chan->trace;
1515 struct ltt_channel_buf_struct *ltt_buf = ltt_chan->buf;
1516
1517 kref_put(&ltt_chan->trace->ltt_transport_kref,
1518 ltt_release_transport);
1519 ltt_relay_print_buffer_errors(ltt_chan);
1520 //ust// free(ltt_buf->commit_seq);
1521 kfree(ltt_buf->commit_count);
1522 ltt_buf->commit_count = NULL;
1523 kref_put(&ltt_chan->kref, ltt_relay_release_channel);
1524 kref_put(&trace->kref, ltt_release_trace);
1525 //ust// wake_up_interruptible(&trace->kref_wq);
1526 }
1527
1528 static void ltt_chan_alloc_ltt_buf(struct ltt_channel_struct *ltt_chan)
1529 {
1530 void *ptr;
1531 int result;
1532
1533 /* Get one page */
1534 /* FIXME: increase size if we have a seq_commit array that overflows the page */
1535 size_t size = PAGE_ALIGN(1);
1536
1537 result = ltt_chan->buf_shmid = shmget(getpid(), size, IPC_CREAT | IPC_EXCL | 0700);
1538 if(ltt_chan->buf_shmid == -1) {
1539 PERROR("shmget");
1540 return;
1541 }
1542
1543 ptr = shmat(ltt_chan->buf_shmid, NULL, 0);
1544 if(ptr == (void *) -1) {
1545 perror("shmat");
1546 goto destroy_shmem;
1547 }
1548
1549 /* Already mark the shared memory for destruction. This will occur only
1550 * when all users have detached.
1551 */
1552 result = shmctl(ltt_chan->buf_shmid, IPC_RMID, NULL);
1553 if(result == -1) {
1554 perror("shmctl");
1555 return;
1556 }
1557
1558 ltt_chan->buf = ptr;
1559
1560 return;
1561
1562 destroy_shmem:
1563 result = shmctl(ltt_chan->buf_shmid, IPC_RMID, NULL);
1564 if(result == -1) {
1565 perror("shmctl");
1566 }
1567
1568 return;
1569 }
1570
1571 /*
1572 * Create channel.
1573 */
1574 static int ltt_relay_create_channel(const char *trace_name,
1575 struct ltt_trace_struct *trace, struct dentry *dir,
1576 const char *channel_name, struct ltt_channel_struct *ltt_chan,
1577 unsigned int subbuf_size, unsigned int n_subbufs,
1578 int overwrite)
1579 {
1580 char *tmpname;
1581 unsigned int tmpname_len;
1582 int err = 0;
1583
1584 tmpname = kmalloc(PATH_MAX, GFP_KERNEL);
1585 if (!tmpname)
1586 return EPERM;
1587 if (overwrite) {
1588 strncpy(tmpname, LTT_FLIGHT_PREFIX, PATH_MAX-1);
1589 strncat(tmpname, channel_name,
1590 PATH_MAX-1-sizeof(LTT_FLIGHT_PREFIX));
1591 } else {
1592 strncpy(tmpname, channel_name, PATH_MAX-1);
1593 }
1594 strncat(tmpname, "_", PATH_MAX-1-strlen(tmpname));
1595
1596 kref_init(&ltt_chan->kref);
1597
1598 ltt_chan->trace = trace;
1599 ltt_chan->buffer_begin = ltt_buffer_begin_callback;
1600 ltt_chan->buffer_end = ltt_buffer_end_callback;
1601 ltt_chan->overwrite = overwrite;
1602 ltt_chan->n_subbufs_order = get_count_order(n_subbufs);
1603 ltt_chan->commit_count_mask = (~0UL >> ltt_chan->n_subbufs_order);
1604 //ust// ltt_chan->buf = percpu_alloc_mask(sizeof(struct ltt_channel_buf_struct), GFP_KERNEL, cpu_possible_map);
1605
1606 ltt_chan_alloc_ltt_buf(ltt_chan);
1607
1608 //ust// ltt_chan->buf = malloc(sizeof(struct ltt_channel_buf_struct));
1609 if (!ltt_chan->buf)
1610 goto alloc_error;
1611 ltt_chan->trans_channel_data = ltt_relay_open(tmpname,
1612 dir,
1613 subbuf_size,
1614 n_subbufs,
1615 ltt_chan);
1616 tmpname_len = strlen(tmpname);
1617 if (tmpname_len > 0) {
1618 /* Remove final _ for pretty printing */
1619 tmpname[tmpname_len-1] = '\0';
1620 }
1621 if (ltt_chan->trans_channel_data == NULL) {
1622 printk(KERN_ERR "LTT : Can't open %s channel for trace %s\n",
1623 tmpname, trace_name);
1624 goto relay_open_error;
1625 }
1626
1627 err = 0;
1628 goto end;
1629
1630 relay_open_error:
1631 //ust// percpu_free(ltt_chan->buf);
1632 alloc_error:
1633 err = EPERM;
1634 end:
1635 kfree(tmpname);
1636 return err;
1637 }
1638
1639 static int ltt_relay_create_dirs(struct ltt_trace_struct *new_trace)
1640 {
1641 //ust// new_trace->dentry.trace_root = debugfs_create_dir(new_trace->trace_name,
1642 //ust// get_ltt_root());
1643 //ust// if (new_trace->dentry.trace_root == NULL) {
1644 //ust// printk(KERN_ERR "LTT : Trace directory name %s already taken\n",
1645 //ust// new_trace->trace_name);
1646 //ust// return EEXIST;
1647 //ust// }
1648
1649 //ust// new_trace->callbacks.create_buf_file = ltt_create_buf_file_callback;
1650 //ust// new_trace->callbacks.remove_buf_file = ltt_remove_buf_file_callback;
1651
1652 return 0;
1653 }
1654
1655 /*
1656 * LTTng channel flush function.
1657 *
1658 * Must be called when no tracing is active in the channel, because of
1659 * accesses across CPUs.
1660 */
1661 static notrace void ltt_relay_buffer_flush(struct rchan_buf *buf)
1662 {
1663 struct ltt_channel_struct *channel =
1664 (struct ltt_channel_struct *)buf->chan->private_data;
1665 struct ltt_channel_buf_struct *ltt_buf = channel->buf;
1666 int result;
1667
1668 buf->finalized = 1;
1669 ltt_force_switch(buf, FORCE_FLUSH);
1670
1671 result = write(ltt_buf->data_ready_fd_write, "1", 1);
1672 if(result == -1) {
1673 PERROR("write (in ltt_relay_buffer_flush)");
1674 ERR("this should never happen!");
1675 }
1676 }
1677
1678 static void ltt_relay_async_wakeup_chan(struct ltt_channel_struct *ltt_channel)
1679 {
1680 //ust// unsigned int i;
1681 //ust// struct rchan *rchan = ltt_channel->trans_channel_data;
1682 //ust//
1683 //ust// for_each_possible_cpu(i) {
1684 //ust// struct ltt_channel_buf_struct *ltt_buf =
1685 //ust// percpu_ptr(ltt_channel->buf, i);
1686 //ust//
1687 //ust// if (atomic_read(&ltt_buf->wakeup_readers) == 1) {
1688 //ust// atomic_set(&ltt_buf->wakeup_readers, 0);
1689 //ust// wake_up_interruptible(&rchan->buf[i]->read_wait);
1690 //ust// }
1691 //ust// }
1692 }
1693
1694 static void ltt_relay_finish_buffer(struct ltt_channel_struct *ltt_channel)
1695 {
1696 struct rchan *rchan = ltt_channel->trans_channel_data;
1697 int result;
1698
1699 if (rchan->buf) {
1700 struct ltt_channel_buf_struct *ltt_buf = ltt_channel->buf;
1701 ltt_relay_buffer_flush(rchan->buf);
1702 //ust// ltt_relay_wake_writers(ltt_buf);
1703 /* closing the pipe tells the consumer the buffer is finished */
1704
1705 //result = write(ltt_buf->data_ready_fd_write, "D", 1);
1706 //if(result == -1) {
1707 // PERROR("write (in ltt_relay_finish_buffer)");
1708 // ERR("this should never happen!");
1709 //}
1710 close(ltt_buf->data_ready_fd_write);
1711 }
1712 }
1713
1714
1715 static void ltt_relay_finish_channel(struct ltt_channel_struct *ltt_channel)
1716 {
1717 unsigned int i;
1718
1719 //ust// for_each_possible_cpu(i)
1720 ltt_relay_finish_buffer(ltt_channel);
1721 }
1722
1723 static void ltt_relay_remove_channel(struct ltt_channel_struct *channel)
1724 {
1725 struct rchan *rchan = channel->trans_channel_data;
1726
1727 ltt_relay_close(rchan);
1728 kref_put(&channel->kref, ltt_relay_release_channel);
1729 }
1730
1731 struct ltt_reserve_switch_offsets {
1732 long begin, end, old;
1733 long begin_switch, end_switch_current, end_switch_old;
1734 long commit_count, reserve_commit_diff;
1735 size_t before_hdr_pad, size;
1736 };
1737
1738 /*
1739 * Returns :
1740 * 0 if ok
1741 * !0 if execution must be aborted.
1742 */
1743 static inline int ltt_relay_try_reserve(
1744 struct ltt_channel_struct *ltt_channel,
1745 struct ltt_channel_buf_struct *ltt_buf, struct rchan *rchan,
1746 struct rchan_buf *buf,
1747 struct ltt_reserve_switch_offsets *offsets, size_t data_size,
1748 u64 *tsc, unsigned int *rflags, int largest_align)
1749 {
1750 offsets->begin = local_read(&ltt_buf->offset);
1751 offsets->old = offsets->begin;
1752 offsets->begin_switch = 0;
1753 offsets->end_switch_current = 0;
1754 offsets->end_switch_old = 0;
1755
1756 *tsc = trace_clock_read64();
1757 if (last_tsc_overflow(ltt_buf, *tsc))
1758 *rflags = LTT_RFLAG_ID_SIZE_TSC;
1759
1760 if (SUBBUF_OFFSET(offsets->begin, buf->chan) == 0) {
1761 offsets->begin_switch = 1; /* For offsets->begin */
1762 } else {
1763 offsets->size = ltt_get_header_size(ltt_channel,
1764 offsets->begin, data_size,
1765 &offsets->before_hdr_pad, *rflags);
1766 offsets->size += ltt_align(offsets->begin + offsets->size,
1767 largest_align)
1768 + data_size;
1769 if ((SUBBUF_OFFSET(offsets->begin, buf->chan) + offsets->size)
1770 > buf->chan->subbuf_size) {
1771 offsets->end_switch_old = 1; /* For offsets->old */
1772 offsets->begin_switch = 1; /* For offsets->begin */
1773 }
1774 }
1775 if (offsets->begin_switch) {
1776 long subbuf_index;
1777
1778 if (offsets->end_switch_old)
1779 offsets->begin = SUBBUF_ALIGN(offsets->begin,
1780 buf->chan);
1781 offsets->begin = offsets->begin + ltt_subbuffer_header_size();
1782 /* Test new buffer integrity */
1783 subbuf_index = SUBBUF_INDEX(offsets->begin, buf->chan);
1784 offsets->reserve_commit_diff =
1785 (BUFFER_TRUNC(offsets->begin, buf->chan)
1786 >> ltt_channel->n_subbufs_order)
1787 - (local_read(&ltt_buf->commit_count[subbuf_index])
1788 & ltt_channel->commit_count_mask);
1789 if (offsets->reserve_commit_diff == 0) {
1790 /* Next buffer not corrupted. */
1791 if (!ltt_channel->overwrite &&
1792 (SUBBUF_TRUNC(offsets->begin, buf->chan)
1793 - SUBBUF_TRUNC(atomic_long_read(
1794 &ltt_buf->consumed),
1795 buf->chan))
1796 >= rchan->alloc_size) {
1797 /*
1798 * We do not overwrite non consumed buffers
1799 * and we are full : event is lost.
1800 */
1801 local_inc(&ltt_buf->events_lost);
1802 return -1;
1803 } else {
1804 /*
1805 * next buffer not corrupted, we are either in
1806 * overwrite mode or the buffer is not full.
1807 * It's safe to write in this new subbuffer.
1808 */
1809 }
1810 } else {
1811 /*
1812 * Next subbuffer corrupted. Force pushing reader even
1813 * in normal mode. It's safe to write in this new
1814 * subbuffer.
1815 */
1816 }
1817 offsets->size = ltt_get_header_size(ltt_channel,
1818 offsets->begin, data_size,
1819 &offsets->before_hdr_pad, *rflags);
1820 offsets->size += ltt_align(offsets->begin + offsets->size,
1821 largest_align)
1822 + data_size;
1823 if ((SUBBUF_OFFSET(offsets->begin, buf->chan) + offsets->size)
1824 > buf->chan->subbuf_size) {
1825 /*
1826 * Event too big for subbuffers, report error, don't
1827 * complete the sub-buffer switch.
1828 */
1829 local_inc(&ltt_buf->events_lost);
1830 return -1;
1831 } else {
1832 /*
1833 * We just made a successful buffer switch and the event
1834 * fits in the new subbuffer. Let's write.
1835 */
1836 }
1837 } else {
1838 /*
1839 * Event fits in the current buffer and we are not on a switch
1840 * boundary. It's safe to write.
1841 */
1842 }
1843 offsets->end = offsets->begin + offsets->size;
1844
1845 if ((SUBBUF_OFFSET(offsets->end, buf->chan)) == 0) {
1846 /*
1847 * The offset_end will fall at the very beginning of the next
1848 * subbuffer.
1849 */
1850 offsets->end_switch_current = 1; /* For offsets->begin */
1851 }
1852 return 0;
1853 }
1854
1855 /*
1856 * Returns :
1857 * 0 if ok
1858 * !0 if execution must be aborted.
1859 */
1860 static inline int ltt_relay_try_switch(
1861 enum force_switch_mode mode,
1862 struct ltt_channel_struct *ltt_channel,
1863 struct ltt_channel_buf_struct *ltt_buf, struct rchan *rchan,
1864 struct rchan_buf *buf,
1865 struct ltt_reserve_switch_offsets *offsets,
1866 u64 *tsc)
1867 {
1868 long subbuf_index;
1869
1870 offsets->begin = local_read(&ltt_buf->offset);
1871 offsets->old = offsets->begin;
1872 offsets->begin_switch = 0;
1873 offsets->end_switch_old = 0;
1874
1875 *tsc = trace_clock_read64();
1876
1877 if (SUBBUF_OFFSET(offsets->begin, buf->chan) != 0) {
1878 offsets->begin = SUBBUF_ALIGN(offsets->begin, buf->chan);
1879 offsets->end_switch_old = 1;
1880 } else {
1881 /* we do not have to switch : buffer is empty */
1882 return -1;
1883 }
1884 if (mode == FORCE_ACTIVE)
1885 offsets->begin += ltt_subbuffer_header_size();
1886 /*
1887 * Always begin_switch in FORCE_ACTIVE mode.
1888 * Test new buffer integrity
1889 */
1890 subbuf_index = SUBBUF_INDEX(offsets->begin, buf->chan);
1891 offsets->reserve_commit_diff =
1892 (BUFFER_TRUNC(offsets->begin, buf->chan)
1893 >> ltt_channel->n_subbufs_order)
1894 - (local_read(&ltt_buf->commit_count[subbuf_index])
1895 & ltt_channel->commit_count_mask);
1896 if (offsets->reserve_commit_diff == 0) {
1897 /* Next buffer not corrupted. */
1898 if (mode == FORCE_ACTIVE
1899 && !ltt_channel->overwrite
1900 && offsets->begin - atomic_long_read(&ltt_buf->consumed)
1901 >= rchan->alloc_size) {
1902 /*
1903 * We do not overwrite non consumed buffers and we are
1904 * full : ignore switch while tracing is active.
1905 */
1906 return -1;
1907 }
1908 } else {
1909 /*
1910 * Next subbuffer corrupted. Force pushing reader even in normal
1911 * mode
1912 */
1913 }
1914 offsets->end = offsets->begin;
1915 return 0;
1916 }
1917
1918 static inline void ltt_reserve_push_reader(
1919 struct ltt_channel_struct *ltt_channel,
1920 struct ltt_channel_buf_struct *ltt_buf,
1921 struct rchan *rchan,
1922 struct rchan_buf *buf,
1923 struct ltt_reserve_switch_offsets *offsets)
1924 {
1925 long consumed_old, consumed_new;
1926
1927 do {
1928 consumed_old = atomic_long_read(&ltt_buf->consumed);
1929 /*
1930 * If buffer is in overwrite mode, push the reader consumed
1931 * count if the write position has reached it and we are not
1932 * at the first iteration (don't push the reader farther than
1933 * the writer). This operation can be done concurrently by many
1934 * writers in the same buffer, the writer being at the farthest
1935 * write position sub-buffer index in the buffer being the one
1936 * which will win this loop.
1937 * If the buffer is not in overwrite mode, pushing the reader
1938 * only happens if a sub-buffer is corrupted.
1939 */
1940 if ((SUBBUF_TRUNC(offsets->end-1, buf->chan)
1941 - SUBBUF_TRUNC(consumed_old, buf->chan))
1942 >= rchan->alloc_size)
1943 consumed_new = SUBBUF_ALIGN(consumed_old, buf->chan);
1944 else {
1945 consumed_new = consumed_old;
1946 break;
1947 }
1948 } while (atomic_long_cmpxchg(&ltt_buf->consumed, consumed_old,
1949 consumed_new) != consumed_old);
1950
1951 if (consumed_old != consumed_new) {
1952 /*
1953 * Reader pushed : we are the winner of the push, we can
1954 * therefore reequilibrate reserve and commit. Atomic increment
1955 * of the commit count permits other writers to play around
1956 * with this variable before us. We keep track of
1957 * corrupted_subbuffers even in overwrite mode :
1958 * we never want to write over a non completely committed
1959 * sub-buffer : possible causes : the buffer size is too low
1960 * compared to the unordered data input, or there is a writer
1961 * that died between the reserve and the commit.
1962 */
1963 if (offsets->reserve_commit_diff) {
1964 /*
1965 * We have to alter the sub-buffer commit count.
1966 * We do not deliver the previous subbuffer, given it
1967 * was either corrupted or not consumed (overwrite
1968 * mode).
1969 */
1970 local_add(offsets->reserve_commit_diff,
1971 &ltt_buf->commit_count[
1972 SUBBUF_INDEX(offsets->begin,
1973 buf->chan)]);
1974 if (!ltt_channel->overwrite
1975 || offsets->reserve_commit_diff
1976 != rchan->subbuf_size) {
1977 /*
1978 * The reserve commit diff was not subbuf_size :
1979 * it means the subbuffer was partly written to
1980 * and is therefore corrupted. If it is multiple
1981 * of subbuffer size and we are in flight
1982 * recorder mode, we are skipping over a whole
1983 * subbuffer.
1984 */
1985 local_inc(&ltt_buf->corrupted_subbuffers);
1986 }
1987 }
1988 }
1989 }
1990
1991
1992 /*
1993 * ltt_reserve_switch_old_subbuf: switch old subbuffer
1994 *
1995 * Concurrency safe because we are the last and only thread to alter this
1996 * sub-buffer. As long as it is not delivered and read, no other thread can
1997 * alter the offset, alter the reserve_count or call the
1998 * client_buffer_end_callback on this sub-buffer.
1999 *
2000 * The only remaining threads could be the ones with pending commits. They will
2001 * have to do the deliver themselves. Not concurrency safe in overwrite mode.
2002 * We detect corrupted subbuffers with commit and reserve counts. We keep a
2003 * corrupted sub-buffers count and push the readers across these sub-buffers.
2004 *
2005 * Not concurrency safe if a writer is stalled in a subbuffer and another writer
2006 * switches in, finding out it's corrupted. The result will be than the old
2007 * (uncommited) subbuffer will be declared corrupted, and that the new subbuffer
2008 * will be declared corrupted too because of the commit count adjustment.
2009 *
2010 * Note : offset_old should never be 0 here.
2011 */
2012 static inline void ltt_reserve_switch_old_subbuf(
2013 struct ltt_channel_struct *ltt_channel,
2014 struct ltt_channel_buf_struct *ltt_buf, struct rchan *rchan,
2015 struct rchan_buf *buf,
2016 struct ltt_reserve_switch_offsets *offsets, u64 *tsc)
2017 {
2018 long oldidx = SUBBUF_INDEX(offsets->old - 1, rchan);
2019
2020 ltt_channel->buffer_end(buf, *tsc, offsets->old, oldidx);
2021 /* Must write buffer end before incrementing commit count */
2022 smp_wmb();
2023 offsets->commit_count =
2024 local_add_return(rchan->subbuf_size
2025 - (SUBBUF_OFFSET(offsets->old - 1, rchan)
2026 + 1),
2027 &ltt_buf->commit_count[oldidx]);
2028 if ((BUFFER_TRUNC(offsets->old - 1, rchan)
2029 >> ltt_channel->n_subbufs_order)
2030 - ((offsets->commit_count - rchan->subbuf_size)
2031 & ltt_channel->commit_count_mask) == 0)
2032 ltt_deliver(buf, oldidx, offsets->commit_count);
2033 }
2034
2035 /*
2036 * ltt_reserve_switch_new_subbuf: Populate new subbuffer.
2037 *
2038 * This code can be executed unordered : writers may already have written to the
2039 * sub-buffer before this code gets executed, caution. The commit makes sure
2040 * that this code is executed before the deliver of this sub-buffer.
2041 */
2042 static /*inline*/ void ltt_reserve_switch_new_subbuf(
2043 struct ltt_channel_struct *ltt_channel,
2044 struct ltt_channel_buf_struct *ltt_buf, struct rchan *rchan,
2045 struct rchan_buf *buf,
2046 struct ltt_reserve_switch_offsets *offsets, u64 *tsc)
2047 {
2048 long beginidx = SUBBUF_INDEX(offsets->begin, rchan);
2049
2050 ltt_channel->buffer_begin(buf, *tsc, beginidx);
2051 /* Must write buffer end before incrementing commit count */
2052 smp_wmb();
2053 offsets->commit_count = local_add_return(ltt_subbuffer_header_size(),
2054 &ltt_buf->commit_count[beginidx]);
2055 /* Check if the written buffer has to be delivered */
2056 if ((BUFFER_TRUNC(offsets->begin, rchan)
2057 >> ltt_channel->n_subbufs_order)
2058 - ((offsets->commit_count - rchan->subbuf_size)
2059 & ltt_channel->commit_count_mask) == 0)
2060 ltt_deliver(buf, beginidx, offsets->commit_count);
2061 }
2062
2063
2064 /*
2065 * ltt_reserve_end_switch_current: finish switching current subbuffer
2066 *
2067 * Concurrency safe because we are the last and only thread to alter this
2068 * sub-buffer. As long as it is not delivered and read, no other thread can
2069 * alter the offset, alter the reserve_count or call the
2070 * client_buffer_end_callback on this sub-buffer.
2071 *
2072 * The only remaining threads could be the ones with pending commits. They will
2073 * have to do the deliver themselves. Not concurrency safe in overwrite mode.
2074 * We detect corrupted subbuffers with commit and reserve counts. We keep a
2075 * corrupted sub-buffers count and push the readers across these sub-buffers.
2076 *
2077 * Not concurrency safe if a writer is stalled in a subbuffer and another writer
2078 * switches in, finding out it's corrupted. The result will be than the old
2079 * (uncommited) subbuffer will be declared corrupted, and that the new subbuffer
2080 * will be declared corrupted too because of the commit count adjustment.
2081 */
2082 static inline void ltt_reserve_end_switch_current(
2083 struct ltt_channel_struct *ltt_channel,
2084 struct ltt_channel_buf_struct *ltt_buf, struct rchan *rchan,
2085 struct rchan_buf *buf,
2086 struct ltt_reserve_switch_offsets *offsets, u64 *tsc)
2087 {
2088 long endidx = SUBBUF_INDEX(offsets->end - 1, rchan);
2089
2090 ltt_channel->buffer_end(buf, *tsc, offsets->end, endidx);
2091 /* Must write buffer begin before incrementing commit count */
2092 smp_wmb();
2093 offsets->commit_count =
2094 local_add_return(rchan->subbuf_size
2095 - (SUBBUF_OFFSET(offsets->end - 1, rchan)
2096 + 1),
2097 &ltt_buf->commit_count[endidx]);
2098 if ((BUFFER_TRUNC(offsets->end - 1, rchan)
2099 >> ltt_channel->n_subbufs_order)
2100 - ((offsets->commit_count - rchan->subbuf_size)
2101 & ltt_channel->commit_count_mask) == 0)
2102 ltt_deliver(buf, endidx, offsets->commit_count);
2103 }
2104
2105 /**
2106 * ltt_relay_reserve_slot - Atomic slot reservation in a LTTng buffer.
2107 * @trace: the trace structure to log to.
2108 * @ltt_channel: channel structure
2109 * @transport_data: data structure specific to ltt relay
2110 * @data_size: size of the variable length data to log.
2111 * @slot_size: pointer to total size of the slot (out)
2112 * @buf_offset : pointer to reserved buffer offset (out)
2113 * @tsc: pointer to the tsc at the slot reservation (out)
2114 * @cpu: cpuid
2115 *
2116 * Return : -ENOSPC if not enough space, else returns 0.
2117 * It will take care of sub-buffer switching.
2118 */
2119 static notrace int ltt_relay_reserve_slot(struct ltt_trace_struct *trace,
2120 struct ltt_channel_struct *ltt_channel, void **transport_data,
2121 size_t data_size, size_t *slot_size, long *buf_offset, u64 *tsc,
2122 unsigned int *rflags, int largest_align)
2123 {
2124 struct rchan *rchan = ltt_channel->trans_channel_data;
2125 struct rchan_buf *buf = *transport_data = rchan->buf;
2126 struct ltt_channel_buf_struct *ltt_buf = ltt_channel->buf;
2127 struct ltt_reserve_switch_offsets offsets;
2128
2129 offsets.reserve_commit_diff = 0;
2130 offsets.size = 0;
2131
2132 /*
2133 * Perform retryable operations.
2134 */
2135 if (ltt_nesting > 4) {
2136 local_inc(&ltt_buf->events_lost);
2137 return -EPERM;
2138 }
2139 do {
2140 if (ltt_relay_try_reserve(ltt_channel, ltt_buf,
2141 rchan, buf, &offsets, data_size, tsc, rflags,
2142 largest_align))
2143 return -ENOSPC;
2144 } while (local_cmpxchg(&ltt_buf->offset, offsets.old,
2145 offsets.end) != offsets.old);
2146
2147 /*
2148 * Atomically update last_tsc. This update races against concurrent
2149 * atomic updates, but the race will always cause supplementary full TSC
2150 * events, never the opposite (missing a full TSC event when it would be
2151 * needed).
2152 */
2153 save_last_tsc(ltt_buf, *tsc);
2154
2155 /*
2156 * Push the reader if necessary
2157 */
2158 ltt_reserve_push_reader(ltt_channel, ltt_buf, rchan, buf, &offsets);
2159
2160 /*
2161 * Switch old subbuffer if needed.
2162 */
2163 if (offsets.end_switch_old)
2164 ltt_reserve_switch_old_subbuf(ltt_channel, ltt_buf, rchan, buf,
2165 &offsets, tsc);
2166
2167 /*
2168 * Populate new subbuffer.
2169 */
2170 if (offsets.begin_switch)
2171 ltt_reserve_switch_new_subbuf(ltt_channel, ltt_buf, rchan,
2172 buf, &offsets, tsc);
2173
2174 if (offsets.end_switch_current)
2175 ltt_reserve_end_switch_current(ltt_channel, ltt_buf, rchan,
2176 buf, &offsets, tsc);
2177
2178 *slot_size = offsets.size;
2179 *buf_offset = offsets.begin + offsets.before_hdr_pad;
2180 return 0;
2181 }
2182
2183 /*
2184 * Force a sub-buffer switch for a per-cpu buffer. This operation is
2185 * completely reentrant : can be called while tracing is active with
2186 * absolutely no lock held.
2187 *
2188 * Note, however, that as a local_cmpxchg is used for some atomic
2189 * operations, this function must be called from the CPU which owns the buffer
2190 * for a ACTIVE flush.
2191 */
2192 static notrace void ltt_force_switch(struct rchan_buf *buf,
2193 enum force_switch_mode mode)
2194 {
2195 struct ltt_channel_struct *ltt_channel =
2196 (struct ltt_channel_struct *)buf->chan->private_data;
2197 struct ltt_channel_buf_struct *ltt_buf = ltt_channel->buf;
2198 struct rchan *rchan = ltt_channel->trans_channel_data;
2199 struct ltt_reserve_switch_offsets offsets;
2200 u64 tsc;
2201
2202 offsets.reserve_commit_diff = 0;
2203 offsets.size = 0;
2204
2205 /*
2206 * Perform retryable operations.
2207 */
2208 do {
2209 if (ltt_relay_try_switch(mode, ltt_channel, ltt_buf,
2210 rchan, buf, &offsets, &tsc))
2211 return;
2212 } while (local_cmpxchg(&ltt_buf->offset, offsets.old,
2213 offsets.end) != offsets.old);
2214
2215 /*
2216 * Atomically update last_tsc. This update races against concurrent
2217 * atomic updates, but the race will always cause supplementary full TSC
2218 * events, never the opposite (missing a full TSC event when it would be
2219 * needed).
2220 */
2221 save_last_tsc(ltt_buf, tsc);
2222
2223 /*
2224 * Push the reader if necessary
2225 */
2226 if (mode == FORCE_ACTIVE)
2227 ltt_reserve_push_reader(ltt_channel, ltt_buf, rchan,
2228 buf, &offsets);
2229
2230 /*
2231 * Switch old subbuffer if needed.
2232 */
2233 if (offsets.end_switch_old)
2234 ltt_reserve_switch_old_subbuf(ltt_channel, ltt_buf, rchan, buf,
2235 &offsets, &tsc);
2236
2237 /*
2238 * Populate new subbuffer.
2239 */
2240 if (mode == FORCE_ACTIVE)
2241 ltt_reserve_switch_new_subbuf(ltt_channel,
2242 ltt_buf, rchan, buf, &offsets, &tsc);
2243 }
2244
2245 /*
2246 * This is called with preemption disabled when user space has requested
2247 * blocking mode. If one of the active traces has free space below a
2248 * specific threshold value, we reenable preemption and block.
2249 */
2250 static int ltt_relay_user_blocking(struct ltt_trace_struct *trace,
2251 unsigned int chan_index, size_t data_size,
2252 struct user_dbg_data *dbg)
2253 {
2254 //ust// struct rchan *rchan;
2255 //ust// struct ltt_channel_buf_struct *ltt_buf;
2256 //ust// struct ltt_channel_struct *channel;
2257 //ust// struct rchan_buf *relay_buf;
2258 //ust// int cpu;
2259 //ust// DECLARE_WAITQUEUE(wait, current);
2260 //ust//
2261 //ust// channel = &trace->channels[chan_index];
2262 //ust// rchan = channel->trans_channel_data;
2263 //ust// cpu = smp_processor_id();
2264 //ust// relay_buf = rchan->buf[cpu];
2265 //ust// ltt_buf = percpu_ptr(channel->buf, cpu);
2266 //ust//
2267 //ust// /*
2268 //ust// * Check if data is too big for the channel : do not
2269 //ust// * block for it.
2270 //ust// */
2271 //ust// if (LTT_RESERVE_CRITICAL + data_size > relay_buf->chan->subbuf_size)
2272 //ust// return 0;
2273 //ust//
2274 //ust// /*
2275 //ust// * If free space too low, we block. We restart from the
2276 //ust// * beginning after we resume (cpu id may have changed
2277 //ust// * while preemption is active).
2278 //ust// */
2279 //ust// spin_lock(&ltt_buf->full_lock);
2280 //ust// if (!channel->overwrite) {
2281 //ust// dbg->write = local_read(&ltt_buf->offset);
2282 //ust// dbg->read = atomic_long_read(&ltt_buf->consumed);
2283 //ust// dbg->avail_size = dbg->write + LTT_RESERVE_CRITICAL + data_size
2284 //ust// - SUBBUF_TRUNC(dbg->read,
2285 //ust// relay_buf->chan);
2286 //ust// if (dbg->avail_size > rchan->alloc_size) {
2287 //ust// __set_current_state(TASK_INTERRUPTIBLE);
2288 //ust// add_wait_queue(&ltt_buf->write_wait, &wait);
2289 //ust// spin_unlock(&ltt_buf->full_lock);
2290 //ust// preempt_enable();
2291 //ust// schedule();
2292 //ust// __set_current_state(TASK_RUNNING);
2293 //ust// remove_wait_queue(&ltt_buf->write_wait, &wait);
2294 //ust// if (signal_pending(current))
2295 //ust// return -ERESTARTSYS;
2296 //ust// preempt_disable();
2297 //ust// return 1;
2298 //ust// }
2299 //ust// }
2300 //ust// spin_unlock(&ltt_buf->full_lock);
2301 return 0;
2302 }
2303
2304 static void ltt_relay_print_user_errors(struct ltt_trace_struct *trace,
2305 unsigned int chan_index, size_t data_size,
2306 struct user_dbg_data *dbg)
2307 {
2308 struct rchan *rchan;
2309 struct ltt_channel_buf_struct *ltt_buf;
2310 struct ltt_channel_struct *channel;
2311 struct rchan_buf *relay_buf;
2312
2313 channel = &trace->channels[chan_index];
2314 rchan = channel->trans_channel_data;
2315 relay_buf = rchan->buf;
2316 ltt_buf = channel->buf;
2317
2318 printk(KERN_ERR "Error in LTT usertrace : "
2319 "buffer full : event lost in blocking "
2320 "mode. Increase LTT_RESERVE_CRITICAL.\n");
2321 printk(KERN_ERR "LTT nesting level is %u.\n", ltt_nesting);
2322 printk(KERN_ERR "LTT avail size %lu.\n",
2323 dbg->avail_size);
2324 printk(KERN_ERR "avai write : %lu, read : %lu\n",
2325 dbg->write, dbg->read);
2326
2327 dbg->write = local_read(&ltt_buf->offset);
2328 dbg->read = atomic_long_read(&ltt_buf->consumed);
2329
2330 printk(KERN_ERR "LTT cur size %lu.\n",
2331 dbg->write + LTT_RESERVE_CRITICAL + data_size
2332 - SUBBUF_TRUNC(dbg->read, relay_buf->chan));
2333 printk(KERN_ERR "cur write : %lu, read : %lu\n",
2334 dbg->write, dbg->read);
2335 }
2336
2337 //ust// static struct ltt_transport ltt_relay_transport = {
2338 //ust// .name = "relay",
2339 //ust// .owner = THIS_MODULE,
2340 //ust// .ops = {
2341 //ust// .create_dirs = ltt_relay_create_dirs,
2342 //ust// .remove_dirs = ltt_relay_remove_dirs,
2343 //ust// .create_channel = ltt_relay_create_channel,
2344 //ust// .finish_channel = ltt_relay_finish_channel,
2345 //ust// .remove_channel = ltt_relay_remove_channel,
2346 //ust// .wakeup_channel = ltt_relay_async_wakeup_chan,
2347 //ust// .commit_slot = ltt_relay_commit_slot,
2348 //ust// .reserve_slot = ltt_relay_reserve_slot,
2349 //ust// .user_blocking = ltt_relay_user_blocking,
2350 //ust// .user_errors = ltt_relay_print_user_errors,
2351 //ust// },
2352 //ust// };
2353
2354 static struct ltt_transport ust_relay_transport = {
2355 .name = "ustrelay",
2356 //ust// .owner = THIS_MODULE,
2357 .ops = {
2358 .create_dirs = ltt_relay_create_dirs,
2359 .remove_dirs = ltt_relay_remove_dirs,
2360 .create_channel = ltt_relay_create_channel,
2361 .finish_channel = ltt_relay_finish_channel,
2362 .remove_channel = ltt_relay_remove_channel,
2363 .wakeup_channel = ltt_relay_async_wakeup_chan,
2364 // .commit_slot = ltt_relay_commit_slot,
2365 .reserve_slot = ltt_relay_reserve_slot,
2366 .user_blocking = ltt_relay_user_blocking,
2367 .user_errors = ltt_relay_print_user_errors,
2368 },
2369 };
2370
2371 //ust// static int __init ltt_relay_init(void)
2372 //ust// {
2373 //ust// printk(KERN_INFO "LTT : ltt-relay init\n");
2374 //ust//
2375 //ust// ltt_file_operations = ltt_relay_file_operations;
2376 //ust// ltt_file_operations.owner = THIS_MODULE;
2377 //ust// ltt_file_operations.open = ltt_open;
2378 //ust// ltt_file_operations.release = ltt_release;
2379 //ust// ltt_file_operations.poll = ltt_poll;
2380 //ust// ltt_file_operations.splice_read = ltt_relay_file_splice_read,
2381 //ust// ltt_file_operations.ioctl = ltt_ioctl;
2382 //ust//#ifdef CONFIG_COMPAT
2383 //ust// ltt_file_operations.compat_ioctl = ltt_compat_ioctl;
2384 //ust//#endif
2385 //ust//
2386 //ust// ltt_transport_register(&ltt_relay_transport);
2387 //ust//
2388 //ust// return 0;
2389 //ust// }
2390
2391 /*
2392 * for flight recording. must be called after relay_commit.
2393 * This function decrements de subbuffer's lost_size each time the commit count
2394 * reaches back the reserve offset (module subbuffer size). It is useful for
2395 * crash dump.
2396 */
2397 //ust// #ifdef CONFIG_LTT_VMCORE
2398 static /* inline */ void ltt_write_commit_counter(struct rchan_buf *buf,
2399 struct ltt_channel_buf_struct *ltt_buf,
2400 long idx, long buf_offset, long commit_count, size_t data_size)
2401 {
2402 long offset;
2403 long commit_seq_old;
2404
2405 offset = buf_offset + data_size;
2406
2407 /*
2408 * SUBBUF_OFFSET includes commit_count_mask. We can simply
2409 * compare the offsets within the subbuffer without caring about
2410 * buffer full/empty mismatch because offset is never zero here
2411 * (subbuffer header and event headers have non-zero length).
2412 */
2413 if (unlikely(SUBBUF_OFFSET(offset - commit_count, buf->chan)))
2414 return;
2415
2416 commit_seq_old = local_read(&ltt_buf->commit_seq[idx]);
2417 while (commit_seq_old < commit_count)
2418 commit_seq_old = local_cmpxchg(&ltt_buf->commit_seq[idx],
2419 commit_seq_old, commit_count);
2420 }
2421 //ust// #else
2422 //ust// static inline void ltt_write_commit_counter(struct rchan_buf *buf,
2423 //ust// long buf_offset, size_t slot_size)
2424 //ust// {
2425 //ust// }
2426 //ust// #endif
2427
2428 /*
2429 * Atomic unordered slot commit. Increments the commit count in the
2430 * specified sub-buffer, and delivers it if necessary.
2431 *
2432 * Parameters:
2433 *
2434 * @ltt_channel : channel structure
2435 * @transport_data: transport-specific data
2436 * @buf_offset : offset following the event header.
2437 * @data_size : size of the event data.
2438 * @slot_size : size of the reserved slot.
2439 */
2440 /* FIXME: make this function static inline in the .h! */
2441 /*static*/ /* inline */ notrace void ltt_commit_slot(
2442 struct ltt_channel_struct *ltt_channel,
2443 void **transport_data, long buf_offset,
2444 size_t data_size, size_t slot_size)
2445 {
2446 struct rchan_buf *buf = *transport_data;
2447 struct ltt_channel_buf_struct *ltt_buf = ltt_channel->buf;
2448 struct rchan *rchan = buf->chan;
2449 long offset_end = buf_offset;
2450 long endidx = SUBBUF_INDEX(offset_end - 1, rchan);
2451 long commit_count;
2452
2453 /* Must write slot data before incrementing commit count */
2454 smp_wmb();
2455 commit_count = local_add_return(slot_size,
2456 &ltt_buf->commit_count[endidx]);
2457 /* Check if all commits have been done */
2458 if ((BUFFER_TRUNC(offset_end - 1, rchan)
2459 >> ltt_channel->n_subbufs_order)
2460 - ((commit_count - rchan->subbuf_size)
2461 & ltt_channel->commit_count_mask) == 0)
2462 ltt_deliver(buf, endidx, commit_count);
2463 /*
2464 * Update lost_size for each commit. It's needed only for extracting
2465 * ltt buffers from vmcore, after crash.
2466 */
2467 ltt_write_commit_counter(buf, ltt_buf, endidx,
2468 buf_offset, commit_count, data_size);
2469
2470 DBG("commited slot. now commit count is %ld", commit_count);
2471 }
2472
2473
2474 static char initialized = 0;
2475
2476 void __attribute__((constructor)) init_ustrelay_transport(void)
2477 {
2478 if(!initialized) {
2479 ltt_transport_register(&ust_relay_transport);
2480 initialized = 1;
2481 }
2482 }
2483
2484 static void __exit ltt_relay_exit(void)
2485 {
2486 //ust// printk(KERN_INFO "LTT : ltt-relay exit\n");
2487
2488 ltt_transport_unregister(&ust_relay_transport);
2489 }
2490
2491 //ust// module_init(ltt_relay_init);
2492 //ust// module_exit(ltt_relay_exit);
2493 //ust//
2494 //ust// MODULE_LICENSE("GPL");
2495 //ust// MODULE_AUTHOR("Mathieu Desnoyers");
2496 //ust// MODULE_DESCRIPTION("Linux Trace Toolkit Next Generation Lockless Relay");