小赖子的英国生活和资讯

使用 Python 和 C++ 数值积分计算圆周率 Pi

阅读 桌面完整版

本文介绍如何利用定积分 近似计算圆周率 Pi。文章先解释该积分为什么等于 Pi,然后介绍梯形法则和中点法则的数学原理,并给出 Python 与 C++ 的单线程、多线程实现。最后进一步说明如何将积分区间划分到五个分布式计算节点上,并比较不同实现方式的精度、性能、并行能力和适用场景。

integral-math-pi 使用 Python 和 C++ 数值积分计算圆周率 Pi 学习笔记 数学 数学 计算机

定积分计算PI

圆周率 Pi,通常写作 ,广泛出现在数学、物理、工程、统计学和计算机科学中。虽然大多数编程语言都已经提供了内置的 Pi 常量,但亲自计算 Pi 仍然是学习微积分、数值积分、多线程和分布式计算的一个很好的学习的例子。

本文将通过下面这个定积分计算 Pi:

本文将介绍:

为什么这个积分等于 Pi?

考虑下面的定积分:

关键在于,反正切函数的导数是:

因此:

将积分上下限零和一代入:

也就是:

我们知道:

并且:

因此:

所以:

这就为我们提供了一种通过数值积分计算 Pi 的方法。

数值积分是如何工作的?

计算机不一定要通过符号运算直接求出积分的解析解。它可以把积分区间分成许多很小的部分,然后近似计算曲线下面的面积。

假设我们需要计算:

将区间 分成 个长度相等的小区间。

每个小区间的宽度为:

划分区间的各个点为:

其中:

近似计算每一个小区域的面积有多种方法。两种常见的方法是:

梯形法则

你可能记得的那个包含 和除以二的公式,就是梯形法则。

对于从 的一个小区间,我们使用一条直线连接函数在两个端点上的值。

这样形成的区域是一个梯形。

它的面积近似为:

将所有梯形的面积相加,可以得到:

一个等价并且通常更容易编程实现的公式是:

两个端点只计算一半的权重:

所有内部点使用完整权重:

对于我们的 Pi 积分:

并且:

因此:

梯形法则对 Pi 的近似公式为:

因为:

并且:

所以还可以写成:

一个简单的梯形法则示例

假设我们把积分区间分成四份:

每个小区间的宽度为:

五个边界点为:

函数在这些点上的值大约为:

应用梯形法则:

因此:

Pi 的真实值大约为:

即使只划分为四个小区间,我们也已经得到了一个比较合理的近似值。增大 可以进一步提高结果的精度。

中点法则

本文后面的主要代码使用的是中点法则。

梯形法则在每个小区间的边界上计算函数值,而中点法则在每个小区间的中心位置计算函数值。

个区间的中点为:

该小区间的面积可以近似看成一个矩形:

将所有矩形面积相加:

对于 Pi 的积分,,并且

因此:

并且:

代入

越大,计算结果通常越接近 Pi 的真实值。

梯形法则与中点法则的比较

方法 函数取值位置 公式
梯形法则 每个区间的边界
中点法则 每个区间的中心

对于足够平滑的函数,这两种方法的误差通常都与下面的量成正比:

不过,在使用相同数量小区间的情况下,中点法则通常比梯形法则更加精确。

使用中点法则的单线程 Python 实现

下面的 Python 程序将积分区间划分成 2500 万个小区间。

import math
import time


def calculate_pi(total_steps: int) -> float:
    if total_steps <= 0:
        raise ValueError("total_steps 必须为正数")

    step_width = 1.0 / total_steps
    total = 0.0

    for step in range(total_steps):
        midpoint = (step + 0.5) * step_width
        total += 4.0 / (1.0 + midpoint * midpoint)

    return total * step_width


def main() -> None:
    total_steps = 25_000_000

    started_at = time.perf_counter()
    pi_estimate = calculate_pi(total_steps)
    elapsed_seconds = time.perf_counter() - started_at

    absolute_error = abs(pi_estimate - math.pi)

    print(f"Pi 估算值:    {pi_estimate:.15f}")
    print(f"Pi 参考值:    {math.pi:.15f}")
    print(f"绝对误差:     {absolute_error:.3e}")
    print(f"运行时间:     {elapsed_seconds:.3f} 秒")


if __name__ == "__main__":
    main()

核心计算代码是:

midpoint = (step + 0.5) * step_width
total += 4.0 / (1.0 + midpoint * midpoint)

