【安全编码】并发和多线程

本篇总结了《C和C++安全编码》中并发和多线程相关的内容。

多线程程序常见错误

当并发实现得不正确时，就会产生漏洞。多线程程序中常见错误有：

• 没有用锁保护共享数据(即数据竞争)。
• 当锁确实存在时，不使用锁访问共享数据。
• 过早释放锁。
• 对操作的一部分获取正确的锁，释放它，后来再次取得它，然后又释放它，而正确的做法是一直持有该锁。
• 在想要用局部变量时，意外地通过使用全局变量共享数据。
• 在不同的时间对共享数据使用两个不同的锁。
• 由下列情况引起死锁：不恰当的锁定序列(加锁和解锁序列必须保持一致)；锁定机制使用不当或错误选择；不释放锁或试图再次获取已经持有的锁。

一些常见的并发陷阱包括以下内容：

• 缺乏公平：所有线程没有得到平等的机会来获得处理。
• 饥饿：当一个线程霸占共享资源、阻止其它线程使用时发生。
• (3).活锁：线程继续执行，但未能获得处理。
• 假设线程将：以一个特定的顺序运行；不能同时运行；同时运行；在一个线程结束前获得处理。
• 假设一个变量不需要锁定，因为开发人员认为只有一个线程写入它且所有其它线程都读取它。这还假定该变量上的操作是原子的。
• 使用非线程安全库。如果一个库能保证由多个线程同时访问时不会产生数据竞争，那么认为它是线程安全的。
• 依托测试，以找到数据竞争和死锁。
• 内存分配和释放问题。当内存在一个线程中分配而在另一个线程中释放时，这些问题可能出现，不正确的同步可能会导致内存仍然被访问时被释放。

1 死锁

传统上，通过使冲突的竞争窗口互斥，使得一旦一个临界区开始执行时，没有额外的线程可以执行，直到前一个线程退出临界区为止，从而消除竞争条件。但是，同步原语的不正确使用可能会导致死锁(deadlock)。当两个或多个控制流以彼此都不可以继续执行的方式阻止对方时，就会发生死锁。特别是，对于一个并发执行流的循环，如果其中在循环中的每个流都已经获得了导致在循环中随后的流悬停的同步对象，则会发生死锁。死锁的一个明显的安全漏洞是拒绝访问。

int shared_data5 = 0;
std::mutex* locks5 = nullptr;
int thread_size5;
void thread_function5(int id)
{
	if (0) { // 产生死锁
		// 此代码将产生一个固定数量的线程,每个线程都修改一个值,然后读取它.虽然通常一个锁就足够了,但是每个
		// 线程(thread_size5)都用一个锁守卫共享数据值.每个线程都必须获得两个锁,然后才能再访问该值.如果
		// 一个线程首先获得锁0,第二个线程获得锁1,那么程序将会出现死锁
		if (id % 2)
			for (int i = 0; i < thread_size5; ++i)
				locks5[i].lock();
		else
			for (int i = thread_size5; i >= 0; --i)
				locks5[i].lock();
 
		shared_data5 = id;
		fprintf(stdout, "thread: %d, set shared value to: %d\n", id, shared_data5);
 
		if (id % 2)
			for (int i = thread_size5; i >= 0; --i)
				locks5[i].unlock();
		else
			for (int i = 0; i < thread_size5; ++i)
				locks5[i].unlock();
	}
	else { // 不会产生死锁
		// 每个线程都以同一顺序获取锁,可以消除潜在的死锁.下面的程序无论创建多少线程都不会出现死锁
		for (int i = 0; i < thread_size5; ++i)
			locks5[i].lock();
 
		shared_data5 = id;
		fprintf(stdout, "thread: %d, set shared value to: %d\n", id, shared_data5);
 
		for (int i = 0; i < thread_size5; ++i)
			locks5[i].unlock();
	}
}
 
void test_concurrency_deadlock()
{
	thread_size5 = 5;
	std::thread* threads = new std::thread[thread_size5];
	locks5 = new std::mutex[thread_size5];
 
	for (size_t i = 0; i < thread_size5; ++i)
		threads[i] = std::thread(thread_function5, i);
 
	for (size_t i = 0; i < thread_size5; ++i)
		threads[i].join();
 
	// test_concurrency_deadlock()继续之前,等待直到线程完成
	delete[] locks5;
	delete[] threads;
	fprintf(stdout, "Done\n");
}

像所有的数据竞争一样，死锁行为对环境的状态而不只是程序的输入敏感。特别是，死锁(和其它的数据竞争)可能对以下条件敏感：

