挑战

编写并发程序本身就很复杂,即便是使用C++11和C++14的新特性,也是如此。我希望通过用一整章的内容来讨论并发编程的挑战,读者们会更清楚其中的陷阱与挑战。

ABA问题

ABA表示读取了一个值两次,每次都返回A值。因此,可以得出这样的结论:两次读取之间,相应的变量没有任何变化。然而,在两次读取之间,变量可能有被更新为B的时刻。

用一个简单的场景来比拟这个问题。

一个例子

这个场景里,你坐在车里等待交通灯变绿,绿色代表B,红色代表A。接下来会发生了什么?

  1. 你看到交通灯,它是红色的(A)。

  2. 因为很无聊,你打开手机看新闻,而忘记了时间。

  3. 当你再看一次交通灯时。该死!还是红色(A)。

当然,交通灯在你两次抬头看之间已经变成绿灯过。对于线程(进程)来说,意味着什么?

  1. 线程1读取值为A的变量var

  2. 线程1被抢占,线程2运行。

  3. 线程2将变量var从A更改为B,再更改为A。

  4. 线程1继续运行并检查变量var的值并得到A。因为获取到值A,线程1继续运行。

通常这不是一个问题,可以忽略。

非关键的ABA

// fetch_mult.cpp

#include <atomic>
#include <iostream>

template <typename T>
T fetch_mult(std::atomic<T>& shared, T mult) {
  T oldValue = shared.load();
  while (!shared.compare_exchange_strong(oldValue, oldValue * mult));
  return oldValue;
}

int main() {
  std::atomic<int> myInt{ 5 };
  std::cout << myInt << std::endl;
  fetch_mult(myInt, 5);
  std::cout << myInt << std::endl;
}

compare_exchange_strongcompare_exchange_weak可以在fetch_mult(第6行)中观察到的ABA问题。fetch_multstd::atomic<t>& sharedmult相乘。

关键是,读取旧值T oldValue = shared.load()第8行和第9行中的新值比较之间有一个小的时间窗口。因此,另一个线程可以介入,将oldValue从更改为另一个值,然后再返回oldValue。旧值是A,另一个线程修改的值是ABA中的B。

通常,当读操作处理相同的、未更改的变量,则没有什么影响。但是,在无锁并发的数据结构中,ABA可能会产生重大影响。

无锁数据结构

这里不会详细介绍无锁数据结构,仅用单链表实现的无锁堆栈,堆栈只支持两个操作:

  1. pop:弹出顶部对象,并返回指向它的指针。

  2. push:将指定的对象推入堆栈。

这里使用伪代码描述pop操作,以便了解ABA问题。pop操作执行以下步骤:

  1. 获取头节点:head

  2. 获取后续节点:headNext

  3. 如果head仍然是堆栈的头节点,则将headNext作为新的头结点。

下面是堆栈的前两个节点:

Stack: TOP -> head -> headNext -> ...

现在,来构造ABA问题的情景。

构造ABA

我们从下面的堆栈开始:

Stack: TOP -> A -> B -> C

线程1处于活动状态,希望弹出堆栈的头节点。

  • Thread 1操作时

    head = A

    headNext = B

线程1完成pop前,线程2开始工作。

  • Thread 2 pop A

    Stack: TOP -> B -> C

  • Thread 2 pop B 并且删除B

    Stack: TOP -> C

  • Thread 2把A推回去

    Stack: TOP -> A -> C

线程1重新调度,并检查A == head,因为当前A == head,那么headNext应该是B,但B已经被删除了。因此,程序具有未定义行为。

用什么来拯救ABA问题呢?接下来就介绍,ABA问题的一些补救措施。

补救措施

ABA的概念问题很容易理解,解决方案是消除节点过早的删除。以下是一些补救措施:

标记参考状态

可以使用地址的低位向每个节点添加标记,以表示节点成功修改的频率。尽管检查返回true,但比较-交换(CAS)会失败。这个想法并不能解决问题,因为标记位可能最终会交换。

引用标记状态通常用于事务内存中。

接下来的三种技术是基于延迟回收的思想。

垃圾收集

垃圾收集只保证在不再需要时删除变量。这听起来很有希望解决ABA问题,但有一个明显的缺点。大多数垃圾收集器不是无锁的,即使有一个无锁的数据结构,整个系统也不是无锁的。

风险指针

维基页面: Hazard Pointers

