From 04899510de2d643ac7809e887ea0b05db2bdd522 Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Mon, 5 Jun 2023 14:35:17 -0400 Subject: [PATCH] =?UTF-8?q?Fixes=20#1107=20-=20Use=20a=20tiered=20list=20o?= =?UTF-8?q?f=20read-grant=20amounts=20based=20on=20percen=E2=80=A6=20(#111?= =?UTF-8?q?0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Fixes #1107 - Use a tiered list of read-grant amounts based on percentage of buffer capacity that's in use. * Fixes #1107 - No logic changes - clarified intent of the grant logic. * Fixes #1107 - Fixed bug identified by kguisti. Infer the max-read-buffers from Proton. * Fixes #1107 - Renamed the env variable and added a log statement. * Fixes #1107 - Suppress the race warning (TSAN) while accessing memory stats. --- include/qpid/dispatch/alloc_pool.h | 3 +- src/adaptors/adaptor_common.c | 112 +++++++++++++++++++++++++++-- src/adaptors/adaptor_common.h | 5 ++ src/adaptors/tcp/tcp_adaptor.c | 1 + tests/tsan.supp | 3 + 5 files changed, 117 insertions(+), 7 deletions(-) diff --git a/include/qpid/dispatch/alloc_pool.h b/include/qpid/dispatch/alloc_pool.h index 699102e97..050dbeb49 100644 --- a/include/qpid/dispatch/alloc_pool.h +++ b/include/qpid/dispatch/alloc_pool.h @@ -111,7 +111,8 @@ static inline void *qd_alloc_deref_safe_ptr(const qd_alloc_safe_ptr_t *sp) void free_##T(T *p); \ typedef qd_alloc_safe_ptr_t T##_sp; \ void set_safe_ptr_##T(T *p, T##_sp *sp); \ - T *safe_deref_##T(T##_sp sp) + T *safe_deref_##T(T##_sp sp); \ + qd_alloc_stats_t *alloc_stats_##T(void) /** * Define allocator configuration. diff --git a/src/adaptors/adaptor_common.c b/src/adaptors/adaptor_common.c index f8c0aecb4..5c38cc3c8 100644 --- a/src/adaptors/adaptor_common.c +++ b/src/adaptors/adaptor_common.c @@ -17,10 +17,12 @@ * under the License. */ #include "adaptor_common.h" +#include "adaptor_buffer.h" #include "qpid/dispatch/amqp.h" #include "qpid/dispatch/connection_manager.h" #include "qpid/dispatch/ctools.h" +#include "qpid/dispatch/log.h" #include @@ -29,6 +31,11 @@ ALLOC_DEFINE(qd_adaptor_config_t); +static uint64_t buffer_ceiling = 0; +static uint64_t buffer_threshold_50; +static uint64_t buffer_threshold_75; +static uint64_t buffer_threshold_85; + void qd_free_adaptor_config(qd_adaptor_config_t *config) { if (!config) @@ -72,16 +79,109 @@ qd_error_t qd_load_adaptor_config(qd_adaptor_config_t *config, qd_entity_t *enti return qd_error_code(); } +void qd_adaptor_common_init(void) +{ + if (buffer_ceiling > 0) { + return; + } + + char *ceiling_string = getenv("SKUPPER_ROUTER_MEMORY_CEILING"); + uint64_t memory_ceiling = (uint64_t) 4 * (uint64_t) 1024 * (uint64_t) 1024 * (uint64_t) 1024; // 4 Gig default + + if (!!ceiling_string) { + long long convert = atoll(ceiling_string); + if (convert > 0) { + memory_ceiling = (uint64_t) convert; + } + } + + buffer_ceiling = MAX(memory_ceiling / QD_ADAPTOR_MAX_BUFFER_SIZE, 100); + buffer_threshold_50 = buffer_ceiling / 2; + buffer_threshold_75 = (buffer_ceiling / 20) * 15; + buffer_threshold_85 = (buffer_ceiling / 20) * 17; + + qd_log(LOG_ROUTER, QD_LOG_INFO, "Adaptor buffer memory ceiling: %"PRIu64" (%"PRIu64" buffers)", memory_ceiling, buffer_ceiling); +} + int qd_raw_connection_grant_read_buffers(pn_raw_connection_t *pn_raw_conn) { + // + // Define the allocation tiers. The tier values are the number of read buffers to be granted + // to raw connections based on the percentage of usage of the router-wide buffer ceiling. + // +#define TIER_1 8 // [0% .. 50%) +#define TIER_2 4 // [50% .. 75%) +#define TIER_3 2 // [75% .. 85%) +#define TIER_4 2 // [85% .. 100%] + assert(pn_raw_conn); pn_raw_buffer_t raw_buffers[RAW_BUFFER_BATCH]; - size_t desired = pn_raw_connection_read_buffers_capacity(pn_raw_conn); - const size_t granted = desired; - while (desired) { + // + // Get the read-buffer capacity for the connection. + // + size_t capacity = pn_raw_connection_read_buffers_capacity(pn_raw_conn); + + // + // If there's no capacity, exit now before doing any further wasted work. + // + if (capacity == 0) { + return 0; + } + + // + // Since we can't query Proton for the maximum read-buffer capacity, we will infer it from + // calls to pn_raw_connection_read_buffers_capacity. + // + static size_t max_capacity = 0; + if (capacity > max_capacity) { + max_capacity = capacity; + } + + // + // Get the "held_by_threads" stats for adaptor buffers as an approximation of how many + // buffers are in-use. This is an approximation since it also counts free buffers held + // in the per-thread free-pools. Since we will be dealing with large numbers here, the + // number of buffers in free-pools will not be significant. + // + // Note that there is a thread race on the access of this value. There's no danger associated + // with getting a partial or corrupted value from time to time. + // + // Note also that the stats pointer may be NULL if no buffers have yet been allocated. + // + qd_alloc_stats_t *stats = alloc_stats_qd_adaptor_buffer_t(); + uint64_t buffers_in_use = !!stats ? stats->held_by_threads : 0; + + // + // Choose the grant-allocation tier based on the number of buffers in use. + // + size_t desired = TIER_4; + if (buffers_in_use < buffer_threshold_50) { + desired = TIER_1; + } else if (buffers_in_use < buffer_threshold_75) { + desired = TIER_2; + } else if (buffers_in_use < buffer_threshold_85) { + desired = TIER_3; + } + + // + // Determine how many of the desired buffers are already granted. This will always be a + // non-negative value. + // + size_t already_granted = max_capacity - capacity; + + // + // If we desire to grant additional buffers, calculate the number to grant now. + // + const size_t to_grant = desired > already_granted ? desired - already_granted : 0; + size_t count = to_grant; + + // + // Grant the buffers in batches. + // + while (count) { int i; - for (i = 0; i < desired && i < RAW_BUFFER_BATCH; ++i) { + for (i = 0; i < count && i < RAW_BUFFER_BATCH; ++i) { qd_adaptor_buffer_t *buf = qd_adaptor_buffer(); raw_buffers[i].bytes = (char *) qd_adaptor_buffer_base(buf); raw_buffers[i].capacity = qd_adaptor_buffer_capacity(buf); @@ -89,11 +189,11 @@ int qd_raw_connection_grant_read_buffers(pn_raw_connection_t *pn_raw_conn) raw_buffers[i].offset = 0; raw_buffers[i].context = (uintptr_t) buf; } - desired -= i; + count -= i; pn_raw_connection_give_read_buffers(pn_raw_conn, raw_buffers, i); } - return granted; + return to_grant; } int qd_raw_connection_write_buffers(pn_raw_connection_t *pn_raw_conn, qd_adaptor_buffer_list_t *blist) diff --git a/src/adaptors/adaptor_common.h b/src/adaptors/adaptor_common.h index 1db8b77ab..db172db75 100644 --- a/src/adaptors/adaptor_common.h +++ b/src/adaptors/adaptor_common.h @@ -67,6 +67,11 @@ ALLOC_DECLARE(qd_adaptor_config_t); qd_error_t qd_load_adaptor_config(qd_adaptor_config_t *config, qd_entity_t *entity); void qd_free_adaptor_config(qd_adaptor_config_t *config); +/** + * Perform router-startup actions used in the common module. + */ +void qd_adaptor_common_init(void); + /** * Grants as many read qd_adaptor buffers as returned by pn_raw_connection_read_buffers_capacity(). * Maximum read capacity is set to 16 in proton raw api. diff --git a/src/adaptors/tcp/tcp_adaptor.c b/src/adaptors/tcp/tcp_adaptor.c index 16b7d7cb2..aea7fdc48 100644 --- a/src/adaptors/tcp/tcp_adaptor.c +++ b/src/adaptors/tcp/tcp_adaptor.c @@ -2234,6 +2234,7 @@ static void qdr_tcp_activate_CT(void *notused, qdr_connection_t *c) */ static void qdr_tcp_adaptor_init(qdr_core_t *core, void **adaptor_context) { + qd_adaptor_common_init(); qdr_tcp_adaptor_t *adaptor = NEW(qdr_tcp_adaptor_t); adaptor->core = core; adaptor->adaptor = qdr_protocol_adaptor(core, diff --git a/tests/tsan.supp b/tests/tsan.supp index 897eea5a1..4e405922c 100644 --- a/tests/tsan.supp +++ b/tests/tsan.supp @@ -83,6 +83,9 @@ race:^__lws_logv$ # ISSUE-537 - Suppress the race in qd_entity_refresh_connector for now. race:^qd_entity_refresh_connector$ +# #1107 - Suppress warnings on benign race in accessing memory stats +race:^qd_raw_connection_grant_read_buffers$ + # # External libraries