【问题标题】:How can I generate sorted uniformly distributed random numbers efficiently in C++?如何在 C++ 中有效地生成排序均匀分布的随机数?
【发布时间】:2020-12-05 01:48:07
【问题描述】:

我想在 C++ 中生成一个大数 n(即 n >= 1,000,000,000),这些随机数是有序且均匀分布的随机数。

我考虑的第一个简单的方法是

  1. 使用std::uniform_real_distribution<double>顺序生成n均匀分布的数字,
  2. 然后使用std::sort对它们进行排序。

但是,这需要几分钟时间。

更复杂的方法是并行化这两个步骤,如下所示:

template <typename T>
void computeUniformDistribution(std::vector<T>& elements)
{
    #pragma omp parallel
    {
        std::seed_seq seed{distribution_seed, static_cast<size_t>(omp_get_thread_num())};
        std::mt19937 prng = std::mt19937(seed);
        std::uniform_real_distribution<double> uniform_dist(0, std::numeric_limits<T>::max());

        #pragma omp for
        for (size_t i = 0; i < elements.size(); ++i)
        {
            elements[i] = static_cast<T>(uniform_dist(prng));
        }
    }

    std::sort(std::execution::par_unseq, elements.begin(), elements.end());
}

但是,即使这样也需要大约 30 秒。鉴于生成均匀分布的数字只需要大约 1.5 秒,瓶颈仍然是排序阶段。

因此,我想问以下问题:我怎样才能以排序的方式有效地生成均匀分布的数据?

【问题讨论】:

  • 评论不用于扩展讨论;这个对话是moved to chat
  • Your code is not seeding the generator correctly,因此您生成的序列将偏向一些可能的样本,而不是对整个可用空间进行采样。
  • @KonradRudolph 你能详细说明一下吗?
  • codereview.stackexchange.com/q/109260/308 和那里的讨论cmets。并且不要只阅读答案,特别要注意当前接受的答案是完全错误的。
  • @KonradRudolph 如果错了,而且你已经说了很多年了,你能写一个更好的答案吗?

标签: c++ algorithm sorting random c++17


【解决方案1】:

有一些方法可以生成已经排序的样本,但我认为生成部分排序的样本可能会更好。

将输出范围划分为 k 个等宽的桶。每个桶中的样本数将具有等概率的多项分布。对多项分布进行采样的慢速方法是在 [0, k) 中生成 n 个整数。一种更有效的方法是抽取 k 个泊松样本,其速率为 n/k,条件是它们的总和不超过 n,然后使用慢速方式添加另一个 n-sum 样本。完美地对泊松分布进行采样是很棘手的,但是当 n/k 非常大时(就像这里所说的那样),泊松分布可以通过对具有均值和方差 n/k 的正态分布进行四舍五入来很好地逼近。如果这是不可接受的,那么慢速方法确实可以很好地并行化。

给定存储桶计数,计算前缀总和以找到存储桶边界。对于并行的每个桶,在桶范围内生成给定数量的样本并对其进行排序。如果我们选择好 n/k,每个桶几乎肯定会适合 L1 缓存。对于 n = 1e9,我想我会尝试 k = 1e5 或 k = 1e6。

这是一个顺序实现。有点粗糙,因为我们确实需要避免对封闭的桶边界进行 2 倍过采样,但我会把它留给你。我对 OMP 不熟悉,但我认为您可以通过在 SortedUniformSamples 末尾的 for 循环中添加 pragma 来获得相当不错的并行实现。

#include <algorithm>
#include <cmath>
#include <iostream>
#include <numeric>
#include <random>
#include <span>
#include <vector>

template <typename Dist, typename Gen>
void SortedSamples(std::span<double> samples, Dist dist, Gen& gen) {
  for (double& sample : samples) {
    sample = dist(gen);
  }
  std::sort(samples.begin(), samples.end());
}

