【问题标题】：How to generate random prime numbers using parallel processing in Python?如何在 Python 中使用并行处理生成随机素数？
【发布时间】：2021-07-26 05:59:14
【问题描述】：

我在 python 中实现 RSA，最初我使用 Python Multiprocessing 包的并行处理生成素数随机数。我正在使用米勒拉宾算法来检查素数。这个并行实现应该运行得更快，但我不明白为什么不是这样。并行处理需要花费大量时间，即使对于较小的素数大小（如 50 位）也是如此。任何人都可以帮助解决这个问题吗？（P.S. 我是并行处理的初学者）

这是我的python代码（实现函数+RSA算法实现）

所有必需的函数：

# Prime Generation using Parallel Processing (Miller Rabin)
import time
from multiprocessing import Pool
import random

# Prime-Generating-Functions start---
from random import randrange, getrandbits
def is_prime(n, k=128):
    # Test if n is not even.
    # But care, 2 is prime !
    if n == 2 or n == 3:
        return True
    if n <= 1 or n % 2 == 0:
        return False
    # find r and s
    s = 0
    r = n - 1
    while r & 1 == 0:
        s += 1
        r //= 2
    flag=1
    p=Pool()
    results = [p.apply_async(primetest, args=(s,r,n,)) for x in range(k)]
    for k in results:
        output = [k.get()]
        #print("VALUE OF OUTPUT",k.get())
        if k.get()==False:
            flag=0
            break
    if flag==0:
        p.close()
        p.join()
        return False
    else:
        p.close()
        p.join()
        return True

def primetest(s,r,n):
    #for _ in range(k):
    #print("Test.value in primetest", test.value)
    a = randrange(2, n - 1)
    x = pow(a, r, n)
    if x != 1 and x != n - 1:
        j = 1
        while j < s and x != n - 1:
            x = pow(x, 2, n)
            if x == 1:
                return False
            j += 1
        if x != n - 1:
            return False
    return True

def generate_prime_candidate(length):
    # generate random bits
    p = getrandbits(length)
    # apply a mask to set MSB and LSB to 1
    p |= (1 << length - 1) | 1
    return p

def generate_prime_number(length=1024):
    p = 4
    # keep generating while the primality test fail
    while not is_prime(p, 128):
        p = generate_prime_candidate(length)
        #print("value of p is",p)
    return p
# Prime-Generating-Functions end---


# Miscellaneous Functions
def gcd(a, b):
    while b != 0:
        a, b = b, a % b
    return a

def modInverse(a, m):
    m0 = m
    y = 0
    x = 1
    if (m == 1):
        return 0
    while (a > 1):
        # q is quotient
        q = a // m
        t = m
        # m is remainder now, process
        # same as Euclid's algo
        m = a % m
        a = t
        t = y
        # Update x and y
        y = x - q * y
        x = t
    # Make x positive
    if (x < 0):
        x = x + m0
    return x

RSA 算法从这里开始：

import time    
#key gen start----------------------------------------
start_key_gen_time = time.time()
# print("Key Generation:\n");
p = generate_prime_number(50) #Takes the size (eg 50) in bits, and generates random prime of that size
q = generate_prime_number(50)
# print("Is p=",p, "prime ? ", is_prime(p));
# print("Is q=",q,"prime ?",is_prime(q));

n = p*q; # evaluating n
# print("\nn=",n);

phi = (p-1)*(q-1); # evaluating phi
# print("phi=", phi)

e = ZZ.random_element(phi)
while (gcd(e, phi) != 1):
    e = ZZ.random_element(phi) #choosing e
# print("\ne=",e);
# print("*e < phi=",e < phi) #checking conditions for e
# print("*gcd(e,phi)=",gcd(e,phi))
# print("*Is e prime ?", is_prime(e))

d = inverse_mod(e, phi) #finding corresponding d, s.t. e*d = 1*mod(phi)
# print("\nd=",d);
# print("*Is d prime ?", is_prime(d))

