Linked list node-level locking

Question

I am new to C++ threading library. I would love to get some comments on how to improve this code (if there are possible bugs, I would appreciate that feedback too). In particular, I want to refactor the code for the methods of the linked list:

1. How to return unique_lock in a function so that it didn't unlock the mutex under the hood?
2. Are there possible deadlocks with the lock acquisition that takes place in my code?

3. Does it look correct?

   #include <iostream>
   #include <thread>
   #include <mutex>
   #include <atomic>
   #include <vector>
   
   using namespace std;
   
   /*
   * This is an implementation of a singly linked list with node-level locks to
   * ensure mutual exclusion. The structure looks like this:
   *          head_ -> node_0 -> node_1 -> ... -> node_n -> tail_
   * Note that head_ and tail_ are dummy nodes.
   */
   class LockBasedLinkedList {
   public:
       // Initialize head_ to point to tail_ (empty list).
       LockBasedLinkedList() {
           tail_ = new Node(0);
           head_ = new Node(0, tail_);
       }
   
       /*
       * Insert a node with value val at position pos.
       *
       * To ensure mutual exclusion, the locks of the current node and the
       * previous nodes must be acquired for insertion to work. As soon as the
       * locks are acquired the code does roughly the following:
       *
       *      | prev -> node | prev -> node | prev    node |
       *      |              |          ^   |   v      ^   |
       *      |   new_node   |   new_node   |   new_node   |
       */
       void insert(int val, int pos) {
           Node *new_node = new Node(val);
           Node *prev = head_;
           unique_lock<mutex> prev_lk(prev->m);
           Node *node = prev->next;
           unique_lock<mutex> node_lk(node->m);
           for (int i = 0; i < pos && node != tail_; i++) {
               prev = node;
               node = node->next;
               prev_lk.swap(node_lk);
               node_lk = unique_lock<mutex>(node->m);
           }
           new_node->next = node;
           prev->next = new_node;
       }
   
       /*
       * Erase the node at position pos.
       *
       * To ensure mutual exclusion, follow the steps from insert(). As soon as
       * the locks are acquired the code does roughly the following:
       *
       *    (*)     (*)            (*)     (*)            (*)
       *  | prev -> node -> next | prev    node -> next | prev ---------> next |
       *  |                      |   v--------------^   |                      |
       *
       * (*) highlights the nodes whose locks are acquired.
       */
       void erase(int pos) {
           Node *prev = head_;
           unique_lock<mutex> prev_lk(prev->m);
           Node *node = prev->next;
           unique_lock<mutex> node_lk(node->m);
           for (int i = 0; i < pos && node != tail_; i++) {
               prev = node;
               node = node->next;
               prev_lk.swap(node_lk);
               node_lk = unique_lock<mutex>(node->m);
           }
           if (node == tail_) {
               return;
           }
           prev->next = node->next;
           node_lk.unlock();
           delete node;
       }
   
       int get(int pos) {
           Node *prev = head_;
           unique_lock<mutex> prev_lk(prev->m);
           Node *node = prev->next;
           unique_lock<mutex> node_lk(node->m);
           for (int i = 0; i < pos && node != tail_; i++) {
               prev = node;
               node = node->next;
               prev_lk.swap(node_lk);
               node_lk = unique_lock<mutex>(node->m);
           }
           if (node == tail_) {
               return 0;
           }
           return node->val;
       }
   
   private:
       struct Node {
           int val;
           Node *next;
           mutex m;
   
           Node(int val_, Node *next_ = nullptr) : val(val_), next(next_) {}
       };
   
       Node *head_, *tail_;
   };
   
   void testNThread(int n) {
       const int N = 10;
       vector<thread> threads(n);
       LockBasedLinkedList lst;
       for (int i = 0; i < n; i++) {
           threads[i] = thread([i, &lst, N]() {
               for (int j = 0; j < N; j++) {
                   lst.insert(i, 0);
               }
           });
       }
       for (int i = 0; i < n; i++) {
           threads[i].join();
       }
       for (int i = 0; i < N * n; i++) {
           int val = lst.get(0);
           lst.erase(0);
           cout << val << " ";
       }
       cout << "\n";
   }
   
   int main(int argc, char **argv) {
       int n = 10;
       if (argc >= 2) {
           n = atoi(argv[1]);
       }
       testNThread(n);
   }
   

Answer 1

Congratulations — I think this is the first concurrency-related code I've reviewed on CodeReview in which I have not managed to find any concurrency-related bugs! It looks like a solid implementation of "hand-over-hand" traversal. (Now that I've said that, I bet someone else will find a bug. :))

First, a few stylistic nitpicks:

using namespace std;