template <typename Gen>
void ApproxMultinomialSample(std::span<std::size_t> samples, std::size_t n,
                             Gen& gen) {
  double lambda = static_cast<double>(n) / samples.size();
  std::normal_distribution<double> approx_poisson{lambda, std::sqrt(lambda)};
  std::size_t sum;
  do {
    for (std::size_t& sample : samples) {
      sample = std::lrint(approx_poisson(gen));
    }
    sum = std::accumulate(samples.begin(), samples.end(), std::size_t{0});
  } while (sum > n);
  std::uniform_int_distribution<std::size_t> uniform{0, samples.size() - 1};
  for (; sum < n; sum++) {
    samples[uniform(gen)]++;
  }
}

template <typename Gen>
void SortedUniformSamples(std::span<double> samples, Gen& gen) {
  static constexpr std::size_t kTargetBucketSize = 1024;
  if (samples.size() < kTargetBucketSize) {
    SortedSamples(samples, std::uniform_real_distribution<double>{0, 1}, gen);
    return;
  }
  std::size_t num_buckets = samples.size() / kTargetBucketSize;
  std::vector<std::size_t> bucket_counts(num_buckets);
  ApproxMultinomialSample(bucket_counts, samples.size(), gen);
  std::vector<std::size_t> prefix_sums(num_buckets + 1);
  std::partial_sum(bucket_counts.begin(), bucket_counts.end(),
                   ++prefix_sums.begin());
  for (std::size_t i = 0; i < num_buckets; i++) {
    SortedSamples(std::span<double>{&samples[prefix_sums[i]],
                                    &samples[prefix_sums[i + 1]]},
                  std::uniform_real_distribution<double>{
                      static_cast<double>(i) / num_buckets,
                      static_cast<double>(i + 1) / num_buckets},
                  gen);
  }
}

int main() {
  std::vector<double> samples(100000000);
  std::default_random_engine gen;
  SortedUniformSamples(samples, gen);
  if (std::is_sorted(samples.begin(), samples.end())) {
    std::cout << "sorted\n";
  }
}

如果您的标准库有 poisson_distribution 的高质量实现,您也可以这样做:

template <typename Gen>
void MultinomialSample(std::span<std::size_t> samples, std::size_t n,
                       Gen& gen) {
  double lambda = static_cast<double>(n) / samples.size();
  std::poisson_distribution<std::size_t> poisson{lambda};
  std::size_t sum;
  do {
    for (std::size_t& sample : samples) {
      sample = poisson(gen);
    }
    sum = std::accumulate(samples.begin(), samples.end(), std::size_t{0});
  } while (sum > n);
  std::uniform_int_distribution<std::size_t> uniform{0, samples.size() - 1};
  for (; sum < n; sum++) {
    samples[uniform(gen)]++;
  }
}

【讨论】:

  • 桶大小(意思是值的数量)是否等同于生成完整向量然后用相同的固定桶边界值对元素进行分桶时获得的桶大小? (后者是我想到的,每个核心抓取其分配的桶范围的值,对它们进行排序,然后将它们写回到它们所属的源向量中。)
  • @superbrain 如果我们能够准确地采样泊松分布,那么是的,它将完全等价。但是,当 lambda > 1024 时,正态分布是一个非常非常好的近似值。
  • @superbrain 添加了一个精确的变体,假设std::poisson_distribution 的快速精确实现将很快。
  • 很好,谢谢。我无法真正判断/理解数学,但很高兴知道这两种方法在这个意义上是等价的。我想说我的更简单,但你的更高效(因为核心不需要查找/收集属于其范围的元素,只需生成它们)。
  • 为什么不简单地使用桶排序呢?这种方法与桶排序有许多相似之处,请参阅我的回答。
【解决方案2】:

我进行了一些测试,基数排序的速度是 std::sort 的 4 到 6 倍,具体取决于系统,但它需要第二个向量,对于 1 GB 的元素,每个双精度向量为 8 GB,对于总共 16 GB 的可用内存,因此您可能需要 32 GB 的 RAM。

如果排序不受内存带宽限制,多线程基数排序可能会有所帮助。

