4.8 并行计算

从20世纪70年代到2000年代中期，单个处理器核心的速度呈指数级增长。大部分速度的提高是通过提高时钟频率来实现的，时钟频率是处理器执行基本操作的频率。然而，在2000年代中期，由于功率和热量的限制，这种指数级增长戛然停止，从那时起，单个处理器核心的速度增长缓慢得多。相反，CPU制造商开始在单个处理器中放置多个核，使更多的操作可以同时执行。

并行并不是一个新概念。大型并行机已经使用了几十年，主要用于科学计算和数据分析。即使在具有单个处理器核心的个人计算机中，操作系统和解释器也提供了并发性的抽象。这是通过上下文切换来实现的，或者是不等任务完成就在不同任务之间快速切换。因此，即使只有一个处理核心，多个程序也可以在同一台机器上同时运行。

鉴于当前处理器内核数量不断增加的趋势，各个应用程序现在必须利用并行性，以便运行得更快。在单个程序中，必须安排计算，以便尽可能多的工作能并行完成。然而，并行性在编写正确的代码方面带来了新的挑战，特别是在共享的可变状态下。

对于在没有共享可变状态的情况下可以在功能模型中有效解决的问题，并行性几乎不会产生问题。纯函数提供了参考透明性，这意味着表达式可以用它们的值替换，反之亦然，而不会影响程序的行为。 这允许并行计算彼此不依赖的表达式。如前一节所讨论的，MapReduce框架允许指定函数式程序，并以最小的程序员工作量并行运行。

不幸的是，并不是所有的问题都可以使用函数式编程有效地解决。 Berkeley View项目已经确定了科学与工程中13种常见的计算模式，其中只有一个MapReduce。 其余模式需要共享状态。

在本节的其余部分中，我们将看到可变共享状态如何将错误引入并行程序，以及防止此类错误的多种方法。 我们将在两个应用程序的背景下研究这些技术，一个网络爬虫和一个粒子模拟器。

4.8.1 Python中的并行性

在深入研究并行性的细节之前，让我们先看看Python对并行计算的支持。 Python提供了两种并行执行方式：线程和多进程。

线程。在线程中，多个“线程”在一个解释器中执行。 每个线程都独立于其他线程执行代码，尽管它们共享相同的数据。 然而，CPython解释器，Python的主要实现，一次只解释一个线程中的代码，在它们之间切换以提供并行的错觉。 另一方面，解释器外部的操作，如写入文件或访问网络，可以并行运行。 threading模块包含允许创建和同步线程的类。 下面是一个多线程程序的简单示例：

>>> import threading
>>> def thread_hello():
        other = threading.Thread(target=thread_say_hello, args=())
        other.start()
        thread_say_hello()

>>> def thread_say_hello():
        print('hello from', threading.current_thread().name)

>>> thread_hello()
hello from Thread-1
hello from MainThread

Thread构造函数创建一个新线程。它需要一个新线程应该运行的目标函数，以及该函数的参数。 在线程对象上调用start将标志它准备好运行。 函数的作用是:返回与当前执行线程相关联的线程对象。

在本例中，打印可以以任何顺序发生，因为我们没有以任何方式同步它们。

多进程。Python还支持多进程，这允许一个程序产生多个解释器或进程，每个解释器或进程都可以独立运行代码。 这些进程通常不共享数据，因此任何共享状态必须在进程之间进行通信。 另一方面，进程根据底层操作系统和硬件提供的并行级别并行执行。 因此，如果CPU有多个处理器核，Python进程可以真正并发运行。

multiprocessing模块包含用于创建和同步进程的类。下面是使用进程的hello示例：

>>> import multiprocessing
>>> def process_hello():
        other = multiprocessing.Process(target=process_say_hello, args=())
        other.start()
        process_say_hello()

>>> def process_say_hello():
        print('hello from', multiprocessing.current_process().name)

>>> process_hello()
hello from MainProcess
>>> hello from Process-1

如本例所示，multiprocessing中的许多类和函数与线程中的类和函数类似。这个例子还演示了缺乏同步是如何影响共享状态的，因为显示可以被视为共享状态。在这里，交互进程的解释器提示出现在另一个进程的打印输出之前。

4.8.2 共享状态问题

为了进一步说明共享状态的问题，让我们看一个在两个线程之间共享的计数器的简单例子：

import threading
from time import sleep

counter = [0]

def increment():
    count = counter[0]
    sleep(0) # try to force a switch to the other thread
    counter[0] = count + 1