风险指针系统中,每个线程都保存一个风险指针列表,指示线程当前正在访问哪些节点(许多系统中,这个“列表”可能仅限于一两个元素)。风险指针列表中的节点不能被任何其他线程修改或释放。当一个线程想要删除一个节点时,它会将其放在一个节点列表中,进行“稍后释放”,直到没有其他线程的危险列表包含该指针时,才释放该节点的内存。一个专门的垃圾收集线程可以手工进行垃圾收集(如果“稍后释放”的列表由所有线程共享);或者,清理“被释放”列表可以由每个工作线程,作为“pop”等操作的一部分。

RCU 读取-复制-更新

RCU是Read Copy Update的缩写,是一种用于只读数据结构的同步技术。RCU是由Paul McKenney创建的,自2002年以来一直在Linux内核中使用。

思想很简单,就跟缩写一样,要修改数据,要复制数据。反之,所有的读取都使用原始数据。如果没有读取操作,那么可以安全地将数据进行修改。

要了解更多关于RCU的细节,请阅读Paul McKenney的这篇文章:What is RCU, Fundamentally?

两个新的提案

作为并发工具包的一部分,有两个关于未来C++标准的提案。关于风险指针的提案是P0233R0,关于RCU的提案是P0461R0

阻塞问题

为了说明我的观点,需要将条件变量与谓词结合。不这样做的话,程序可能会出现伪唤醒或未唤醒的情况。

如果使用没有谓词的条件变量,则通知线程可能在等待线程等待之前发送通知,等待线程将永远等待,这种现象被称为“未唤醒“。

程序如下。

// conditionVariableBlock.cpp

#include <iostream>
#include <condition_variable>
#include <mutex>
#include <thread>

std::mutex mutex_;
std::condition_variable condVar;

bool dataReady;


void waitingForWork() {

  std::cout << "Worker: Waiting for work." << std::endl;

  std::unique_lock<std::mutex> lck(mutex_);
  condVar.wait(lck);
  // do the work
  std::cout << "Work done." << std::endl;

}

void setDataReady() {

  std::cout << "Sender: Data is ready." << std::endl;
  condVar.notify_one();

}

int main() {

  std::cout << std::endl;

  std::thread t1(setDataReady);
  std::thread t2(waitingForWork);

  t1.join();
  t2.join();

  std::cout << std::endl;
}

程序的第一次工作得很好,第二次锁定的原因是notify(第28行)发生在线程t2(第37行)等待之前(第19行)。

当然,死锁和活锁是条件竞争的副产物。死锁通常取决于线程的交错,有时会发生,有时不会。活锁与死锁类似,当死锁阻塞时,活锁“似乎''没有阻塞程序。

破坏程序的不变量

程序不变量,应该在程序的整个生命周期中”保持不变“。

恶性条件竞争破坏程序的不变量。下面程序的不变量是所有余额的总和,例子中是200欧元,因为每个账户起步都是100欧元(第9行)。

// breakingInvariant.cpp

#include <atomic>
#include <functional>
#include <iostream>
#include <thread>

struct Account {
  std::atomic<int> balance{ 100 };
};

void transferMoney(int amount, Account& from, Account& to) {
  using namespace std::chrono_literals;
  if (from.balance >= amount) {
    from.balance -= amount;
    std::this_thread::sleep_for(1ns);
    to.balance += amount;
  }
}

void printSum(Account& a1, Account& a2) {
  std::cout << (a1.balance + a2.balance) << std::endl;
}

int main() {

  std::cout << std::endl;

  Account acc1;
  Account acc2;

  std::cout << "Initial sum: ";
  printSum(acc1, acc2);

  std::thread thr1(transferMoney, 5, std::ref(acc1), std::ref(acc2));
  std::thread thr2(transferMoney, 13, std::ref(acc2), std::ref(acc1));
  std::cout << "Intermediate sum: ";
  std::thread thr3(printSum, std::ref(acc1), std::ref(acc2));

  thr1.join();
  thr2.join();
  thr3.join();

  std::cout << "  acc1.balance: " << acc1.balance << std::endl;
  std::cout << "  acc2.balance: " << acc2.balance << std::endl;

  std::cout << "Final sum: ";
  printSum(acc1, acc2);

  std::cout << std::endl;

}

开始时,账户的总数是200欧元。第33行,通过使用第21 - 23行中的printSum函数来显示金额和。第38行使不变量可见。因为第16行有1ns的短睡眠,所以中间的金额是182欧元。最后,每个账户的余额都是正确的(第44行和第45行),金额是200欧元(第48行)。