单线程代码示例:

#include <algorithm>
#include <iostream>
#include <random>
#include <vector>
#include <time.h>

clock_t ctTimeStart;            // clock values
clock_t ctTimeStop;

typedef unsigned long long uint64_t;

//  a is input array, b is working array
uint64_t * RadixSort(uint64_t * a, uint64_t *b, size_t count)
{
uint32_t mIndex[8][256] = {0};          // count / index matrix
uint32_t i,j,m,n;
uint64_t u;
    for(i = 0; i < count; i++){         // generate histograms
        u = a[i];
        for(j = 0; j < 8; j++){
            mIndex[j][(size_t)(u & 0xff)]++;
            u >>= 8;
        }
    }
    for(j = 0; j < 8; j++){             // convert to indices
        m = 0;
        for(i = 0; i < 256; i++){
            n = mIndex[j][i];
            mIndex[j][i] = m;
            m += n;
        }
    }
    for(j = 0; j < 8; j++){             // radix sort
        for(i = 0; i < count; i++){     //  sort by current LSB
            u = a[i];
            m = (size_t)(u>>(j<<3))&0xff;
            b[mIndex[j][m]++] = u;
        }
        std::swap(a, b);                //  swap ptrs
    }
    return(a);
}

#define COUNT (1024*1024*1024)

int main(int argc, char**argv)
{
    std::vector<double> v(COUNT);       // vctr to be generated
    std::vector<double> t(COUNT);       // temp vector
    std::random_device rd;
    std::mt19937 gen(rd());
//  std::uniform_real_distribution<> dis(0, std::numeric_limits<double>::max());
    std::uniform_real_distribution<> dis(0, COUNT);
    ctTimeStart = clock();
    for(size_t i = 0; i < v.size(); i++)
        v[i] = dis(gen);
    ctTimeStop = clock();
    std::cout << "# of ticks " << ctTimeStop - ctTimeStart << std::endl;
    ctTimeStart = clock();
//  std::sort(v.begin(), v.end());
    RadixSort((uint64_t *)&v[0], (uint64_t *)&t[0], COUNT);
    ctTimeStop = clock();
    std::cout << "# of ticks " << ctTimeStop - ctTimeStart << std::endl;
    return(0);
}

如果对包含负值的双精度数(转换为 64 位无符号整数)进行排序,您需要将它们视为符号 + 幅度 64 位整数。用于将符号 + 幅度 (SM) 与 64 位无符号整数 (ULL) 相互转换的 C++ 宏:

// converting doubles to unsigned long long for radix sort or something similar
// note -0 converted to 0x7fffffffffffffff, +0 converted to 0x8000000000000000
// -0 is unlikely to be produced by a float operation

#define SM2ULL(x) ((x)^(((~(x) >> 63)-1) | 0x8000000000000000ull))
#define ULL2SM(x) ((x)^((( (x) >> 63)-1) | 0x8000000000000000ull))