第一行计算当前小区间的中点:

第二行计算函数值:

在累加所有函数值之后,再乘以每个小区间的宽度:

return total * step_width

这在数学上对应:

使用梯形法则的单线程 Python 实现

梯形法则版本在各个小区间的边界位置计算函数值,而不是在中点计算。

import math
import time


def calculate_pi_trapezoidal(total_steps: int) -> float:
    if total_steps <= 0:
        raise ValueError("total_steps 必须为正数")

    h = 1.0 / total_steps

    def f(x: float) -> float:
        return 4.0 / (1.0 + x * x)

    total = 0.5 * (f(0.0) + f(1.0))

    for step in range(1, total_steps):
        x = step * h
        total += f(x)

    return total * h


def main() -> None:
    total_steps = 25_000_000

    started_at = time.perf_counter()
    pi_estimate = calculate_pi_trapezoidal(total_steps)
    elapsed_seconds = time.perf_counter() - started_at

    absolute_error = abs(pi_estimate - math.pi)

    print(f"Pi 估算值:    {pi_estimate:.15f}")
    print(f"Pi 参考值:    {math.pi:.15f}")
    print(f"绝对误差:     {absolute_error:.3e}")
    print(f"运行时间:     {elapsed_seconds:.3f} 秒")


if __name__ == "__main__":
    main()

两个端点的贡献通过下面的代码计算:

total = 0.5 * (f(0.0) + f(1.0))

这对应:

循环计算所有内部点的函数值之和:

for step in range(1, total_steps):
    x = step * h
    total += f(x)

这对应:

最后再乘以

return total * h

使用中点法则的多线程 Python 实现

我们可以把 2500 万个小区间分配给五个工作线程。

对于总共 个工作线程中的第 个工作线程,它的起始位置为:

结束位置为:

对于五个工作线程和 2500 万个小区间:

工作线程 起始步骤 结束步骤 步骤数量
0 0 5,000,000 5,000,000
1 5,000,000 10,000,000 5,000,000
2 10,000,000 15,000,000 5,000,000
3 15,000,000 20,000,000 5,000,000
4 20,000,000 25,000,000 5,000,000

下面使用 ThreadPoolExecutor 实现:

import math
import time
from concurrent.futures import ThreadPoolExecutor


def calculate_partial_sum(
    start_step: int,
    end_step: int,
    step_width: float,
) -> float:
    partial_sum = 0.0

    for step in range(start_step, end_step):
        midpoint = (step + 0.5) * step_width
        partial_sum += 4.0 / (1.0 + midpoint * midpoint)

    return partial_sum


def calculate_pi_multithreaded(
    total_steps: int,
    worker_count: int,
) -> float:
    if total_steps <= 0:
        raise ValueError("total_steps 必须为正数")

    if worker_count <= 0:
        raise ValueError("worker_count 必须为正数")

    step_width = 1.0 / total_steps
    ranges = []

    for worker_id in range(worker_count):
        start_step = total_steps * worker_id // worker_count
        end_step = total_steps * (worker_id + 1) // worker_count
        ranges.append((start_step, end_step))

    with ThreadPoolExecutor(max_workers=worker_count) as executor:
        futures = [
            executor.submit(
                calculate_partial_sum,
                start_step,
                end_step,
                step_width,
            )
            for start_step, end_step in ranges
        ]

        partial_sums = [
            future.result()
            for future in futures
        ]

    return sum(partial_sums) * step_width


def main() -> None:
    total_steps = 25_000_000
    worker_count = 5

    started_at = time.perf_counter()

    pi_estimate = calculate_pi_multithreaded(
        total_steps,
        worker_count,
    )

    elapsed_seconds = time.perf_counter() - started_at
    absolute_error = abs(pi_estimate - math.pi)

    print(f"工作线程数:   {worker_count}")
    print(f"Pi 估算值:    {pi_estimate:.15f}")
    print(f"Pi 参考值:    {math.pi:.15f}")
    print(f"绝对误差:     {absolute_error:.3e}")
    print(f"运行时间:     {elapsed_seconds:.3f} 秒")


if __name__ == "__main__":
    main()

每个工作线程计算一个部分和:

主线程再把这些结果合并起来:

Python 多线程的一个重要限制

这个多线程 Python 版本展示了如何把计算划分为多个独立区间,但它不一定比单线程版本运行得更快。