other = threading.Thread(target=increment, args=())
other.start()
increment()
print('count is now: ', counter[0])

在这个程序中，两个线程试图增加相同的计数器。 CPython解释器几乎可以在任何时间在线程之间切换。 只有最基本的操作是原子的，这意味着它们似乎是立即发生的，在求值或执行过程中不可能切换。 递增计数器需要多个基本操作：读旧值，加1，再写新值。 解释器可以在这些操作之间切换线程。

为了显示解释器在错误的时间切换线程时会发生什么，我们试图通过休眠0秒来强制切换。 当这段代码运行时，解释器通常会在睡眠调用时切换线程。 这可能导致以下操作序列：

Thread 0                    Thread 1
read counter[0]: 0
                            read counter[0]: 0
calculate 0 + 1: 1
write 1 -> counter[0]
                            calculate 0 + 1: 1
                            write 1 -> counter[0]

最终结果是计数器的值为1，尽管它被加了两次！更糟糕的是，解释器可能很少会在错误的时间切换，这使得调试非常困难。即使使用sleep调用，这个程序有时也会产生正确的计数2，有时会产生错误的计数1。

这个问题只在共享数据可能被一个线程改变而另一个线程访问它时才会出现。这样的冲突称为竞争条件，它是只存在于平行世界中的bug的一个例子。

为了避免竞争条件，必须保护可能被多个线程修改和访问的共享数据，防止并发访问。例如，如果我们可以确保线程1只在线程0完成对计数器的访问后才访问计数器，或者反之亦然，我们就可以保证计算出正确的结果。 如果保护共享数据不受并发访问的影响，我们说共享数据是同步的。 在接下来的几个小节中，我们将看到提供同步的多种机制。

4.8.3 当不需要同步时

在某些情况下，如果并发访问不能导致不正确的行为，那么对共享数据的访问不需要同步。最简单的例子是只读数据。因为这样的数据永远不会发生变化，所以所有线程无论何时访问数据，都会读取相同的值。

在极少数情况下，发生突变的共享数据可能不需要同步。然而，理解这种情况需要深入了解解释器和底层软件和硬件的工作方式。考虑下面的例子：

items = []
flag = []

def consume():
    while not flag:
        pass
    print('items is', items)

def produce():
    consumer = threading.Thread(target=consume, args=())
    consumer.start()
    for i in range(10):
        items.append(i)
    flag.append('go')

produce()

这里，生产者线程将条目添加到条目中，而消费者则等待flag非空。当生产者添加完条目后，它会添加一个标记元素，允许消费者继续。

在大多数Python实现中，这个例子都可以正常工作。然而，在其他编译器和解释器，甚至是硬件本身中，一种常见的优化是在单个线程中重新排序操作，这些线程不依赖于彼此的数据。在这样的系统中，语句flag.append('go')可以移动到循环之前，因为两者都不依赖对方获取数据。 通常，您应该避免编写这样的代码，除非您确定底层系统不会重新排序相关操作。

4.8.4 同步数据结构

同步共享数据的最简单方法是使用提供同步操作的数据结构。queue模块包含一个queue类，它提供了先入先出的同步数据访问。put方法向队列添加一个项目，get方法检索一个项目。类本身确保这些方法是同步的，因此无论线程操作如何交错，项目都不会丢失。下面是一个生产者/消费者使用队列的例子：

from queue import Queue

queue = Queue()

def synchronized_consume():
    while True:
        print('got an item:', queue.get())
        queue.task_done()

def synchronized_produce():
    consumer = threading.Thread(target=synchronized_consume, args=())
    consumer.daemon = True
    consumer.start()
    for i in range(10):
        queue.put(i)
    queue.join()

synchronized_produce()

除了队列、get和put调用之外，这段代码还做了一些修改。我们已经将使用者线程标记为守护进程，这意味着程序在退出之前不会等待该线程完成。 这允许我们在消费者中使用一个无限循环。 然而，我们确实需要确保主线程退出，但只能在队列中的所有项都被消耗完之后。消费者调用task_done方法来通知队列它已经完成了对一个项的处理，主线程调用join方法，该方法将等待直到所有项都处理完毕，以确保程序只在此情况下才退出。

使用队列的一个更复杂的例子是并行web爬虫，它搜索网站上的死链接。 该爬虫跟踪同一站点托管的所有链接，因此它必须处理大量url，不断向队列中添加新url，并删除url进行处理。 通过使用同步队列，多个线程可以安全地并发地向数据结构添加或删除数据。

