X-Git-Url: https://git.lttng.org/?p=lttngtop.git;a=blobdiff_plain;f=src%2Flttngtop.c;h=15d44e02e08eec5660d607c6b593146646e9e4e4;hp=f15eea5518f4119c896114da3e4936f8caf860e7;hb=b520ab458ea26829d00fca98f4bbb90309fd30df;hpb=16b22a0fe150f4923b9902e802bbf28bacce8d0e diff --git a/src/lttngtop.c b/src/lttngtop.c index f15eea5..15d44e0 100644 --- a/src/lttngtop.c +++ b/src/lttngtop.c @@ -40,6 +40,7 @@ #include #include #include +#include #include "lttngtoptypes.h" #include "cputop.h" @@ -54,16 +55,25 @@ const char *opt_input_path; struct lttngtop *copy; pthread_t display_thread; pthread_t timer_thread; +pthread_t live_trace_thread; unsigned long refresh_display = 1 * NSEC_PER_SEC; unsigned long last_display_update = 0; int quit = 0; /* LIVE */ +pthread_t thread_live_consume; /* list of FDs available for being read with snapshots */ struct mmap_stream_list mmap_list; GPtrArray *lttng_consumer_stream_array; int sessiond_metadata, consumerd_metadata; +struct lttng_consumer_local_data *ctx = NULL; +/* list of snapshots currently not consumed */ +GPtrArray *available_snapshots; +sem_t metadata_available; +FILE *metadata_fp; +int trace_opened = 0; +int metadata_ready = 0; enum { OPT_NONE = 0, @@ -122,6 +132,29 @@ void *ncurses_display(void *p) } } +/* + * hook on each event to check the timestamp and refresh the display if + * necessary + */ +enum bt_cb_ret print_timestamp(struct bt_ctf_event *call_data, void *private_data) +{ + unsigned long timestamp; + struct tm start; + uint64_t ts_nsec_start; + + + timestamp = bt_ctf_get_timestamp(call_data); + + start = format_timestamp(timestamp); + ts_nsec_start = timestamp % NSEC_PER_SEC; + + printf("%02d:%02d:%02d.%09" PRIu64 " %s\n", start.tm_hour, + start.tm_min, start.tm_sec, ts_nsec_start, + bt_ctf_event_name(call_data)); + + return BT_CB_OK; +} + /* * hook on each event to check the timestamp and refresh the display if * necessary @@ -380,6 +413,11 @@ void iter_trace(struct bt_context *bt_ctx) begin_pos.type = BT_SEEK_BEGIN; iter = bt_ctf_iter_create(bt_ctx, &begin_pos, NULL); + bt_ctf_iter_add_callback(iter, 0, NULL, 0, + print_timestamp, + NULL, NULL, NULL); + +#if 0 /* at each event check if we need to refresh */ bt_ctf_iter_add_callback(iter, 0, NULL, 0, check_timestamp, @@ -424,7 +462,7 @@ void iter_trace(struct bt_context *bt_ctx) "lttng_statedump_file_descriptor"), NULL, 0, handle_statedump_file_descriptor, NULL, NULL, NULL); - +#endif while ((event = bt_ctf_iter_read_event(iter)) != NULL) { ret = bt_iter_next(bt_ctf_get_iter(iter)); if (ret < 0) @@ -644,7 +682,6 @@ ssize_t read_subbuffer(struct lttng_consumer_stream *kconsumerd_fd, if (helper_get_lttng_consumer_stream_output(kconsumerd_fd) == LTTNG_EVENT_SPLICE) { /* Get the next subbuffer */ - printf("get_next : %d\n", infd); err = helper_kernctl_get_next_subbuf(infd); if (err != 0) { ret = errno; @@ -659,7 +696,6 @@ ssize_t read_subbuffer(struct lttng_consumer_stream *kconsumerd_fd, perror("Getting sub-buffer len failed."); goto end; } - printf("len : %ld\n", len); /* splice the subbuffer to the tracefile */ ret = helper_lttng_consumer_on_read_subbuffer_splice(ctx, kconsumerd_fd, len); @@ -670,8 +706,6 @@ ssize_t read_subbuffer(struct lttng_consumer_stream *kconsumerd_fd, */ fprintf(stderr,"Error splicing to tracefile\n"); } - printf("ret : %ld\n", ret); - printf("put_next : %d\n", infd); err = helper_kernctl_put_next_subbuf(infd); if (err != 0) { ret = errno; @@ -679,7 +713,7 @@ ssize_t read_subbuffer(struct lttng_consumer_stream *kconsumerd_fd, "it is due to concurrency)"); goto end; } -// sem_post(&metadata_available); + sem_post(&metadata_available); } end: @@ -697,10 +731,8 @@ int on_update_fd(int key, uint32_t state) int on_recv_fd(struct lttng_consumer_stream *kconsumerd_fd) { int ret; - struct mmap_stream *new_info; - size_t tmp_mmap_len; + struct mmap_stream *new_mmap_stream; - printf("Receiving %d\n", helper_get_lttng_consumer_stream_wait_fd(kconsumerd_fd)); /* Opening the tracefile in write mode */ if (helper_get_lttng_consumer_stream_path_name(kconsumerd_fd) != NULL) { ret = open(helper_get_lttng_consumer_stream_path_name(kconsumerd_fd), @@ -713,31 +745,13 @@ int on_recv_fd(struct lttng_consumer_stream *kconsumerd_fd) } if (helper_get_lttng_consumer_stream_output(kconsumerd_fd) == LTTNG_EVENT_MMAP) { - new_info = malloc(sizeof(struct mmap_stream)); - new_info->fd = helper_get_lttng_consumer_stream_wait_fd(kconsumerd_fd); - bt_list_add(&new_info->list, &mmap_list.head); - - /* get the len of the mmap region */ - ret = helper_kernctl_get_mmap_len(helper_get_lttng_consumer_stream_wait_fd(kconsumerd_fd), - &tmp_mmap_len); - if (ret != 0) { - ret = errno; - perror("helper_kernctl_get_mmap_len"); - goto end; - } - helper_set_lttng_consumer_stream_mmap_len(kconsumerd_fd, tmp_mmap_len); - printf("mmap len : %ld\n", tmp_mmap_len); - - helper_set_lttng_consumer_stream_mmap_base(kconsumerd_fd, - mmap(NULL, helper_get_lttng_consumer_stream_mmap_len(kconsumerd_fd), - PROT_READ, MAP_PRIVATE, helper_get_lttng_consumer_stream_wait_fd(kconsumerd_fd), 0)); - if (helper_get_lttng_consumer_stream_mmap_base(kconsumerd_fd) == MAP_FAILED) { - perror("Error mmaping"); - ret = -1; - goto end; - } + new_mmap_stream = malloc(sizeof(struct mmap_stream)); + new_mmap_stream->fd = helper_get_lttng_consumer_stream_wait_fd( + kconsumerd_fd); + bt_list_add(&new_mmap_stream->list, &mmap_list.head); g_ptr_array_add(lttng_consumer_stream_array, kconsumerd_fd); + /* keep mmap FDs internally */ ret = 1; } else { consumerd_metadata = helper_get_lttng_consumer_stream_wait_fd(kconsumerd_fd); @@ -749,6 +763,35 @@ end: return ret; } +void *live_consume() +{ + struct bt_context *bt_ctx = NULL; + int ret; + + if (!metadata_ready) { + sem_wait(&metadata_available); + if (access("/tmp/livesession/kernel/metadata", F_OK) != 0) { + fprintf(stderr,"no metadata\n"); + return NULL; + } + metadata_ready = 1; + metadata_fp = fopen("/tmp/livesession/kernel/metadata", "r"); + } + + if (!trace_opened) { + bt_ctx = bt_context_create(); + ret = bt_context_add_trace(bt_ctx, NULL, "ctf", + lttngtop_ctf_packet_seek, &mmap_list, metadata_fp); + if (ret < 0) { + printf("Error adding trace\n"); + return NULL; + } + trace_opened = 1; + } + iter_trace(bt_ctx); + + return NULL; +} int setup_consumer(char *command_sock_path, pthread_t *threads, struct lttng_consumer_local_data *ctx) @@ -768,14 +811,14 @@ int setup_consumer(char *command_sock_path, pthread_t *threads, ret = pthread_create(&threads[0], NULL, helper_lttng_consumer_thread_receive_fds, (void *) ctx); if (ret != 0) { - perror("pthread_create"); + perror("pthread_create receive fd"); goto end; } /* Create thread to manage the polling/writing of traces */ ret = pthread_create(&threads[1], NULL, helper_lttng_consumer_thread_poll_fds, (void *) ctx); if (ret != 0) { - perror("pthread_create"); + perror("pthread_create poll fd"); goto end; } @@ -783,19 +826,16 @@ end: return ret; } -int setup_live_tracing() +void *setup_live_tracing() { struct lttng_domain dom; struct lttng_channel chan; char *channel_name = "mmapchan"; - struct lttng_consumer_local_data *ctx = NULL; struct lttng_event ev; int ret = 0; char *command_sock_path = "/tmp/consumerd_sock"; static pthread_t threads[2]; /* recv_fd, poll */ struct lttng_event_context kctxpid, kctxcomm, kctxppid, kctxtid; - /* list of snapshots currently not consumed */ - GPtrArray *available_snapshots; struct lttng_handle *handle; @@ -815,6 +855,7 @@ int setup_live_tracing() ret = system("rm -rf /tmp/livesession"); + lttng_destroy_session("test"); if ((ret = lttng_create_session("test", "/tmp/livesession")) < 0) { fprintf(stderr,"error creating the session : %s\n", helper_lttcomm_get_readable_code(ret)); @@ -833,9 +874,9 @@ int setup_live_tracing() } strcpy(chan.name, channel_name); - chan.attr.overwrite = 1; -// chan.attr.subbuf_size = 32768; - chan.attr.subbuf_size = 1048576; /* 1MB */ + chan.attr.overwrite = 0; + chan.attr.subbuf_size = 32768; +// chan.attr.subbuf_size = 1048576; /* 1MB */ chan.attr.num_subbuf = 4; chan.attr.switch_timer_interval = 0; chan.attr.read_timer_interval = 200; @@ -870,18 +911,21 @@ int setup_live_tracing() } helper_kernctl_buffer_flush(consumerd_metadata); - sleep(10); - lttng_stop_tracing("test"); - lttng_destroy_session("test"); + /* Create thread to manage the polling/writing of traces */ + ret = pthread_create(&thread_live_consume, NULL, live_consume, NULL); + if (ret != 0) { + perror("pthread_create"); + goto end; + } /* block until metadata is ready */ - //sem_init(&metadata_available, 0, 0); + sem_init(&metadata_available, 0, 0); //init_lttngtop(); end: - return ret; + return NULL; } int main(int argc, char **argv) @@ -899,8 +943,15 @@ int main(int argc, char **argv) } if (!opt_input_path) { - printf("live tracing enabled\n"); - setup_live_tracing(); + pthread_create(&live_trace_thread, NULL, setup_live_tracing, (void *) NULL); + sleep(2000); + printf("STOPPING\n"); + lttng_stop_tracing("test"); + printf("DESTROYING\n"); + lttng_destroy_session("test"); + + printf("CANCELLING\n"); + pthread_cancel(live_trace_thread); goto end; } else { init_lttngtop();