标准 CPython 存在全局解释器锁,也就是通常所说的 GIL。

对于 CPU 密集型 Python 代码,在同一时刻通常只有一个线程能够执行 Python 字节码。

因此,五个 Python 线程并不意味着能够有效地同时使用五个 CPU 核心。

这个多线程版本可能会出现以下情况:

如果需要在 Python 中实现真正的 CPU 并行,可以考虑:

使用中点法则的单线程 C++ 实现

C++ 非常适合执行这种紧密的数值循环,因为它可以被编译为本地机器代码/Native Code。

#include <chrono>
#include <cmath>
#include <cstdint>
#include <iomanip>
#include <iostream>
#include <stdexcept>


double calculate_pi(std::uint64_t total_steps) {
    if (total_steps == 0) {
        throw std::invalid_argument(
            "total_steps 必须为正数"
        );
    }

    const double step_width =
        1.0 / static_cast<double>(total_steps);

    double total = 0.0;

    for (
        std::uint64_t step = 0;
        step < total_steps;
        ++step
    ) {
        const double midpoint =
            (static_cast<double>(step) + 0.5)
            * step_width;

        total += 4.0 / (1.0 + midpoint * midpoint);
    }

    return total * step_width;
}


int main() {
    constexpr std::uint64_t total_steps = 25'000'000;

    const auto started_at =
        std::chrono::steady_clock::now();

    const double pi_estimate =
        calculate_pi(total_steps);

    const auto finished_at =
        std::chrono::steady_clock::now();

    const double elapsed_seconds =
        std::chrono::duration<double>(
            finished_at - started_at
        ).count();

    const double reference_pi = std::acos(-1.0);

    const double absolute_error =
        std::abs(pi_estimate - reference_pi);

    std::cout
        << std::fixed
        << std::setprecision(15);

    std::cout
        << "Pi 估算值:    "
        << pi_estimate
        << '\n';

    std::cout
        << "Pi 参考值:    "
        << reference_pi
        << '\n';

    std::cout
        << std::scientific
        << std::setprecision(3);

    std::cout
        << "绝对误差:     "
        << absolute_error
        << '\n';

    std::cout
        << std::fixed
        << std::setprecision(3);

    std::cout
        << "运行时间:     "
        << elapsed_seconds
        << " 秒\n";

    return 0;
}

使用优化选项进行编译:

g++ -O3 -std=c++17 pi_single.cpp -o pi_single

运行程序:

./pi_single

-O3 选项会启用较为激进的编译器优化。如果不启用优化,程序的运行速度可能会慢很多。

使用梯形法则的单线程 C++ 实现

下面的 C++ 程序实现了梯形法则:

#include <chrono>
#include <cmath>
#include <cstdint>
#include <iomanip>
#include <iostream>
#include <stdexcept>


double f(double x) {
    return 4.0 / (1.0 + x * x);
}


double calculate_pi_trapezoidal(
    std::uint64_t total_steps
) {
    if (total_steps == 0) {
        throw std::invalid_argument(
            "total_steps 必须为正数"
        );
    }

    const double h =
        1.0 / static_cast<double>(total_steps);

    double total =
        0.5 * (f(0.0) + f(1.0));

    for (
        std::uint64_t step = 1;
        step < total_steps;
        ++step
    ) {
        const double x =
            static_cast<double>(step) * h;

        total += f(x);
    }

    return total * h;
}


int main() {
    constexpr std::uint64_t total_steps = 25'000'000;

    const auto started_at =
        std::chrono::steady_clock::now();

    const double pi_estimate =
        calculate_pi_trapezoidal(total_steps);

    const auto finished_at =
        std::chrono::steady_clock::now();

    const double elapsed_seconds =
        std::chrono::duration<double>(
            finished_at - started_at
        ).count();

    const double reference_pi = std::acos(-1.0);

    const double absolute_error =
        std::abs(pi_estimate - reference_pi);

    std::cout
        << std::fixed
        << std::setprecision(15);

    std::cout
        << "Pi 估算值:    "
        << pi_estimate
        << '\n';

    std::cout
        << "Pi 参考值:    "
        << reference_pi
        << '\n';

    std::cout
        << std::scientific
        << std::setprecision(3);

    std::cout
        << "绝对误差:     "
        << absolute_error
        << '\n';

    std::cout
        << std::fixed
        << std::setprecision(3);

    std::cout
        << "运行时间:     "
        << elapsed_seconds
        << " 秒\n";

    return 0;
}