4.8.5 锁

当一个特定数据结构的同步版本不可用时，我们必须提供我们自己的同步。锁是实现这一目的的基本机制。 它最多可以被一个线程获取，在此之后，其他线程都不能获取它，直到之前获取它的线程释放它。

在Python中，threading模块包含一个Lock类来提供锁定。锁有获取和释放方法来获取和释放锁，并且该类保证一次只有一个线程可以获取它。所有其他试图在锁已经被持有时获取锁的线程都被迫等待，直到锁被释放。

对于保护特定数据集的锁，所有线程都需要遵循一个规则:任何线程都不会访问任何共享数据，除非它拥有特定的锁。 实际上，所有线程都需要在对锁的获取和释放调用中“包装”它们对共享数据的操作。

在并行web爬虫中，使用一个集合来跟踪任何线程遇到的所有URL，从而避免多次处理特定URL(并可能陷入一个循环)。然而，Python没有提供同步集，所以我们必须使用锁来保护对普通集的访问：

seen = set()
seen_lock = threading.Lock()

def already_seen(item):
    seen_lock.acquire()
    result = True
    if item not in seen:
        seen.add(item)
        result = False
    seen_lock.release()
    return result

这里需要一个锁，以防止另一个线程在这个线程检查URL是否在集合中并将其添加到集合之间将URL添加到集合中。此外，向set添加数据并不是原子的，因此并发尝试向set添加数据可能会破坏其内部数据。

在这段代码中，我们必须小心在释放锁之后才返回。 通常，我们必须确保在不再需要锁时释放它。 这可能非常容易出错，特别是在出现异常的情况下，因此Python为我们提供了一个with复合语句来处理获取和释放锁的问题：

def already_seen(item):
    with seen_lock:
        if item not in seen:
            seen.add(item)
            return False
        return True

with语句确保seen_lock在套件执行之前被获取，并且在套件因为任何原因退出时被释放。(with语句实际上可以用于锁之外的操作，不过这里不介绍其他用途。)

必须相互同步的操作必须使用相同的锁。但是，必须用同一集中的操作来同步的两个互不相连的操作集应该使用两个不同的锁对象，以避免过度同步。

4.8.6 壁垒

另一种避免对共享数据的冲突访问的方法是将程序划分为几个阶段，确保共享数据在一个没有其他线程访问它的阶段发生变化。 barrier将程序分为几个阶段，要求所有的线程在任何一个线程继续运行之前都要到达它。在barrier之后执行的代码不能与barrier之前执行的代码并发。

在Python中，threading模块以barrier实例的wait方法的形式提供了barrier：

counters = [0, 0]
barrier = threading.Barrier(2)

def count(thread_num, steps):
    for i in range(steps):
        other = counters[1 - thread_num]
        barrier.wait() # wait for reads to complete
        counters[thread_num] = other + 1
        barrier.wait() # wait for writes to complete

def threaded_count(steps):
    other = threading.Thread(target=count, args=(1, steps))
    other.start()
    count(0, steps)
    print('counters:', counters)

threaded_count(10)

在本例中，对共享数据的读写发生在由屏障分隔的不同阶段。写发生在同一阶段，但它们是不连贯的；为了避免在同一阶段对同一数据进行并发写入，这种脱节是必要的。因为这段代码是正确同步的，所以两个计数器的末尾都是10。

多线程粒子模拟器以类似的方式使用barrier来同步对共享数据的访问。在模拟中，每个线程拥有一些粒子，所有这些粒子在许多离散时间间隔的过程中相互作用。一个粒子有一个位置，速度和加速度，一个新的加速度是根据其他粒子的位置在每个时间步计算的。粒子的速度必须相应地更新，它的位置也要根据它的速度来调整。

与上面的简单例子一样，有一个读阶段，在这个阶段中所有粒子的位置都被所有线程读取。每个线程更新自己的粒子在这个阶段的加速，但因为这些是不间断的写，他们不需要同步。在写入阶段，每个线程更新自己的粒子的速度和位置。同样，这些是不连贯的写操作，它们在读阶段受到屏障的保护。

4.8.7 消息传递

避免共享数据不适当变化的最后一种机制是完全避免对同一数据的并发访问。在Python中，使用多进程而不是线程自然会导致这种情况，因为进程运行在单独的解释器中，有它们自己的数据。多个进程所需的任何状态都可以通过在进程之间传递消息来进行通信。