# print("\npublic key = (",n,", ",e,")"); #Generated Key pairs
# print("private key = (",n,", ",d,")\n");
final_key_gen_time = time.time() 
#key gen finish------------------------------------------------

total_key_gen_time = final_key_gen_time - start_key_gen_time
print("\nTotal Key generation time taken in seconds: ", total_key_gen_time )


#Encryption start--------------------------------------------
start_encrypt_time = time.time()
# print("Encryption & Decryption:\n"); 
m = 59; #print("original msg=",m) #original message m
c = power_mod(m,e,n); #print("encrypted msg=",c) #encrypted message c
finish_encrypt_time = time.time()
#Encryption finish---------------------------------------------

total_encrypt_time = finish_encrypt_time - start_encrypt_time
print("\nTotal encrypt taken in seconds: ", total_encrypt_time )

#Decryption start----------------------------------------------
start_decrypt_time = time.time() 
decrypt_msg = power_mod(c,d,n) # decrypting 
# print("decrypted msg=",decrypt_msg)# so m was correctly decrypted
finish_decrypt_time = time.time()
#Decryption finish----------------------------------------------

total_decrypt_time = finish_decrypt_time - start_decrypt_time
print("\nTotal decrypt time taken in seconds: ", total_decrypt_time )
    
total_time_taken = total_key_gen_time + total_encrypt_time + total_decrypt_time
print("\n\nTotal algorithm time taken in seconds: ", total_time_taken)
print("\nIs decrypted msg & original msg same?", (decrypt_msg==m))

【问题讨论】：

Stackoverflow 是一个问答系统，因此您的帖子应该包含一个问题。请编辑您的帖子并包含一个有人可以回答的真实问题。
我还没有看过代码，但是多处理涉及到每个创建的进程都有一定数量的固定开销，所以进程越长，开销就越不明显。 50 位素数非常小，开销将使运行素数查找算法所需的时间相形见绌。尝试使用 2048 位素数进行测试，以获得更真实的测试。
感谢@Robert，我已根据您的建议更新了帖子。
@PresidentJamesK.Polk ，我也尝试使用 2048 位素数对其进行测试。但现在，我没有得到任何结果，因为与 50 位相比，时间非常高。
你之前问过这个问题，而我之前的评论是成立的。您在并行 Miller-Rabin (MR) 中所做的工作比在串行版本中所做的要多得多，因为您测试的 几乎每个 数字 n 都是复合的，并且会失败 您的 128 个 MR 测试中的第一个，使其他 127 个不必要。是的，您对此进行检查并将结果发送给其他进程，但它们不太可能在开始工作之前收到此信号。

标签： python parallel-processing cryptography rsa primes

【解决方案1】：

最佳解决方案

我不确定在这种情况下我是否理解 Time complexity: O(sqrt(n)) 和 Space complexity: O(1) 的概念，但是函数prime(n) 可能是fastest way (least iterations) 计算一个数是否是任意大小的素数。

这可能是截至今天 11 日的互联网上最好的解决方案 2022 年 3 月。欢迎提供反馈和使用。

同样的代码可以应用于任何语言，如 C、C++、Go Lang、 Java、.NET、Python、Rust 等具有相同的逻辑并具有性能好处。它非常快。我以前没有见过这个实现并且已经在本土完成了。

这也可以帮助您避免并行处理。这大大减少了迭代次数。

如果您在这里查看速度和性能，我可以提供"""BEST""" 有希望的解决方案：

n == 100000 的最大迭代次数为 16666，而不是传统的 100000 方式

function prime(n) {
    
    if ((n === 2 || n === 3 || n === 5 || n === 7)) {
        return true
    }
    if (n === 1 || ((n > 7) && (n % 5 == 0 || n % 7 == 0 || n % 2 == 0 || n % 3 == 0))) {
        return false
    }
    if ((Number.isInteger(((n - 1) / 6))) || (Number.isInteger((n + 1) / 6))) {
        for (let i = 1; i < n; i++) {

            let five = (5 + (i * 6)), seven = (7 + (i * 6))
            if ((((n / five) > 1) && Number.isInteger((n / five))) || (((n / seven) > 1) && (Number.isInteger(n / seven)))) {
                return false;
            }
            if ((i * 6) > n) {
                break;
            }
        }
        return true
    }
    return false
}

