在数学上，斐波那契数列是以递归的方法来定义
……续上回Fibonacci数列高效解法大全及时间复杂度分析 连载【8】
在家用电脑、手机都是一堆核心，数框框的今日
自然会想到的一项提速手段是——并行运算
就以 Fibonacci数列高效解法大全及时间复杂度分析 连载【7】中第14节的“生成数列的GMP内置fib2函数解法”来说
生成从0至n的斐波那契数列是一个数算完放入结果列表，再算下一个数
这完全可以用我们的CPU多核同时算几个数，四核的话同时算四个数多好多快，完美
这就是分布在多核的多进程并行算法
四核下会是单核的4倍速吗？也就是加速比能到达4吗？
以四核为例，当实现加速比为4，那么并行算法效率就为100%
呃，那是理想状态，实际远远到不了
到不了的原因在于变成多核多进程，一个大任务要在主进程分解成几个子任务（子进程）分配到各核运算，子任务算完后再汇总回主进程，这就有了附加开销。包括任务分解、任务调度、传参、子进程上下文切换、进程间传递数据、进程间同步锁、结果合并等，这些开销很大。让效率下降不少
就生成斐波那契数列而言，实现其并行算法，中间开销最大的应该是进程间传递数据，子进程要返回主进程超大规模的数据
那么对于单机这情况，都知道通常是共享内存方式最快
可以套用时髦的概念“内存计算”
那么下面我来实现一下，共享内存并行算法
17. 共享内存并行解法
import time
import multiprocessing as mp
import ctypes
import array
import gmpy2
def fib2_processes(shared_array_for_fib_seq_results: ‘array ctypes.c_c_ulonglong’, element_length_record: ‘array ctypes.c_ulonglong’, write_protection_pointer: ‘ctypes.c_longlong’, unread_quantity: ‘ctypes.c_ulonglong’, shared_array_overflow: ‘ctypes.c_bool’, mutexlock: ‘Lock’, n_iterable: ‘iterable’) -> None:
start_time = time.process_time_ns()
Fib_seq_array_size = len(shared_array_for_fib_seq_results)
sizeof_c_ulonglong = ctypes.sizeof(ctypes.c_ulonglong)
write_pointer = 0
for n in n_iterable:
for element in gmpy2.fib2(n):
element_bytes = gmpy2.to_binary(element)[2:]
element_byte_length = len(element_bytes)
element_length = (element_byte_length >> 3) + (0 if element_byte_length & 0b111 == 0 else 1)
next_write_pointer = write_pointer + element_length
while shared_array_overflow.value and (next_write_pointer > write_protection_pointer.value):
pass #在溢出状态未消除情况下，如果要写入的范围高于写保护，就空转等待。
if next_write_pointer > Fib_seq_array_size:
shared_array_overflow.value = True
write_pointer = 0
next_write_pointer = element_length
element_length_record[n] = element_length
with mutexlock:
shared_array_for_fib_seq_results[write_pointer: next_write_pointer] = array.array(‘Q’, element_bytes + (b’\x00′ * (element_length * sizeof_c_ulonglong – element_byte_length)))
unread_quantity.value += 1
write_pointer = next_write_pointer
n -= 1
end_time = time.process_time_ns()
print (‘计算和序列化用时 %.1f seconds.’ % ((end_time – start_time)/10**9))
return None
def Fibonacci_sequence_21 (n: int) -> list: #参数n是表示求n项Fibonacci数列
‘多进程共享内存的返回列表的GMP内置fib函数解法’
assert isinstance(n, int), ‘n is an error of non-integer type.’
if n>=0:
start_time = time.process_time_ns()
Number_of_processes = max(mp.cpu_count() – 1, 1)
size_of_shared_variables_for_fib_processes = 1 * (1024 ** 2) // ctypes.sizeof(ctypes.c_ulonglong) // Number_of_processes #就是占用的内存数量，1的单位MB。注意太大的n也要随之加大这个尺寸
list_of_shared_array_for_fib_seq_results = []
list_of_element_length_record = []
list_of_write_protection_pointer = []
list_of_unread_quantity = []
list_of_shared_array_overflow = []
list_of_mutexlock = []
element_length_record = mp.RawArray(ctypes.c_ulonglong, n + 1)
for i in range(Number_of_processes):
list_of_shared_array_for_fib_seq_results.append(mp.RawArray(ctypes.c_ulonglong, size_of_shared_variables_for_fib_processes))
list_of_element_length_record.append(element_length_record)
list_of_write_protection_pointer.append(mp.RawValue(ctypes.c_longlong, -1))
list_of_unread_quantity.append(mp.RawValue(ctypes.c_ulonglong, 0))
list_of_shared_array_overflow.append(mp.RawValue(ctypes.c_bool, False))
list_of_mutexlock.append(mp.Lock())
list_of_n_iterable_for_fib2_processes = [range(n – i * 2, 0, -(Number_of_processes * 2)) for i in range(Number_of_processes)]
fib2_process_list = [None] * Number_of_processes
for i in range(Number_of_processes):
fib2_process_list[i] = mp.Process(target = fib2_processes, args = (list_of_shared_array_for_fib_seq_results[i], list_of_element_length_record[i], list_of_write_protection_pointer[i], list_of_unread_quantity[i], list_of_shared_array_overflow[i], list_of_mutexlock[i], list_of_n_iterable_for_fib2_processes[i]))
fib2_process_list[i].start()
fib_list = [None] * (n + 1)
n_list_for_fib2_processes = []
for n_iterable in list_of_n_iterable_for_fib2_processes:
n_list_for_fib2_processes.append(list(n_iterable))
mv_list_of_results = []
for i in range(Number_of_processes):
mv_list_of_results.append(memoryview(list_of_shared_array_for_fib_seq_results[i]))
list_of_pointers_to_n_list = [0] * Number_of_processes
list_of_number_of_reciprocating = [0] * Number_of_processes
list_of_read_pointer = [0] * Number_of_processes
while True:
if mp.active_children() == []: #检测 当全部子进程都不是活的以后，进行下一个检测准备结束循环
for unread_quantity in list_of_unread_quantity:
if unread_quantity.value != 0: break
else: #检测当所有进程结果的未读数都等于零就结束循环
break
for i in range(Number_of_processes):
if list_of_unread_quantity[i].value != 0:
n_for_fib2_processes = n_list_for_fib2_processes[i][list_of_pointers_to_n_list[i]] – list_of_number_of_reciprocating[i]
next_read_pointer = list_of_read_pointer[i] + list_of_element_length_record[i][n_for_fib2_processes]
if next_read_pointer > size_of_shared_variables_for_fib_processes:
list_of_read_pointer[i] = 0
next_read_pointer = list_of_element_length_record[i][n_for_fib2_processes]
list_of_shared_array_overflow[i].value = False
fib_list[n_for_fib2_processes] = int.from_bytes(mv_list_of_results[i][list_of_read_pointer[i]: next_read_pointer], ‘little’)
with list_of_mutexlock[i]:
list_of_unread_quantity[i].value -= 1
list_of_read_pointer[i] = next_read_pointer
list_of_write_protection_pointer[i].value = next_read_pointer
list_of_pointers_to_n_list[i] += list_of_number_of_reciprocating[i]
list_of_number_of_reciprocating[i] ^= 1
if n & 1 == 0:
fib_list[0] = 0
end_time = time.process_time_ns()
print (‘主进程用时 %.1f seconds.’ % ((end_time – start_time)/10**9))
return fib_list
else:
return None
if __name__ == ‘__main__’:
start_time = time.perf_counter()
Fibonacci_sequence_21(1000000)
end_time = time.perf_counter()
print (‘最终用时 %.1f seconds.’ % (end_time – start_time))
在我的四核16GB物理内存电脑上算100万斐波那契数列（数据总占用内存40GB，所以包括虚拟内存）用时为：
1134秒
在之前14节单进程解法算100万斐波那契数列用时为：
1857秒
缩短用时38.9%
然而这是4核并行，并行效率为40.9%（(1857/1134)/4=0.409）。效率一般的并行算法
各位大佬，以14节那个单进程解法为基础，来尝试写出你们的高效并行解法，在下方留言吧
结尾，嘿，欢迎点赞和赞赏支持
未完待续……