使用下面的命令编译:

g++ -O3 -std=c++17 pi_trapezoidal.cpp -o pi_trapezoidal

使用中点法则的多线程 C++ 实现

与普通 CPython 线程不同,C++ 线程可以在多个 CPU 核心上同时执行 CPU 密集型计算。

下面的实现将计算任务分配给五个线程。

#include <chrono>
#include <cmath>
#include <cstdint>
#include <iomanip>
#include <iostream>
#include <numeric>
#include <stdexcept>
#include <thread>
#include <vector>


double calculate_partial_sum(
    std::uint64_t start_step,
    std::uint64_t end_step,
    double step_width
) {
    double partial_sum = 0.0;

    for (
        std::uint64_t step = start_step;
        step < end_step;
        ++step
    ) {
        const double midpoint =
            (static_cast<double>(step) + 0.5)
            * step_width;

        partial_sum +=
            4.0 / (1.0 + midpoint * midpoint);
    }

    return partial_sum;
}


double calculate_pi_multithreaded(
    std::uint64_t total_steps,
    std::size_t worker_count
) {
    if (total_steps == 0) {
        throw std::invalid_argument(
            "total_steps 必须为正数"
        );
    }

    if (worker_count == 0) {
        throw std::invalid_argument(
            "worker_count 必须为正数"
        );
    }

    const double step_width =
        1.0 / static_cast<double>(total_steps);

    std::vector<std::thread> workers;

    std::vector<double> partial_sums(
        worker_count,
        0.0
    );

    workers.reserve(worker_count);

    for (
        std::size_t worker_id = 0;
        worker_id < worker_count;
        ++worker_id
    ) {
        const std::uint64_t start_step =
            total_steps
            * worker_id
            / worker_count;

        const std::uint64_t end_step =
            total_steps
            * (worker_id + 1)
            / worker_count;

        workers.emplace_back(
            [
                &,
                worker_id,
                start_step,
                end_step
            ]() {
                partial_sums[worker_id] =
                    calculate_partial_sum(
                        start_step,
                        end_step,
                        step_width
                    );
            }
        );
    }

    for (std::thread& worker : workers) {
        worker.join();
    }

    const double combined_sum =
        std::accumulate(
            partial_sums.begin(),
            partial_sums.end(),
            0.0
        );

    return combined_sum * step_width;
}


int main() {
    constexpr std::uint64_t total_steps = 25'000'000;
    constexpr std::size_t worker_count = 5;

    const auto started_at =
        std::chrono::steady_clock::now();

    const double pi_estimate =
        calculate_pi_multithreaded(
            total_steps,
            worker_count
        );

    const auto finished_at =
        std::chrono::steady_clock::now();

    const double elapsed_seconds =
        std::chrono::duration<double>(
            finished_at - started_at
        ).count();

    const double reference_pi = std::acos(-1.0);

    const double absolute_error =
        std::abs(pi_estimate - reference_pi);

    std::cout
        << "工作线程数:   "
        << worker_count
        << '\n';

    std::cout
        << std::fixed
        << std::setprecision(15);

    std::cout
        << "Pi 估算值:    "
        << pi_estimate
        << '\n';

    std::cout
        << "Pi 参考值:    "
        << reference_pi
        << '\n';

    std::cout
        << std::scientific
        << std::setprecision(3);

    std::cout
        << "绝对误差:     "
        << absolute_error
        << '\n';

    std::cout
        << std::fixed
        << std::setprecision(3);

    std::cout
        << "运行时间:     "
        << elapsed_seconds
        << " 秒\n";

    return 0;
}

使用线程支持和优化选项进行编译:

g++ -O3 -std=c++17 -pthread pi_multithreaded.cpp -o pi_multithreaded

运行程序:

./pi_multithreaded

为什么每个线程都使用自己的局部求和变量?

每个线程内部首先使用一个局部变量进行计算:

double partial_sum = 0.0;

线程不会在每一次循环中都修改同一个共享的全局变量。

如果所有线程不断更新同一个共享变量,程序就需要使用互斥锁或者原子操作。

这会引入同步开销和线程竞争,可能抵消多线程带来的性能优势。

因此,程序采用了类似 MapReduce 的模式:

  1. 为每个工作线程分配一段积分区间。
  2. 让每个工作线程独立完成计算。
  3. 每个线程保存一个部分结果。
  4. 等待所有线程结束。
  5. 将所有部分结果相加。