下面是对所有计算方式的分析：

检查素数的常规方法：

def isPrimeConventionalWay(n):
    count = 0
    if (n <= 1):
        return False;
    # Check from 2 to n-1
    # Max iterations 99998 for n == 100000 
    for i in range(2,n):
        # Counting Iterations
        count += 1
        if (n % i == 0):
            print("count: Prime Conventional way", count)
            return False;
    print("count: Prime Conventional way", count)
    return True;

检查素数的 SQUAREROOT 方法：

def isPrimeSquarerootWay(num):
    count = 0
    # if not is_number num return False
    if (num < 2):
        print("count: Prime Squareroot way", count)
        return False
    
    s = math.sqrt(num)
    for  i in range(2, num):
        # Counting Iterations
        count += 1
        if (num % i == 0):
            print("count: Prime Squareroot way", count)
            return False
    print("count: Prime Squareroot way", count)
    return True

其他方式：

def isprimeAKSWay(n):
    """Returns True if n is prime."""
    count = 0
    if n == 2:
        return True
    if n == 3:
        return True
    if n % 2 == 0:
        return False
    if n % 3 == 0:
        return False

    i = 5
    w = 2

    while i * i <= n:
        count += 1
        if n % i == 0:
            print("count: Prime AKS - Mersenne primes - Fermat's little theorem or whatever way", count)
            return False

        i += w
        w = 6 - w
    print("count: Prime AKS - Mersenne primes - Fermat's little theorem or whatever way", count)
    return True

检查素数的建议方法：

def prime(n):
    count = 0
    if ((n == 2 or n == 3 or n == 5 or n == 7)):
        print("count: Prime Unconventional way", count)
        return True
    
    if (n == 1 or ((n > 7) and (n % 5 == 0 or n % 7 == 0 or n % 2 == 0 or n % 3 == 0))):
        print("count: Prime Unconventional way", count)
        return False
    
    if (((n - 1) / 6).is_integer()) or (((n + 1) / 6).is_integer()):
        for i in range(1, n):
            # Counting Iterations
            count += 1
            five = 5 + (i * 6)
            seven = 7 + (i * 6)
            if ((((n / five) > 1) and (n / five).is_integer()) or (((n / seven) > 1) and ((n / seven).is_integer()))):
                print("count: Prime Unconventional way", count)
                return False;
            
            if ((i * 6) > n):
                # Max iterations 16666 for n == 100000 instead of 100000
                break;
            
        print("count: Prime Unconventional way", count)
        return True
    
    print("count: Prime Unconventional way", count)
    return False

与检查素数的传统方法进行比较的测试。

def test_primecalculations():
    count = 0
    iterations = 100000
    arr = []
    for i in range(1, iterations):
        traditional = isPrimeConventionalWay(i)
        newer = prime(i)
        if (traditional == newer):
            count = count + 1
        else:
            arr.push([traditional, newer, i])
    print("[count, iterations, arr] list: ", count, iterations, arr)
    if (count == iterations):
        return True
    return False


# print("Tests Passed: ", test_primecalculations())

您将看到check of prime number: 100007 的迭代次数计数结果如下：

print("Is Prime 100007: ", isPrimeConventionalWay(100007))
print("Is Prime 100007: ", isPrimeSquarerootWay(100007))
print("Is Prime 100007: ", prime(100007))
print("Is Prime 100007: ", isprimeAKSWay(100007))

count: Prime Conventional way 96
Is Prime 100007:  False
count: Prime Squareroot way 96
Is Prime 100007:  False
count: Prime Unconventional way 15
Is Prime 100007:  False
count: Prime AKS - Mersenne primes - Fermat's little theorem or whatever way 32
Is Prime 100007:  False

【讨论】：