multiprocessing模块中的Pipe类提供了进程之间的通信通道。 默认情况下，它是双工的，这意味着双向通道，尽管传入参数False会导致单向通道。send方法通过通道发送对象，而recv方法接收对象。后者是阻塞的，这意味着调用recv的进程将等待直到接收到对象。

下面是一个使用流程和管道的生产者/消费者示例：

def process_consume(in_pipe):
    while True:
        item = in_pipe.recv()
        if item is None:
            return
        print('got an item:', item)

def process_produce():
    pipe = multiprocessing.Pipe(False)
    consumer = multiprocessing.Process(target=process_consume, args=(pipe[0],))
    consumer.start()
    for i in range(10):
        pipe[1].send(i)
    pipe[1].send(None) # done signal

process_produce()

在本例中，我们使用一个None消息来表示通信结束。在创建使用者流程时，我们还将管道的一端作为参数传递给目标函数。 这是必要的，因为状态必须在进程之间显式地共享。

粒子模拟器的多进程版本使用管道在每个时间步的进程之间通信粒子位置。事实上，它使用管道在进程之间建立一个完整的循环管道，以减少通信。每个过程都将其自身粒子的位置注入其管道阶段，最终通过管道的完整旋转。在旋转的每一个步骤中，一个过程从它自己的管道阶段的位置向它自己的粒子施加力，所以在完全旋转之后，所有的力都被施加到它自己的粒子上。

multiprocessing模块为进程提供了其他同步机制，包括同步队列、锁和从Python 3.3开始的barrier。 例如，可以使用锁或屏障将打印同步到屏幕上，从而避免我们前面看到的不正确的显示输出。

4.8.8 同步陷阱

虽然同步方法可以有效地保护共享状态，但它们也可能被错误地使用，无法完成正确的同步、过度同步或导致程序因死锁而挂起。

同步不足。并行计算的一个常见缺陷是忽略了正确同步共享访问。在set示例中，我们需要同步成员检查和插入，以便另一个线程不能在这两个操作之间执行插入操作。不能同时同步这两个操作是错误的，即使它们是分别同步的。

过度同步。另一个常见错误是过度同步程序，这样非冲突操作就不能并发发生。作为一个简单的例子，我们可以避免对共享数据的所有冲突访问，方法是在线程启动时获取主锁，并在线程完成时释放主锁。这将序列化我们的整个代码，这样就不会并行运行任何东西。在某些情况下，这甚至会导致我们的程序无限期挂起。例如，考虑一个消费者/生产者程序，在这个程序中，消费者获得了锁，但从不释放它。 这就阻止了生产者生产任何物品，反过来又阻止了消费者做任何事情，因为它没有东西可以消费。

虽然这个例子很简单，但在实践中，程序员经常在某种程度上过度同步他们的代码，从而阻止了他们的代码完全利用可用的并行性。

死锁。因为它们会导致线程或进程彼此等待，同步机制很容易出现死锁，也就是两个或多个线程或进程被卡住，等待彼此完成。我们刚刚看到了忽略释放锁会如何导致线程无限期地卡住。但是，即使线程或进程正确地释放了锁，程序仍然可能出现死锁。

死锁的来源是循环等待，下面用进程说明了这一点。 任何进程都不能继续，因为它正在等待其他正在等待它完成的进程。

作为示例，我们将用两个进程设置死锁。假设它们共享一个双工管道，并尝试以如下方式相互通信：

def deadlock(in_pipe, out_pipe):
    item = in_pipe.recv()
    print('got an item:', item)
    out_pipe.send(item + 1)

def create_deadlock():
    pipe = multiprocessing.Pipe()
    other = multiprocessing.Process(target=deadlock, args=(pipe[0], pipe[1]))
    other.start()
    deadlock(pipe[1], pipe[0])

create_deadlock()

两个进程都尝试先接收数据。回想一下，recv方法会阻塞直到有可用的项。由于两个进程都没有发送任何东西，所以它们都将无限期地等待对方发送数据，从而导致死锁。

同步操作必须正确对齐，以避免死锁。这可能需要在接收之前通过管道发送，以相同的顺序获取多个锁，并确保所有线程在正确的时间到达正确的屏障。

4.8.9 结论

正如我们所看到的，并行为编写正确和高效的代码带来了新的挑战。在可预见的未来，随着硬件级并行度的不断提高，并行计算在应用程序设计中将变得越来越重要。有一个非常活跃的研究团体，旨在让程序员更容易实现并行，减少出错的可能性。我们这里的讨论只是作为对计算机科学这个关键领域的基本介绍。