• 处理器速度。
• 进程或线程调度算法的变动。
• 在执行的时候，强加的不同内存限制。
• 任何异步事件中断程序执行的能力。
• 其它并发执行进程的状态。

2 过早释放锁

std::mutex shared_lock6;
int shared_data6 = 0;
void thread_function6(int id)
{
	// 每个线程都把一个共享变量设置为它的线程编号,然后打印出共享变量的值.为了防止数据竞争,每个线程都
	// 锁定一个互斥量,以使变量被正确地设置
	if (0) { // 过早地释放锁
		// 当共享变量的每一个写操作都由互斥量所保护时,随后的读取是不受保护的
		shared_lock6.lock();
		shared_data6 = id;
		fprintf(stdout, "thread: %d, set shared value to: %d\n", id, shared_data6);
		shared_lock6.unlock();
		std::this_thread::sleep_for(std::chrono::milliseconds(id) * 100);
		fprintf(stdout, "thread: %d, has shared value to: %d\n", id, shared_data6);
	}
	else {
		// 读取和写入共享数据都必须受到保护,以确保每一个线程读取到它写入的相同的值.将临界区扩展为包括读取值,此代码就呈现为线程安全的
		// 需要注意的是,线程的顺序仍然可以有所不同,但每个线程都正确地打印出线程编号
		shared_lock6.lock();
		shared_data6 = id;
		fprintf(stdout, "thread: %d, set shared value to: %d\n", id, shared_data6);
		std::this_thread::sleep_for(std::chrono::milliseconds(id) * 100);
		fprintf(stdout, "thread: %d, has shared value to: %d\n", id, shared_data6);
		shared_lock6.unlock();
	}
}
 
void test_concurrency_prematurely_release_lock()
{
	const size_t thread_size = 10;
	std::thread threads[thread_size];
 
	for (size_t i = 0; i < thread_size; ++i)
		threads[i] = std::thread(thread_function6, i);
 
	for (size_t i = 0; i < thread_size; ++i)
		threads[i].join();
 
	// test_concurrency_prematurely_release_lock()继续之前,等待直到线程完成
	fprintf(stdout, "Done\n");
}

3 争用

当一个线程试图获取另一个线程持有的锁时，就会发生锁争用。有些锁争用是正常的，这表明，锁正在”工作”，以防止竞争条件。过多的锁争用会导致性能不佳。减少持有锁的时间量或通过降低每个锁保护的粒度或资源量，可以解决锁争用导致的性能差的问题。持有锁的时间越长，另一个线程尝试获取锁，并被迫等待的概率将越大。反之，减少持有锁的持续时间就减少了争用。例如，不会作用于共享资源的代码，不需要在临界区之内得到保护，并可以与其它线程并行运行。在一个临界区之内执行一个阻塞操作延伸了临界区的持续时间，从而增加了潜在的争用。在临界区之内的阻塞操作也可能导致死锁。在临界区之内执行阻塞操作几乎始终是一个严重的错误。

锁的粒度也可以影响争用。增加由一个单一锁保护的共享资源的数量，或扩大共享资源的范围(例如，锁定整个表以访问一个单元格)，将使在同一时间多个线程尝试访问该资源的概率增大。在选择锁的数量时，增加锁的开销和减少锁争用之间有一个权衡。更细的粒度(每个保护少量的数据)需要更多的锁，使得锁本身的开销增加。额外的锁也会增加死锁的风险。锁一般是相当快的，但是，当然单个执行线程运行速度会比没有锁更慢。

4 ABA 问题

在同步过程中，当一个位置被读取两次，并有相同的值供读取时，就发生 ABA 的问题。然而，第二个线程已在两次读取之间执行并修改了这个值，执行其它工作，然后把值再修改回来，从而愚弄第一个线程，让它以为第二个线程尚未执行。实现无锁数据结构时，经常会遇到 ABA 问题。如果将一个条目从列表中移除，并删除，然后分配一个新的条目，并把它添加到列表中，因为优化，新的对象通常会放置在被删除的对象的相同位置。因此，指向新条目的指针可能等于旧项目的指针，这可能会导致 ABA 问题。

自旋锁(spinlock)是一种类型的锁实现，其中线程在一个循环中反复尝试获得锁，直到它终于成功。一般而言，只有当等待获得锁的时间很短时，自旋锁才是有效的。在这种情况下，自旋锁避免了昂贵的上下文切换时间和在传统的锁中等待资源时，调度运行需要花费的时间。当获得锁的等待时间是明显长时，自旋锁在试图获取一个锁时就会浪费大量的 CPU 时间。一个常见的防止自旋锁浪费 CPU 周期的缓解措施是，在 while 循环中让该线程休眠或把控制让给其它线程。