在数学上:

最终近似值为:

将算法扩展到五个分布式计算节点

多线程和分布式计算使用相同的数学划分方法,但它们是两种不同的执行模型。

线程通常具有以下特点:

分布式工作进程通常具有以下特点:

在一个五节点任务中,每个进程都会获得一个 Rank:

进程总数为:

每个 Rank 使用下面的公式计算自己的任务起始位置:

结束位置为:

在 Python 中,可以这样分配计算范围:

import os


rank = int(os.environ["RANK"])
world_size = int(os.environ["WORLD_SIZE"])

start_step = total_steps * rank // world_size
end_step = total_steps * (rank + 1) // world_size

每个 Rank 只计算自己负责的区间:

partial_sum = 0.0

for step in range(start_step, end_step):
    midpoint = (step + 0.5) * step_width

    partial_sum += 4.0 / (
        1.0 + midpoint * midpoint
    )

对于五个 Rank,最终结果为:

在前面的 Singularity 示例中,每个 Rank 都会写入一个结果文件:

rank-0.json
rank-1.json
rank-2.json
rank-3.json
rank-4.json

Rank 0 等待全部五个文件生成,然后读取每个文件中的部分和,将它们相加,最终得到 Pi 的估算值。

这个过程实现了三种分布式操作:

基于文件的实现比较简单,也很容易理解,但它假设所有节点都可以访问同一个共享输出目录。

在正式的生产级分布式程序中,通常会使用 MPI 或其他支持集合通信的框架,以获得更加可靠和高效的通信机制。

数值积分方法的精度

对于足够平滑的函数,梯形法则和中点法则的误差通常都与下面的量成正比:

这意味着,当小区间数量加倍时,数值积分误差可能大约缩小为原来的四分之一。

不过,无限增大 并不意味着可以获得无限精度。

计算机使用有限精度的浮点数存储数据。当程序累加几百万甚至几十亿个数值时,舍入误差也可能逐渐累积。

并行版本和单线程版本在最后几位数字上也可能略有不同,因为浮点数加法并不严格满足结合律:

这些表达式在数学上是等价的,但是在有限精度浮点计算中,改变加法顺序可能会改变最终的舍入结果。

性能方面的注意事项

在以下条件下,多线程 C++ 实现通常会比单线程版本更快:

使用五个线程并不保证一定能够获得五倍性能提升。

程序性能还会受到以下因素影响:

对于 Python,GIL 是主要限制。增加 Python 线程数量通常无法加速纯 Python 的 CPU 密集型循环。

对于分布式计算,节点分配、进程启动、共享存储访问以及结果合并也都会产生额外开销。

使用 2500 万个区间计算 Pi 是一个很好的教学示例,但在真实生产环境中,这个计算任务过于简单,不值得为此分配多个昂贵的 GPU 计算节点。

不同实现方式的比较

实现方式 执行模型 真正的 CPU 并行 预期性能
Python 单线程 一个 Python 线程 实现简单,但速度相对较慢
Python 多线程 多个 Python 线程 通常不能,因为受到 GIL 限制 通常不会比单线程更快
C++ 单线程 一个本地线程 通常比纯 Python 快很多
C++ 多线程 多个本地线程 通常是本地计算中最快的版本
五节点分布式计算 多个独立工作进程 可以扩展,但存在启动和通信开销

总结

下面这个恒等式:

为我们提供了一个简单的例子,展示如何把微积分问题转换成计算机可以执行的数值计算问题。

梯形法则使用相邻边界点之间的直线近似曲线:

中点法则使用每个小区间的中心位置计算函数值:

同一个数学计算可以通过多种方式执行:

无论使用哪一种执行模型,背后的数学算法基本不变。

真正发生变化的是:如何划分积分区间、在哪里计算部分和,以及如何将所有部分结果重新合并起来。

这也是很多并行算法背后的核心模式:

任务分割 —> 计算部分结果 —> 合并计算最终结果

虽然计算 Pi 只是一个简单的示例,但同样的 MapReduce 思想广泛应用于科学模拟、机器学习、数据处理、图形渲染、金融建模和大规模分布式系统中。

数学

英文:Calculating Pi with Numerical Integration in Python and C

强烈推荐

微信公众号: 小赖子的英国生活和资讯 JustYYUK

阅读 桌面完整版
Exit mobile version