gh-426 Implement lock-free fast_random function

Co-authored-by: Victor Gaydov <[email protected]>

src/internal_modules/roc_audio/packetizer.cpp

@@ -34,10 +34,11 @@ Packetizer::Packetizer(packet::IWriter& writer,
     , payload_size_(payload_encoder.encoded_byte_count(samples_per_packet_))
     , packet_pos_(0)
     , valid_(false) {
-    source_ = (packet::stream_source_t)core::fast_random(0, packet::stream_source_t(-1));
-    seqnum_ = (packet::seqnum_t)core::fast_random(0, packet::seqnum_t(-1));
-    stream_ts_ =
-        (packet::stream_timestamp_t)core::fast_random(0, packet::stream_timestamp_t(-1));
+    source_ =
+        (packet::stream_source_t)core::fast_random_range(0, packet::stream_source_t(-1));
+    seqnum_ = (packet::seqnum_t)core::fast_random_range(0, packet::seqnum_t(-1));
+    stream_ts_ = (packet::stream_timestamp_t)core::fast_random_range(
+        0, packet::stream_timestamp_t(-1));
     capture_ts_ = 0;
     valid_ = true;
     roc_log(LogDebug, "packetizer: initializing: n_channels=%lu samples_per_packet=%lu",

src/internal_modules/roc_core/fast_random.cpp

@@ -0,0 +1,70 @@
+/*
+ * Copyright (c) 2015 Roc Streaming authors
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ */
+
+#include "roc_core/fast_random.h"
+#include "roc_core/atomic_ops.h"
+#include "roc_core/panic.h"
+#include "roc_core/time.h"
+
+namespace roc {
+namespace core {
+
+namespace {
+
+uint32_t state;
+
+} // namespace
+
+// PRNG implementation is a lock-free adaptation of splitmix32 by Tommy Ettinger:
+// https://gist.github.com/tommyettinger/46a874533244883189143505d203312c
+//
+// This implementation is not a cryptographically secure PRNG.
+uint32_t fast_random() {
+    if (AtomicOps::load_relaxed(state) == 0) {
+        uint32_t expected_state = 0;
+        uint32_t new_state = (uint32_t)core::timestamp(core::ClockMonotonic);
+        AtomicOps::compare_exchange_seq_cst(state, expected_state, new_state);
+    }
+
+    uint32_t z;
+
+    z = AtomicOps::fetch_add_seq_cst(state, 0x9E3779B9);
+    z = z ^ (z >> 16);
+    z *= 0x21F0AAAD;
+    z = z ^ (z >> 15);
+    z *= 0x735A2D97;
+    z = z ^ (z >> 15);
+    return z;
+}
+
+// Bounded PRNG implementation is based on "Debiased Modulo (Once) — Java's Method"
+// algorithm: https://www.pcg-random.org/posts/bounded-rands.html
+//
+// This implementation is not a cryptographically secure PRNG.
+uint32_t fast_random_range(uint32_t from, uint32_t to) {
+    roc_panic_if_not(from <= to);
+
+    const uint64_t range = uint64_t(to) - from + 1;
+
+    uint64_t z, r;
+
+    do {
+        z = fast_random();
+        r = z % range;
+    } while (z - r > (-range));
+
+    const uint32_t ret = from + (uint32_t)r;
+
+    roc_panic_if_not(ret >= from);
+    roc_panic_if_not(ret <= to);
+
+    return ret;
+}
+
+} // namespace core
+} // namespace roc

src/internal_modules/roc_core/target_posix/roc_core/fast_random.h → src/internal_modules/roc_core/fast_random.h

@@ -6,7 +6,7 @@
  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
  */
 
-//! @file roc_core/target_posix/roc_core/fast_random.h
+//! @file roc_core/fast_random.h
 //! @brief Helpers to work with random numbers.
 
 #ifndef ROC_CORE_FAST_RANDOM_H_
@@ -20,7 +20,12 @@ namespace core {
 //! Get a random integer from a non cryptographically secure, but fast PRNG.
 //! Thread-safe.
 //! @returns random value in range [from; to].
-uint32_t fast_random(uint32_t from, uint32_t to);
+uint32_t fast_random_range(uint32_t from, uint32_t to);
+
+//! Get a random integer from a non cryptographically secure, but fast PRNG.
+//! Thread-safe.
+//! @returns random value between 0 and UINT32_MAX.
+uint32_t fast_random();
 
 } // namespace core
 } // namespace roc

src/internal_modules/roc_fec/writer.cpp

@@ -41,8 +41,9 @@ Writer::Writer(const WriterConfig& config,
     , fec_scheme_(fec_scheme)
     , valid_(false)
     , alive_(true) {
-    cur_sbn_ = (packet::blknum_t)core::fast_random(0, packet::blknum_t(-1));
-    cur_block_repair_sn_ = (packet::seqnum_t)core::fast_random(0, packet::seqnum_t(-1));
+    cur_sbn_ = (packet::blknum_t)core::fast_random_range(0, packet::blknum_t(-1));
+    cur_block_repair_sn_ =
+        (packet::seqnum_t)core::fast_random_range(0, packet::seqnum_t(-1));
     if (!resize(config.n_source_packets, config.n_repair_packets)) {
         return;
     }

src/internal_modules/roc_packet/interleaver.cpp

@@ -81,7 +81,7 @@ void Interleaver::reinit_seq_() {
         send_seq_[i] = i;
     }
     for (size_t i = block_size_; i > 0; --i) {
-        const size_t j = core::fast_random(0, (unsigned int)i - 1);
+        const size_t j = core::fast_random_range(0, (unsigned int)i - 1);
         const size_t buff = send_seq_[i - 1];
         send_seq_[i - 1] = send_seq_[j];
         send_seq_[j] = buff;

src/internal_modules/roc_rtcp/session.cpp

@@ -31,7 +31,8 @@ Session::Session(IReceiverHooks* recv_hooks,
     , next_deadline_(0)
     , ssrc_(0)
     , valid_(false) {
-    ssrc_ = (packet::stream_source_t)core::fast_random(0, packet::stream_source_t(-1));
+    ssrc_ =
+        (packet::stream_source_t)core::fast_random_range(0, packet::stream_source_t(-1));
 
     // TODO
     strcpy(cname_, "TODO");

src/tests/roc_core/bench_random.cpp

@@ -17,7 +17,7 @@ namespace {
 void BM_Random_Fast(benchmark::State& state) {
     uint32_t r = 0;
     while (state.KeepRunning()) {
-        r = fast_random(r, (uint32_t)-1);
+        r = fast_random_range(r, (uint32_t)-1);
         benchmark::DoNotOptimize(r);
     }
 }

src/tests/roc_core/test_fast_random.cpp

@@ -20,20 +20,20 @@ TEST_GROUP(fast_random) {};
 
 TEST(fast_random, min_min) {
     uint32_t lo_hi = 0;
-    uint32_t res = fast_random(lo_hi, lo_hi);
+    uint32_t res = fast_random_range(lo_hi, lo_hi);
 
     LONGS_EQUAL(res, lo_hi);
 }
 
 TEST(fast_random, max_max) {
     uint32_t lo_hi = UINT32_MAX;
-    uint32_t res = fast_random(lo_hi, lo_hi);
+    uint32_t res = fast_random_range(lo_hi, lo_hi);
 
     LONGS_EQUAL(res, lo_hi);
 }
 
 TEST(fast_random, limits) {
-    uint32_t res = fast_random(1, 100);
+    uint32_t res = fast_random_range(1, 100);
 
     CHECK(1 <= res && res <= 100);
 }

src/tests/roc_ctl/bench_task_queue_contention.cpp

@@ -80,7 +80,7 @@ BENCHMARK_DEFINE_F(BM_QueueContention, ScheduleAt)(benchmark::State& state) {
 
     core::nanoseconds_t* delays = new core::nanoseconds_t[NumScheduleAfterIterations];
     for (int n = 0; n < NumScheduleAfterIterations; n++) {
-        delays[n] = core::fast_random(0, MaxDelay);
+        delays[n] = core::fast_random_range(0, MaxDelay);
     }
 
     while (state.KeepRunningBatch(BatchSize)) {

src/tests/roc_fec/test_encoder_decoder.cpp

@@ -84,7 +84,7 @@ class Codec {
         core::Slice<uint8_t> buf = buffer_factory.new_buffer();
         buf.reslice(0, p_size);
         for (size_t j = 0; j < buf.size(); ++j) {
-            buf.data()[j] = (uint8_t)core::fast_random(0, 0xff);
+            buf.data()[j] = (uint8_t)core::fast_random_range(0, 0xff);
         }
         return buf;
     }
@@ -170,7 +170,8 @@ TEST(encoder_decoder, random_losses) {
 
             size_t curr_loss = 0;
             for (size_t i = 0; i < NumSourcePackets + NumRepairPackets; ++i) {
-                if (core::fast_random(0, 100) < LossPercent && curr_loss <= MaxLoss) {
+                if (core::fast_random_range(0, 100) < LossPercent
+                    && curr_loss <= MaxLoss) {
                     total_loss++;
                     curr_loss++;
                 } else {

src/tests/roc_netio/test_helpers/conn_reader.h

@@ -43,7 +43,7 @@ class ConnReader : public core::Thread {
             handler_.wait_readable();
 
             while (num_read < total_bytes_) {
-                size_t bufsz = core::fast_random(1, MaxBatch);
+                size_t bufsz = core::fast_random_range(1, MaxBatch);
                 if (bufsz > (total_bytes_ - num_read)) {
                     bufsz = (total_bytes_ - num_read);
                 }

src/tests/roc_netio/test_helpers/conn_writer.h

@@ -44,12 +44,13 @@ class ConnWriter : public core::Thread {
             handler_.wait_writable();
 
             while (num_written < total_bytes_) {
-                const core::nanoseconds_t delay = (core::nanoseconds_t)core::fast_random(
-                    0, MaxDelayNs * core::Nanosecond);
+                const core::nanoseconds_t delay =
+                    (core::nanoseconds_t)core::fast_random_range(
+                        0, MaxDelayNs * core::Nanosecond);
 
                 core::sleep_for(core::ClockMonotonic, delay);
 
-                size_t bufsz = core::fast_random(1, MaxBatch);
+                size_t bufsz = core::fast_random_range(1, MaxBatch);
                 if (bufsz > (total_bytes_ - num_written)) {
                     bufsz = (total_bytes_ - num_written);
                 }

src/tests/roc_pipeline/bench_pipeline_loop_peak_load.cpp

@@ -336,8 +336,8 @@ class TestPipeline : public PipelineLoop,
     virtual bool process_task_imp(PipelineTask& basic_task) {
         Task& task = (Task&)basic_task;
         stats_.task_processing_started(task.elapsed_time());
-        busy_wait(
-            core::fast_random(MinTaskProcessingDuration, MaxTaskProcessingDuration));
+        busy_wait(core::fast_random_range(MinTaskProcessingDuration,
+                                          MaxTaskProcessingDuration));
         return true;
     }
 
@@ -375,9 +375,9 @@ class TaskThread : public core::Thread, private IPipelineTaskCompleter {
     virtual void run() {
         while (!stop_) {
             core::sleep_for(core::ClockMonotonic,
-                            core::fast_random(MinTaskDelay, MaxTaskDelay));
+                            core::fast_random_range(MinTaskDelay, MaxTaskDelay));
 
-            const size_t n_tasks = core::fast_random(MinTaskBurst, MaxTaskBurst);
+            const size_t n_tasks = core::fast_random_range(MinTaskBurst, MaxTaskBurst);
 
             for (size_t n = 0; n < n_tasks; n++) {
                 TestPipeline::Task* task = new TestPipeline::Task;