| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #ifndef NETDATA_STREAM_CIRCULAR_BUFFER_H |
| 4 | #define NETDATA_STREAM_CIRCULAR_BUFFER_H |
| 5 | |
| 6 | #include "libnetdata/libnetdata.h" |
| 7 | #include "stream-traffic-types.h" |
| 8 | |
| 9 | #ifdef __cplusplus |
| 10 | extern "C" { |
| 11 | #endif |
| 12 | |
| 13 | #define CBUFFER_INITIAL_SIZE (16ULL * 1024) |
| 14 | #define CBUFFER_INITIAL_MAX_SIZE (10ULL * 1024 * 1024) |
| 15 | #define HOST_THREAD_BUFFER_INITIAL_SIZE (256ULL * 1024) |
| 16 | #define REPLICATION_THREAD_BUFFER_INITIAL_SIZE (512ULL * 1024) |
| 17 | |
| 18 | #define STREAM_CIRCULAR_BUFFER_ADAPT_TO_TIMES_MAX_SIZE 3 |
| 19 | |
| 20 | typedef struct stream_circular_buffer_stats { |
| 21 | size_t adds; |
| 22 | size_t sends; |
| 23 | size_t recreates; |
| 24 | |
| 25 | size_t bytes_added; |
| 26 | size_t bytes_uncompressed; |
| 27 | size_t bytes_sent; |
| 28 | |
| 29 | uint32_t bytes_size; |
| 30 | uint32_t bytes_max_size; |
| 31 | uint32_t bytes_outstanding; |
| 32 | uint32_t bytes_available; |
| 33 | |
| 34 | double buffer_ratio; |
| 35 | |
| 36 | size_t bytes_sent_by_type[STREAM_TRAFFIC_TYPE_MAX]; |
| 37 | } STREAM_CIRCULAR_BUFFER_STATS; |
| 38 | |
| 39 | struct stream_circular_buffer; |
| 40 | typedef struct stream_circular_buffer STREAM_CIRCULAR_BUFFER; |
| 41 | |
| 42 | // -------------------------------------------------------------------------------------------------------------------- |
| 43 | // management |
| 44 | |
| 45 | STREAM_CIRCULAR_BUFFER *stream_circular_buffer_create(void); |
| 46 | void stream_circular_buffer_destroy(STREAM_CIRCULAR_BUFFER *scb); |
| 47 | |
| 48 | // flushes all data in the buffer |
| 49 | void stream_circular_buffer_flush_unsafe(STREAM_CIRCULAR_BUFFER *scb, size_t buffer_max_size); |
| 50 | |
| 51 | // recreates the buffer, but it does so every 5 minutes and only if the buffer has no data in it |
| 52 | // it does not alter the since_ut time of the buffer, so this is assumed to be the same session |
| 53 | // use this after deleting data from the buffer, to minimize the memory footprint of the buffer |
| 54 | void stream_circular_buffer_recreate_timed_unsafe(STREAM_CIRCULAR_BUFFER *scb, usec_t now_ut, bool force); |
| 55 | |
| 56 | // returns true if it increased the buffer size |
| 57 | // if it changes the size, it updates the statistics |
| 58 | bool stream_circular_buffer_set_max_size_unsafe(STREAM_CIRCULAR_BUFFER *scb, size_t max_size, bool force); |
| 59 | |
| 60 | // returns a pointer to the current circular buffer statistics |
| 61 | // copy it if you plan to use it without a lock |
| 62 | STREAM_CIRCULAR_BUFFER_STATS *stream_circular_buffer_stats_unsafe(STREAM_CIRCULAR_BUFFER *scb); |
| 63 | |
| 64 | // -------------------------------------------------------------------------------------------------------------------- |
| 65 | // atomic operations - no lock needed |
| 66 | |
| 67 | // returns the max size of the buffer in bytes |
| 68 | size_t stream_circular_buffer_get_max_size(STREAM_CIRCULAR_BUFFER *scb); |
| 69 | |
| 70 | // returns the current buffer used ratio |
| 71 | size_t stream_sender_get_buffer_used_percent(STREAM_CIRCULAR_BUFFER *scb); |
| 72 | |
| 73 | // return the monotonic timestamp of the last time the buffer was created |
| 74 | usec_t stream_circular_buffer_last_flush_ut(STREAM_CIRCULAR_BUFFER *scb); |
| 75 | |
| 76 | // return the monotonic timestamp of the last time we removed data from the buffer |
| 77 | usec_t stream_circular_buffer_last_sent_ut(STREAM_CIRCULAR_BUFFER *scb); |
| 78 | |
| 79 | // -------------------------------------------------------------------------------------------------------------------- |
| 80 | // data operations (add, get, remove data from/to the buffer) |
| 81 | |
| 82 | // adds data to the end of the circular buffer, returns false when it can't (buffer is full) |
| 83 | // it updates the statistics |
| 84 | bool stream_circular_buffer_add_unsafe( |
| 85 | STREAM_CIRCULAR_BUFFER *scb, const char *data, size_t bytes_actual, size_t bytes_uncompressed, |
| 86 | STREAM_TRAFFIC_TYPE type, bool autoscale); |
| 87 | |
| 88 | // returns a pointer to the beginning of the buffer, and its size in bytes |
| 89 | size_t stream_circular_buffer_get_unsafe(STREAM_CIRCULAR_BUFFER *scb, char **chunk); |
| 90 | |
| 91 | // removes data from the beginning of circular buffer |
| 92 | // it updates the statistics |
| 93 | void stream_circular_buffer_del_unsafe(STREAM_CIRCULAR_BUFFER *scb, size_t bytes, usec_t now_ut); |
| 94 | |
| 95 | #ifdef __cplusplus |
| 96 | } |
| 97 | #endif |
| 98 | |
| 99 | #endif //NETDATA_STREAM_CIRCULAR_BUFFER_H |