X-Git-Url: http://git.lttng.org/?a=blobdiff_plain;f=libringbuffer%2Ffrontend_internal.h;h=43a2d0a2639b29d319ddb7b9d1d417855f9c96bc;hb=b07cd987c6856becc4ce1c20e5118b006f4431b0;hp=00b9508d4df54768767fbe0aaff401cb0f965b0d;hpb=77d7fa989f79f3c2cb14d26576cbbb54b0637594;p=lttng-ust.git diff --git a/libringbuffer/frontend_internal.h b/libringbuffer/frontend_internal.h index 00b9508d..43a2d0a2 100644 --- a/libringbuffer/frontend_internal.h +++ b/libringbuffer/frontend_internal.h @@ -165,6 +165,15 @@ void lib_ring_buffer_switch_slow(struct lttng_ust_lib_ring_buffer *buf, enum switch_mode mode, struct lttng_ust_shm_handle *handle); +void lib_ring_buffer_check_deliver_slow(const struct lttng_ust_lib_ring_buffer_config *config, + struct lttng_ust_lib_ring_buffer *buf, + struct channel *chan, + unsigned long offset, + unsigned long commit_count, + unsigned long idx, + struct lttng_ust_shm_handle *handle, + uint64_t tsc); + /* Buffer write helpers */ static inline @@ -195,60 +204,6 @@ void lib_ring_buffer_reserve_push_reader(struct lttng_ust_lib_ring_buffer *buf, consumed_new) != consumed_old)); } -static inline -void lib_ring_buffer_vmcore_check_deliver(const struct lttng_ust_lib_ring_buffer_config *config, - struct lttng_ust_lib_ring_buffer *buf, - unsigned long commit_count, - unsigned long idx, - struct lttng_ust_shm_handle *handle) -{ - if (config->oops == RING_BUFFER_OOPS_CONSISTENCY) - v_set(config, &shmp_index(handle, buf->commit_hot, idx)->seq, commit_count); -} - -static inline -int lib_ring_buffer_poll_deliver(const struct lttng_ust_lib_ring_buffer_config *config, - struct lttng_ust_lib_ring_buffer *buf, - struct channel *chan, - struct lttng_ust_shm_handle *handle) -{ - unsigned long consumed_old, consumed_idx, commit_count, write_offset; - - consumed_old = uatomic_read(&buf->consumed); - consumed_idx = subbuf_index(consumed_old, chan); - commit_count = v_read(config, &shmp_index(handle, buf->commit_cold, consumed_idx)->cc_sb); - /* - * No memory barrier here, since we are only interested - * in a statistically correct polling result. The next poll will - * get the data is we are racing. The mb() that ensures correct - * memory order is in get_subbuf. - */ - write_offset = v_read(config, &buf->offset); - - /* - * Check that the subbuffer we are trying to consume has been - * already fully committed. - */ - - if (((commit_count - chan->backend.subbuf_size) - & chan->commit_count_mask) - - (buf_trunc(consumed_old, chan) - >> chan->backend.num_subbuf_order) - != 0) - return 0; - - /* - * Check that we are not about to read the same subbuffer in - * which the writer head is. - */ - if (subbuf_trunc(write_offset, chan) - subbuf_trunc(consumed_old, chan) - == 0) - return 0; - - return 1; - -} - static inline int lib_ring_buffer_pending_data(const struct lttng_ust_lib_ring_buffer_config *config, struct lttng_ust_lib_ring_buffer *buf, @@ -300,71 +255,6 @@ int lib_ring_buffer_reserve_committed(const struct lttng_ust_lib_ring_buffer_con - (commit_count & chan->commit_count_mask) == 0); } -static inline -void lib_ring_buffer_wakeup(struct lttng_ust_lib_ring_buffer *buf, - struct lttng_ust_shm_handle *handle) -{ - int wakeup_fd = shm_get_wakeup_fd(handle, &buf->self._ref); - sigset_t sigpipe_set, pending_set, old_set; - int ret, sigpipe_was_pending = 0; - - if (wakeup_fd < 0) - return; - - /* - * Wake-up the other end by writing a null byte in the pipe - * (non-blocking). Important note: Because writing into the - * pipe is non-blocking (and therefore we allow dropping wakeup - * data, as long as there is wakeup data present in the pipe - * buffer to wake up the consumer), the consumer should perform - * the following sequence for waiting: - * 1) empty the pipe (reads). - * 2) check if there is data in the buffer. - * 3) wait on the pipe (poll). - * - * Discard the SIGPIPE from write(), not disturbing any SIGPIPE - * that might be already pending. If a bogus SIGPIPE is sent to - * the entire process concurrently by a malicious user, it may - * be simply discarded. - */ - ret = sigemptyset(&pending_set); - assert(!ret); - /* - * sigpending returns the mask of signals that are _both_ - * blocked for the thread _and_ pending for either the thread or - * the entire process. - */ - ret = sigpending(&pending_set); - assert(!ret); - sigpipe_was_pending = sigismember(&pending_set, SIGPIPE); - /* - * If sigpipe was pending, it means it was already blocked, so - * no need to block it. - */ - if (!sigpipe_was_pending) { - ret = sigemptyset(&sigpipe_set); - assert(!ret); - ret = sigaddset(&sigpipe_set, SIGPIPE); - assert(!ret); - ret = pthread_sigmask(SIG_BLOCK, &sigpipe_set, &old_set); - assert(!ret); - } - do { - ret = write(wakeup_fd, "", 1); - } while (ret == -1L && errno == EINTR); - if (ret == -1L && errno == EPIPE && !sigpipe_was_pending) { - struct timespec timeout = { 0, 0 }; - do { - ret = sigtimedwait(&sigpipe_set, NULL, - &timeout); - } while (ret == -1L && errno == EINTR); - } - if (!sigpipe_was_pending) { - ret = pthread_sigmask(SIG_SETMASK, &old_set, NULL); - assert(!ret); - } -} - /* * Receive end of subbuffer TSC as parameter. It has been read in the * space reservation loop of either reserve or switch, which ensures it @@ -387,107 +277,9 @@ void lib_ring_buffer_check_deliver(const struct lttng_ust_lib_ring_buffer_config /* Check if all commits have been done */ if (caa_unlikely((buf_trunc(offset, chan) >> chan->backend.num_subbuf_order) - - (old_commit_count & chan->commit_count_mask) == 0)) { - /* - * If we succeeded at updating cc_sb below, we are the subbuffer - * writer delivering the subbuffer. Deals with concurrent - * updates of the "cc" value without adding a add_return atomic - * operation to the fast path. - * - * We are doing the delivery in two steps: - * - First, we cmpxchg() cc_sb to the new value - * old_commit_count + 1. This ensures that we are the only - * subbuffer user successfully filling the subbuffer, but we - * do _not_ set the cc_sb value to "commit_count" yet. - * Therefore, other writers that would wrap around the ring - * buffer and try to start writing to our subbuffer would - * have to drop records, because it would appear as - * non-filled. - * We therefore have exclusive access to the subbuffer control - * structures. This mutual exclusion with other writers is - * crucially important to perform record overruns count in - * flight recorder mode locklessly. - * - When we are ready to release the subbuffer (either for - * reading or for overrun by other writers), we simply set the - * cc_sb value to "commit_count" and perform delivery. - * - * The subbuffer size is least 2 bytes (minimum size: 1 page). - * This guarantees that old_commit_count + 1 != commit_count. - */ - - /* - * Order prior updates to reserve count prior to the - * commit_cold cc_sb update. - */ - cmm_smp_wmb(); - if (caa_likely(v_cmpxchg(config, &shmp_index(handle, buf->commit_cold, idx)->cc_sb, - old_commit_count, old_commit_count + 1) - == old_commit_count)) { - /* - * Start of exclusive subbuffer access. We are - * guaranteed to be the last writer in this subbuffer - * and any other writer trying to access this subbuffer - * in this state is required to drop records. - */ - v_add(config, - subbuffer_get_records_count(config, - &buf->backend, - idx, handle), - &buf->records_count); - v_add(config, - subbuffer_count_records_overrun(config, - &buf->backend, - idx, handle), - &buf->records_overrun); - config->cb.buffer_end(buf, tsc, idx, - lib_ring_buffer_get_data_size(config, - buf, - idx, - handle), - handle); - - /* - * Increment the packet counter while we have exclusive - * access. - */ - subbuffer_inc_packet_count(config, &buf->backend, idx, handle); - - /* - * Set noref flag and offset for this subbuffer id. - * Contains a memory barrier that ensures counter stores - * are ordered before set noref and offset. - */ - lib_ring_buffer_set_noref_offset(config, &buf->backend, idx, - buf_trunc_val(offset, chan), handle); - - /* - * Order set_noref and record counter updates before the - * end of subbuffer exclusive access. Orders with - * respect to writers coming into the subbuffer after - * wrap around, and also order wrt concurrent readers. - */ - cmm_smp_mb(); - /* End of exclusive subbuffer access */ - v_set(config, &shmp_index(handle, buf->commit_cold, idx)->cc_sb, - commit_count); - /* - * Order later updates to reserve count after - * the commit cold cc_sb update. - */ - cmm_smp_wmb(); - lib_ring_buffer_vmcore_check_deliver(config, buf, - commit_count, idx, handle); - - /* - * RING_BUFFER_WAKEUP_BY_WRITER wakeup is not lock-free. - */ - if (config->wakeup == RING_BUFFER_WAKEUP_BY_WRITER - && uatomic_read(&buf->active_readers) - && lib_ring_buffer_poll_deliver(config, buf, chan, handle)) { - lib_ring_buffer_wakeup(buf, handle); - } - } - } + - (old_commit_count & chan->commit_count_mask) == 0)) + lib_ring_buffer_check_deliver_slow(config, buf, chan, offset, + commit_count, idx, handle, tsc); } /*