【讨论】:

    【解决方案3】:

    我很想依赖这样一个事实,即一组有序的均匀分布变量的连续元素之间的差异呈指数分布。这可以被利用在O(N) 时间而不是O(N*log N) 中运行。

    快速实现会执行以下操作:

    template<typename T> void
    computeSorteUniform2(std::vector<T>& elements)
    {
        std::random_device rd;
        std::mt19937 prng(rd());
    
        std::exponential_distribution<T> dist(static_cast<T>(1));
    
        auto sum = dist(prng);
    
        for (auto& elem : elements) {
            elem = sum += dist(prng);
        }
    
        sum += dist(prng);
    
        for (auto& elem : elements) {
            elem /= sum;
        }
    }
    

    通过假设您需要 Uniform(0, 1) 中的值来简化此示例,但它应该很容易概括。使用 OMP 完成这项工作并非易事,但应该不会太难。

    如果您关心最后约 50% 的性能,那么有一些数字技巧可能会加速生成随机偏差(例如,有比 MT 更快更好的 PRNG)以及将它们转换为 doubles(但最近编译器可能知道这些技巧)。几个参考:Daniel Lemire's blogMelissa O'Neill's PCG site

    我刚刚对此进行了基准测试,发现clang 的std::uniform_real_distributionstd::exponential_distribution 都非常慢。 numpy's Ziggurat based implementations 快 8 倍,因此我可以使用上述算法在我的笔记本电脑上使用单个线程在约 10 秒内生成 1e9 个double(即 std 实现需要约 80 秒)。我没有在 1e9 元素上尝试过 OP 的实现,但使用 1e8 元素我的速度要快 15 倍左右。

    【讨论】:

    • 这是否意味着我的答案的偏见(如 cmets 中所述)有些好?
    • 哎呀,显然我们的两个统计数据都是错误的。我刚刚进行了实证检查并提醒自己分布是指数的,而不是均匀的......将修正我的答案感谢您的评论!
    • 指数分布是正确的。有了这种变化,这是一个很好的解决方案!具体结果:设E_1, E_2, ..., E_{n+1}为独立的指数(1)随机变量。令 S=E_1+...+E_{n+1}。那么(E_1/S, E_2/S..., E_n/S)的分布与n个独立随机变量均匀分布在(0,1)上并按升序排列的分布相同。参见例如djalil.chafai.net/blog/2014/06/03/…
    • 从技术上讲,这仍然看起来像 order n log(n) 而不是 order n。您必须以足够的精度表示变量“总和”,以便可以将 1 阶(或实际上是 1/n 阶)的项添加到 n 阶的总和中,这需要 log n 位。这在某种程度上是不可避免的,无论您使用什么算法:只是为了记录输出,以足够的精度区分序列中的相邻项,需要空间顺序 n log(n)。
    • 啊抱歉,我没有充分考虑除以sum 的效果(我将删除我之前的cmets)。感谢@JamesMartin 的链接,我不熟悉这个结果。最初的问题表明doubles 对于 O.P. 的需求来说足够精确,对于 log10 ndouble 有超过 log2 n 尾数位。因此,这可能是公认的答案,因为它比排序快 log2 n 倍,而且非常简单。此外,样本生成可以并行化,前缀和计算也可以并行化。
    【解决方案4】:

    有一个简单的观察,涉及 [0, 1] 中的有序均匀随机数:

    1. 每个统一的 [0, 1] 数字同样可能小于一半或大于一半。因此,小于一半与大于一半的均匀 [0, 1] 数的数量遵循二项式 (n, 1/2) 分布。
    2. 在小于一半的数字中,每个数字都可能小于 1/4 和大于 1/4,因此小于 1/4 与大于 1 /4 数字遵循相同的分布。
    3. 等等。

    因此,每个数字可以一次生成一位,在二进制点之后从左到右。下面是如何生成 n 个排序的统一随机数的草图:

    1. 如果 n 为 0 或 1,则停止。否则,生成 b,一个二项式 (n, 1/2) 随机数。
    2. 将 0 附加到第一个 b 个随机数,将 1 附加到其余随机数。
    3. 对前 b 个数字递归运行此算法,但 n = b
    4. 对其余数字递归运行此算法,但 n = n - b

    此时,我们有一个随机数的排序列表,其中包含不同的位数。剩下要做的就是根据需要用统一的随机位填充每个数字(或砍掉或舍入多余的位)以给出数字 p 位(例如,53 位用于双精度)。然后,将每个数字除以 2p

    我给出一个similar algorithm 来找到 n 个随机数中最小的 k 个。

    【讨论】:

    • 很好的观察和解决方案!样本也可以在单个长度为 n 的数组中就地生成。看起来它平均需要 log2(n) 递归调用才能达到基本情况,所以我希望平均时间复杂度为 O(n log(n)),这在渐近上并不比生成 n 个均匀样本好并对它们进行排序。你检查过时间复杂度吗?
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2021-12-28
    • 1970-01-01
    相关资源
    最近更新 更多