跳过列表中的细粒度锁定

Question

我正在尝试使用细粒度锁定机制在 c 中实现基于锁定的跳过列表。在运行代码时，应用的锁定机制似乎是粗粒度的。我已经在前面的节点中放置了锁，用于插入，使用在节点结构中定义的 pthread_mutex_t 锁变量，并在使用后释放它们。整个列表没有被锁定，只有节点被锁定，它似乎仍然在实现粗粒度锁定机制。下面提供了完成锁定机制的代码片段。是不是执行错了？

for(level = 0; valid && (level <=topLevel); level++){
            pred = preds[level];
            succ = succs[level];

            if(pred != prevPred){
                pthread_mutex_lock(pred -> lock);
                highestLocked   = level;
                prevPred        = pred;
            }

            valid = !(pred -> marked) && !(succ -> marked) && (pred -> next[level] == succ);
        }
        if(!valid){
            for(level = 0;level <= highestLocked; level++)
                pthread_mutex_unlock(preds[level] -> lock);
            continue;
        }
        newNode = createNewNode(x -> val, topLevel);
        for(level = 0; level <= topLevel; level++){
            newNode -> next[level] = succs[level];
            preds[level] -> next[level] = newNode;
        }
        newNode -> fullyLinked = 1;
        for(level = 0;level <= highestLocked; level++)
                pthread_mutex_unlock(preds[level] -> lock);

skiplist的编码遵循的论文是

@inproceedings{DBLP:dblp_conf/sirocco/HerlihyLLS07,
   author              = {Maurice Herlihy and 
                          Yossi Lev and 
                          Victor Luchangco and 
                          Nir Shavit},
   title               = {A Simple Optimistic Skiplist Algorithm.},
   booktitle           = {SIROCCO},
   year                = {2007},
   pages               = {124-138},
   ee                  = {http://dx.doi.org/10.1007/978-3-540-72951-8_11},
   crossref            = {2007},
}

编辑： 将节点插入skiplist的代码

int add(Node *x, int *preval){
    int lFound, highestLocked, valid, level;
    Node *nodeFound, *pred, *succ, *prevPred, *newNode;
    // int topLevel = randomLevel(MAX_LEVEL);
    int topLevel = (rand()%MAX_LEVEL)+1;
    *preval=topLevel;
    Node **preds = (Node **)malloc(sizeof(Node *) * (MAX_LEVEL + 1));//predecessor list
    Node **succs = (Node **)malloc(sizeof(Node *) * (MAX_LEVEL + 1));//successor list
    while(1){
        lFound = find(x, preds, succs);//gets predecessor and successor list of node where x to be inserted
        if(lFound != -1){
            nodeFound = succs[lFound];
            if(!nodeFound->marked){
                while(!nodeFound->fullyLinked){;}
                return 0;
            }
            continue;
        }
        highestLocked   = -1;
        prevPred        = NULL;
        valid           = 1;
        for(level = 0; valid && (level <=topLevel); level++){
            pred = preds[level];
            succ = succs[level];
            //locking the predecessor node level
            if(pred != prevPred){
                pthread_mutex_lock(pred -> lock);
                highestLocked   = level;
                prevPred        = pred;
            }

            valid = !(pred -> marked) && !(succ -> marked) && (pred -> next[level] == succ);
        }
        //releasing locked nodes
        if(!valid){
            for(level = 0;level <= highestLocked; level++)
                pthread_mutex_unlock(preds[level] -> lock);
            continue;
        }
        newNode = createNewNode(x -> val, topLevel);
        for(level = 0; level <= topLevel; level++){
            newNode -> next[level] = succs[level];
            preds[level] -> next[level] = newNode;
        }
        newNode -> fullyLinked = 1;
        //releasing locked nodes
        for(level = 0;level <= highestLocked; level++)
                pthread_mutex_unlock(preds[level] -> lock);
        return 1;
    }   
}

Answer 1

考虑一下您的“细粒度”锁定正在锁定哪些数据。您正在锁定对跳过列表内部列表的访问。任何线程都需要访问这些列表来搜索跳过列表。在 for 循环的第一次迭代中（假设 pred != prevPred 在这种情况下），您将锁定 pred->lock 级别 0。因此每个线程将尝试通过跳过列表同时获取相同的锁。

我建议寻找“多处理器编程的艺术”的副本，第 14 章谈到了 Skiplist。