下面是程序的输出。

数据竞争

数据竞争是指至少两个线程同时访问一个共享变量的情况,并且至少有一个线程尝试修改该变量。

程序有数据竞争,则会出现未定义行为,结果是不可预期的。

来看一个数据竞争的程序。

// addMoney.cpp

#include <functional>
#include <iostream>
#include <thread>
#include <vector>

struct Account {
  int balance{ 100 };
};

void addMoney(Account& to, int amount) {
    to.balance += amount;
}

int main() {

  std::cout << std::endl;

  Account account;

  std::vector<std::thread> vecThreads(100);


  for (auto& thr : vecThreads) thr = std::thread(addMoney, std::ref(account), 50);

  for (auto& thr : vecThreads) thr.join();


  std::cout << "account.balance: " << account.balance << std::endl;

  std::cout << std::endl;

}

100个线程addMoney函数将向相同的帐户(第20行)添加50欧元(第25行)。关键的,对账户的写入是不同步的,这里有一个数据竞争,因为是未定义行为,所以结果无效。最后的余额(第30行)会在5000欧元和5100欧元之间。

死锁

死锁是一种状态,因为要等待没有得到的资源的释放,所以至少有一个线程会永久阻塞。

造成死锁的主要原因有两个:

  1. 互斥锁未解锁。

  2. 以不同的顺序锁定互斥锁。

为了避免第二个问题,在经典C++中使用了诸如层次锁之类的技术。

有关死锁,以及如何用现代C++克服死锁的详细信息,请参阅互斥量和锁的章节内容。

多次锁定非递归互斥锁

多次锁定非递归互斥锁会导致未定义行为。

// lockTwice.cpp

#include <iostream>
#include <mutex>

int main() {

    std::mutex mut;

    std::cout << std::endl;

    std::cout << "first lock call" << std::endl;

    mut.lock();

    std::cout << "second lock call" << std::endl;

    mut.lock();

    std::cout << "third lock call" << std::endl;
}

通常会死锁。

伪共享

当处理器从主存中读取一个变量(如int)时,从内存中读取的数据要大于int的大小。处理器会从缓存中读取整个高速缓存行(通常为64字节)。

如果两个线程,同时读取位于同一高速缓存行上的不同变量a和b,则会发生伪共享。虽然a和b在逻辑上是分开的,但在物理地址上是相连的。由于a和b共享同一条高速缓存线行,因此有必要在高速缓存行上进行硬件同步。得到了正确的结果,但是并发的性能下降了。正是这种现象发生在下面的程序中:

// falseSharing.cpp

#include <algorithm>
#include <chrono>
#include <iostream>
#include <random>
#include <thread>
#include <vector>

constexpr long long size{ 100'000'000 };

struct Sum {
  long long a{ 0 };
  long long b{ 0 };
};

int main() {

  std::cout << std::endl;

  Sum sum;

  std::cout << &sum.a << std::endl;
  std::cout << &sum.b << std::endl;

  std::cout << std::endl;

  std::vector<int> randValues, randValues2;
  randValues.reserve(size);
  randValues2.reserve(size);

  std::mt19937 engine;
  std::uniform_int_distribution<> uniformDist(1, 10);

  int randValue;
  for (long long i = 0; i < size; ++i) {
    randValue = uniformDist(engine);
    randValues.push_back(randValue);
    randValues2.push_back(randValue);
  }

  auto sta = std::chrono::steady_clock::now();

  std::thread t1([&sum, &randValues] {
    for (auto val : randValues) sum.a += val;
    });

  std::thread t2([&sum, &randValues2] {
    for (auto val : randValues2)sum.b += val;
    });

  t1.join(), t2.join();

  std::chrono::duration<double> dur = std::chrono::steady_clock::now() - sta;
  std::cout << "Time for addition " << dur.count()
    << " seconds" << std::endl;

  std::cout << "sum.a: " << sum.a << std::endl;
  std::cout << "sum.b: " << sum.b << std::endl;

  std::cout << std::endl;

}

第13行和第14行中的变量ab共享同个缓存行。线程t1(第44行)和线程t2同时使用两个变量,对向量randValuesrandValues2中的元素进行求和。两个向量在1到10之间都有1亿个整数。程序的输出显示了一些有趣的事情,ab在8字节边界上对齐,因为我的操作系统中的long long int是8字节对齐的。

