FIX: Improve PRNG seeding on Windows to ensure uniqueness of generated numbers.

Fix for the pseudo-random seed on Windows. The function `rand_r` isn't present on Windows and the global seed wasn't based on the current microseconds and thread id. Also it wasn't called on every thread as required on this platform but only once per process. The fix allows on this platform the uniqueness of client side member id generation in next-generation consumer group protocol.
Happening since 1.x

Author: emasab

CHANGELOG.md

@@ -1,3 +1,33 @@
+# librdkafka v2.13.0
+
+librdkafka v2.13.0 is a feature release:
+
+* Fix memory management for interceptors in rd_kafka_conf to prevent
+double-free errors (#5240).
+* Fix for the pseudo-random generator seed on Windows involving as well
+  the uniqueness of the new consumer group protocol member id (#).
+
+
+## Fixes
+
+### General fixes
+
+* Issues: #4142.
+  Fix memory management for interceptors in rd_kafka_conf to prevent double-free errors.
+  In case the client instance fails the users needs to destroy the configuration
+  data structure, it was causing a double-free because the interceptors were
+  already freed in the constructor.
+  Happening since 1.x (#5240).
+* Issues: #.
+  Fix for the pseudo-random seed on Windows. The function `rand_r` isn't present
+  on Windows and the global seed wasn't based on the current microseconds and thread
+  id. Also it wasn't called on every thread as required on this platform but
+  only once per process. The fix allows on this platform the uniqueness of client side 
+  member id generation in next-generation consumer group protocol.
+  Happening since 1.x (#).
+
+
+
 # librdkafka v2.12.1
 
 librdkafka v2.12.1 is a maintenance release:

src/rdkafka.c

@@ -57,6 +57,7 @@
 #include "rdkafka_interceptor.h"
 #include "rdkafka_idempotence.h"
 #include "rdkafka_sasl_oauthbearer.h"
+#include "rdmurmur2.h"
 #if WITH_OAUTHBEARER_OIDC
 #include "rdkafka_sasl_oauthbearer_oidc.h"
 #endif
@@ -82,8 +83,14 @@
 #endif
 
 
-static once_flag rd_kafka_global_init_once  = ONCE_FLAG_INIT;
-static once_flag rd_kafka_global_srand_once = ONCE_FLAG_INIT;
+static once_flag rd_kafka_global_init_once = ONCE_FLAG_INIT;
+#ifdef _WIN32
+/* On Windows srand needs to be called on each thread. */
+static RD_TLS once_flag rd_kafka_srand_once = ONCE_FLAG_INIT;
+#else
+static once_flag rd_kafka_srand_once = ONCE_FLAG_INIT;
+#endif
+
 
 /**
  * @brief Global counter+lock for all active librdkafka instances
@@ -130,6 +137,22 @@ void rd_kafka_set_thread_name(const char *fmt, ...) {
  */
 static char RD_TLS rd_kafka_thread_sysname[16] = "app";
 
+/**
+ * @brief Seed the PRNG with current microseconds and thread ID.
+ */
+static void rd_kafka_srand(void) {
+        unsigned int seed = 0;
+        struct timeval tv;
+        rd_gettimeofday(&tv, NULL);
+        seed = (unsigned int)(tv.tv_usec);
+        seed ^= thrd_current_id();
+
+        /* Apply the murmur2 hash to distribute entropy to
+         * the whole seed. */
+        seed = (unsigned int)rd_murmur2(&seed, sizeof(seed));
+        srand(seed);
+}
+
 void rd_kafka_set_thread_sysname(const char *fmt, ...) {
         va_list ap;
 
@@ -141,6 +164,30 @@ void rd_kafka_set_thread_sysname(const char *fmt, ...) {
         thrd_setname(rd_kafka_thread_sysname);
 }
 
+/**
+ * @brief Seed the PRNG for the current thread or for the whole process.
+ *        Depending on the platform implementation of srand() the seed can
+ *        be a thread local or global one. In case it's thread local we
+ *        need to call it on each thread.
+ *
+ * @param rk Client instance.
+ * @param internal_thread If true, seed the PRNG if
+ *                        it's required per-thread.
+ */
+void rd_kafka_thread_srand(rd_kafka_t *rk, rd_bool_t internal_thread) {
+#ifdef _WIN32
+        rd_bool_t required_per_thread = rd_true;
+#else
+        rd_bool_t required_per_thread = rd_false;
+#endif
+        if ((required_per_thread &&
+             (rk->rk_conf.enable_random_seed || internal_thread)) ||
+            (!required_per_thread && rk->rk_conf.enable_random_seed &&
+             !internal_thread)) {
+                call_once(&rd_kafka_srand_once, rd_kafka_srand);
+        }
+}
+
 static void rd_kafka_global_init0(void) {
         cJSON_Hooks json_hooks = {.malloc_fn = rd_malloc, .free_fn = rd_free};
 
@@ -171,18 +218,6 @@ void rd_kafka_global_init(void) {
 }
 
 
-/**
- * @brief Seed the PRNG with current_time.milliseconds
- */
-static void rd_kafka_global_srand(void) {
-        struct timeval tv;
-
-        rd_gettimeofday(&tv, NULL);
-
-        srand((unsigned int)(tv.tv_usec / 1000));
-}
-
-
 /**
  * @returns the current number of active librdkafka instances
  */
@@ -2218,6 +2253,7 @@ static int rd_kafka_thread_main(void *arg) {
 
         rd_kafka_set_thread_name("main");
         rd_kafka_set_thread_sysname("rdk:main");
+        rd_kafka_thread_srand(rk, rd_true /* we're in an internal thread */);
 
         rd_kafka_interceptors_on_thread_start(rk, RD_KAFKA_THREAD_MAIN);
 
@@ -2367,10 +2403,7 @@ rd_kafka_t *rd_kafka_new(rd_kafka_type_t type,
                                 * freed from rd_kafka_destroy_internal()
                                 * as the rk itself is destroyed. */
 
-        /* Seed PRNG, don't bother about HAVE_RAND_R, since it is pretty cheap.
-         */
-        if (rk->rk_conf.enable_random_seed)
-                call_once(&rd_kafka_global_srand_once, rd_kafka_global_srand);
+        rd_kafka_thread_srand(rk, rd_false /* we're on an app thread */);
 
         /* Call on_new() interceptors */
         rd_kafka_interceptors_on_new(rk, &rk->rk_conf);

src/rdkafka_background.c

@@ -111,6 +111,7 @@ int rd_kafka_background_thread_main(void *arg) {
 
         rd_kafka_set_thread_name("background");
         rd_kafka_set_thread_sysname("rdk:bg");
+        rd_kafka_thread_srand(rk, rd_true /* we're in an internal thread */);
 
         rd_kafka_interceptors_on_thread_start(rk, RD_KAFKA_THREAD_BACKGROUND);
 

src/rdkafka_broker.c

@@ -4512,6 +4512,7 @@ static int rd_kafka_broker_thread_main(void *arg) {
 
         rd_kafka_set_thread_name("%s", rkb->rkb_name);
         rd_kafka_set_thread_sysname("rdk:broker%" PRId32, rkb->rkb_nodeid);
+        rd_kafka_thread_srand(rk, rd_true /* we're in an internal thread */);
 
         rd_kafka_interceptors_on_thread_start(rk, RD_KAFKA_THREAD_BROKER);
 

src/rdkafka_int.h

@@ -1122,6 +1122,7 @@ extern char RD_TLS rd_kafka_thread_name[64];
 
 void rd_kafka_set_thread_name(const char *fmt, ...) RD_FORMAT(printf, 1, 2);
 void rd_kafka_set_thread_sysname(const char *fmt, ...) RD_FORMAT(printf, 1, 2);
+void rd_kafka_thread_srand(rd_kafka_t *rk, rd_bool_t internal_thread);
 
 int rd_kafka_path_is_dir(const char *path);
 rd_bool_t rd_kafka_dir_is_empty(const char *path);

src/rdkafka_mock.c

@@ -1490,6 +1490,8 @@ static int rd_kafka_mock_cluster_thread_main(void *arg) {
 
         rd_kafka_set_thread_name("mock");
         rd_kafka_set_thread_sysname("rdk:mock");
+        rd_kafka_thread_srand(mcluster->rk,
+                              rd_true /* we're in an internal thread */);
         rd_kafka_interceptors_on_thread_start(mcluster->rk,
                                               RD_KAFKA_THREAD_BACKGROUND);
         rd_atomic32_add(&rd_kafka_thread_cnt_curr, 1);

src/rdkafka_ssl.c

@@ -1993,14 +1993,7 @@ rd_kafka_transport_ssl_lock_cb(int mode, int i, const char *file, int line) {
 #endif
 
 static RD_UNUSED unsigned long rd_kafka_transport_ssl_threadid_cb(void) {
-#ifdef _WIN32
-        /* Windows makes a distinction between thread handle
-         * and thread id, which means we can't use the
-         * thrd_current() API that returns the handle. */
-        return (unsigned long)GetCurrentThreadId();
-#else
-        return (unsigned long)(intptr_t)thrd_current();
-#endif
+        return thrd_current_id();
 }
 
 #ifdef HAVE_OPENSSL_CRYPTO_THREADID_SET_CALLBACK

src/rdrand.c

@@ -42,7 +42,7 @@ int rd_jitter(int low, int high) {
                 struct timeval tv;
                 rd_gettimeofday(&tv, NULL);
                 seed = (unsigned int)(tv.tv_usec);
-                seed ^= (unsigned int)(intptr_t)thrd_current();
+                seed ^= thrd_current_id();
 
                 /* When many threads are created at the same time and the
                  * thread id is different only by a few bits it's possible that

src/tinycthread_extra.c

@@ -59,6 +59,17 @@ int thrd_is_current(thrd_t thr) {
 #endif
 }
 
+unsigned long thrd_current_id(void) {
+#ifdef _WIN32
+        /* Windows makes a distinction between thread handle
+         * and thread id, which means we can't use the
+         * thrd_current() API that returns the handle. */
+        return (unsigned long)GetCurrentThreadId();
+#else
+        return (unsigned long)(intptr_t)thrd_current();
+#endif
+}
+
 
 #ifdef _WIN32
 void cnd_wait_enter(cnd_t *cond) {

src/tinycthread_extra.h

@@ -55,6 +55,12 @@ int thrd_setname(const char *name);
  */
 int thrd_is_current(thrd_t thr);
 
+/**
+ * @brief Get current thread ID as an unsigned long.
+ * @return Current thread ID.
+ */
+unsigned long thrd_current_id(void);
+
 
 #ifdef _WIN32
 /**