Never! Just write std::vector<std::thread> if that's what you mean. The small space savings isn't worth dealing with all the people like me who'll tell you over and over not to write using namespace std; at global scope. :)

---

const int N = 10;

Make this constexpr int N = 10; and you won't need to capture a copy of it in your lambda. (EDIT: Looks like I'm wrong. MSVC captures a copy in either case; GCC and Clang don't-capture in either case, unless the lambda takes the address of its copy, in which case they both capture. What a mess.)

---

for (int i = 0; i < n; i++) {
    threads[i].join();
}

Prefer to write this with a ranged for loop:

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

(EDIT: I use auto&& because it Always Works. You could use auto& or const auto& as appropriate, but then you'd have to worry about which one was appropriate in each given case, and so would your code-reviewer. auto&& Always Works.)

---

struct Node {
    int val;
    Node *next;
    mutex m;

    Node(int val_, Node *next_ = nullptr) : val(val_), next(next_) {}
};

Defaulted function arguments are the devil. And, completely independently, so are implicit conversions! I would prefer to write this as:

struct Node {
    int val;
    Node *next = nullptr;
    mutex m;

    explicit Node(int val_) : val(val_) {}
    explicit Node(int val_, Node *next_) : val(val_), next(next_) {}
};

or even

struct Node {
    int val;
    Node *next;
    mutex m;

    explicit Node(int val_) : Node(val_, nullptr) {}
    explicit Node(int val_, Node *next_) : val(val_), next(next_) {}
};

(EDIT: I put explicit on every constructor, even the multi-argument ones, unless I have a specific reason that I want implicit conversion. In C++11, an implicit multi-arg constructor permits e.g.

void foo(const Node&);
Node bar() {
    foo({1, nullptr});
    return {1, nullptr};
}

and I consider preventing that "feature" to be a good thing.)

---

Okay, now for the serious stuff.

void erase(int pos);

This is a really weird API for a linked list. For one thing, even for a non-concurrent linked list, you're turning "erase a node" into an O(n) operation. More importantly for a concurrent linked list, I can't see how this operation is useful at all! Suppose some thread says "please erase the 42nd element." How does it even know what the 42nd element is, given that any other thread could modify the list at any time?

---

Your use of raw new and delete looks safe to me, but it would be a lot easier to verify its safety if you just didn't use raw new and delete! Consider this trivial rewrite of insert:

void insert(int val, int pos) {
    auto new_node = std::make_unique<Node>(val);
    Node *prev = head_;
    unique_lock<mutex> prev_lk(prev->m);
    Node *node = prev->next;
    unique_lock<mutex> node_lk(node->m);
    for (int i = 0; i < pos && node != tail_; i++) {
        prev = node;
        node = node->next;
        prev_lk.swap(node_lk);
        node_lk = unique_lock<mutex>(node->m);
    }
    new_node->next = node;
    prev->next = new_node.release();
}

Only two lines changed, but now it's obvious that we never leak memory...

---

...except in the destructor of LockBasedLinkedList, which (being defaulted) just leaks every node remaining in the list! Was that intentional, or an oversight?

---

Also consider the behavior of

LockBasedLinkedList a;
auto b = a;

This certainly doesn't do what you want. What do you want it to do? If the answer is "I want it not to compile," then you should =delete your copy operations:

LockBasedLinkedList(const LockBasedLinkedList&) = delete;
LockBasedLinkedList& operator=(const LockBasedLinkedList&) = delete;

(EDIT: Deleting your copy operations will automatically cause your move operations to not-get-implicitly-generated. The full rules are a mess: see here. Feel free to explicitly =delete your move operations too, if you want.)

---

In get:

    if (node == tail_) {
        return 0;
    }

How would your user distinguish the 0 that means "no such node" from the 0 that means "this node exists and has value 0"? I would strongly recommend designing your API so that this distinction is obvious to the user. For example:

std::optional<int> get(int);

---

How to return unique_lock in a function...?

I think this is related to your get API, yeah? That is, maybe you want something like

LockedNodePointer get(int);

class LockedNodePointer {
public:
    LockedNodePointer(std::nullptr_t) : ptr_(nullptr) {}
    Node& operator*() const { return *ptr_; }
    Node& operator->() const { return ptr_; }
private:
    friend class LockBasedLinkedList;
    using Lock = std::unique_lock<std::mutex>;
    explicit LockedNodePointer(Lock lk, Node *ptr) : lk_(lk), ptr_(ptr) {}
    Lock lk_;
    Node *ptr_;
};

This is an even cleaner way to distinguish "node with value 0" from "no such node."

---

It would also be interesting to consider: do you need a version of your get member function that is const-qualified? (Getting an element doesn't really modify the list, right?)

Do you need two overloads of get — one const and one non-const?