如果将ab的对齐方式改为64字节会发生什么?64字节是我系统上的高速缓存行的大小。我要对结构做点小改动,这次不用种子来生成随机数,所以每次都得到的随机数相同。

struct Sum{
  alignas(64) long long a{0};
  alignas(64) long long b{0};
};

现在,ab在64字节边界处对齐,程序速度提高了6倍多。原因是ab现在不在同一高速缓存行上。

用优化器检测伪共享

如果我用最大的优化选项编译的程序,优化器会检测到伪共享并消除它。这意味着,我得到了相同的性能数据与真共享,这也适用于Windows。以下是优化后的性能数字。

C++17中的std:: hardware_destructive_interference_size和与std:: hardware_constructive_interference_size

std::hardware_destructive_interference_sizestd::hardware_constructive_interference_size允许以一种可移植的方式处理高速缓存行的大小。std::hardware_destructive_interference_size返回两个对象之间的最小偏移量,以避免伪共享;std::hardware_constructive_interference_size返回相邻内存的最大大小,以满足真共享。

在C++17中,Sum可以以一种平台无关的方式编写。

struct Sum{
 alignas(std::hardware_destructive_interference_size) long long a{0};
 alignas(std::hardware_destructive_interference_size) long long b{0};
};

变量的生命周期问题

写一个具有生命周期相关问题的C++示例非常容易。让创建的线程t在后台运行(也就是说,它通过调用t.detach()来分离),并且让它只完成一半的工作。这里,创建者线程不会等待子线程完成。在这种情况下,必须非常小心,最好不要在子线程中使用属于创建线程的任何东西。

// lifetimeIssues.cpp

#include <iostream>
#include <string>
#include <thread>

int main() {

  std::cout << "Begin: " << std::endl;

  std::string mess{ "Child thread" };

  std::thread t([&mess] {std::cout << mess << std::endl; });
  t.detach();

  std::cout << "End:" << std::endl;

}

这程序太简单了。线程t使用std::cout和变量mess,它们都属于主线程。结果是,在第二次运行时,我看不到子线程的输出。只有“Begin:”(第9行)和“End:”(第16行)打印了出来。

移动线程

移动线程会使线程的生命周期问题变得更加复杂。

线程支持移动语义,但不支持复制语义。原因是std::thread的复制构造函数被设置为deletethread (const thread&) = delete;。试想,如果线程在持有锁的情况下能进行复制,会发生什么。

让我们移动一个线程。

错误地移动线程

// threadMoved.cpp

#include <iostream>
#include <thread>
#include <utility>

int main(){

  std::thread t([]{std::cout << std::this_thread::get_id();});
  std::thread t2([]{std::cout << std::this_thread::get_id();});

  t = std::move(t2);
  t.join();
  t2.join();
}

线程tt2应该完成它们的工作:打印它们的id。除此之外,线程t2的所有权移动到t(第12行)。最后,主线程处理它的子线程并汇入它们。等一下,结果与我的预期大不相同:

出了什么问题?这里有两个问题:

  1. 通过移动线程t2, t获得一个新的可调用单元,并调用它的析构函数。结果,t的析构函数调用std::terminate,原始的t线程仍然是可汇入的。

  2. 线程t2没有相关的可调用单元,在没有可调用单元的线程上调用join会导致异常std::system_error

了解了这一点,修复工作就很简单了。

// threadMovedFixed.cpp

#include <iostream>
#include <thread>
#include <utility>

int main(){

  std::thread t([]{std::cout << std::this_thread::get_id();});
  std::thread t2([]{std::cout << std::this_thread::get_id();});

  t.join();
  t = std::move(t2);
  t2.join();

  std::cout << "\n";
  std::cout << std::boolalpha << "t2.joinable(): " << t2.joinable() << std::endl;

}

结果是线程t2不可汇入。

竞态条件

竞态条件是一种情况,其中操作的结果取决于某些操作的交错。

竞态条件很难发现。由于其取决于线程是否交错出现,也就是内核的数量、系统的利用率或可执行文件的优化级别,都可能是导致出现竞态条件的原因。

竞态条件本身并没什么。但线程以不同的方式交织在一起后,常常会导致严重的问题。这种情况下,称其为恶性竞争条件。恶意竞争条件的典型症状表现:数据竞争、破坏程序不变量、阻塞线程,或变量有生存周期问题等。

Last updated