Lock-free Programming Practical

Lock-free Queue

• We will implement a simple queue

  • No internal or external locks

• The queue is only accessed by two threads

  • A producer thread inserts elements into the queue
  • A consumer thread removes elements from the queue

• The code is carefully designed

  • The consumer and producer threads never work on adjacent elements
  • The two threads always work on different parts of the queue

• The queue has two iterators, iHead and iTail

  • iHead points to the element before the oldest element
  • iTail points to the element after the newest (most recently added)

Consumer Thread

• The Consumer thread does not modify the queue
  • It "removes" an element by incrementing iHead

Producer Thread

• The Producer inserts elements
  • It increments iTail
  • It also erases any elements which the Consumer has removed

Thread Separation

• Only the Producer thread can modify the queue

  • The Producer queue inserts elements
  • The Producer queue erases elements

• The two threads never overlap

  • iHead and iTail never refer to the same element
  • The Producer thread never modifies iHead
  • The Consumer thread never accesses elements after iHead

template <typename T> struct LockFreeQueue {
private:
  std::list<T> list;
  typename std::list<T>::iterator iHead, iTail;

public:
  LockFreeQueue() {
    list.push_back(T()); // Create a "dummy" element
    iHead = list.begin();
    iTail = list.end();
  }

  bool Consume(T &t) {
    auto iFirst = iHead; // Go to the first element
    ++iFirst;
    if (iFirst != iTail) { // If queue is not empty
      iHead = iFirst;      // update iHead
      t = *iHead;          // Fetch this first element
      return true;
    }
    return false; // No elements to fetch
  }

  void Produce(const T &t) {
    list.push_back(t);               // Add the new element
    iTail = list.end();              // Update tail
    list.erase(list.begin(), iHead); // Erase the removed elements
  }

  void Print() {
    auto head = iHead;
    ++head;

    for (auto el = head; el != iTail; ++el) {
      std::cout << *el << ", ";
    }
    std::cout << '\n';
  }
};

int main() {
  LockFreeQueue<int> lfq;
  std::vector<std::thread> threads;
  int j = 1;

  for (int i = 0; i < 10; ++i) {
    std::thread produce(&LockFreeQueue<int>::Produce, &lfq, std::ref(i));
    threads.push_back(std::move(produce));
    std::thread consume(&LockFreeQueue<int>::Consume, &lfq, std::ref(j));
    threads.push_back(std::move(consume));
  }

  for (auto &thr : threads) {
    thr.join();
  }

  lfq.Print();
}

Thread Safety

• The code has a data race
  • iHead and iTail can be accessed from different threads
  • At least one thread modifies them
  • The threads are not synchronized when they access these variables

Avoiding the Data Race

• iHead and iTail cannot be atomic types

  • std::list<T>::iterator is not trivially copyable

• We must use mutexes