Lockfree связанный список - PullRequest
Новая
       50

Lockfree связанный список

0 голосов
/ 10 мая 2018

Я реализую связанный список блокировок в C, аналогично тому, что доступно в linux kernel llist.h. Я использую атомарную операцию "__sync_bool_compare_and_swap".

Вот фрагмент кода: 

    struct llist_head {
    struct llist_node *head;
    struct llist_node *tail;
};

struct llist_node {
    struct llist_node *next;
    int mm_ref;
};

#define LLIST_HEAD_INIT(name)   { NULL, NULL }
#define LLIST_HEAD(name)    struct llist_head name = LLIST_HEAD_INIT(name)

void llist_insert_tail(struct llist_head *head, struct llist_node *new)
{
    volatile struct llist_node *last;
    new->mm_ref = 0;

    mb();
    new->next = NULL;
    do {
        last = head->tail;
        if (last)
            last->next = new;
    } while(!CAS(&(head->tail), last, new));
    CAS(&(head->head), NULL, new);
    mb();
}

struct llist_node *llist_remove_head(struct llist_head *head)
{
    volatile struct llist_node *first, *next;

    do {
        first = head->head;
        if (first != head->head)
            continue;
        if (first == NULL)
            return 0;
        next = first->next;
        printf("%s: tid=%lx first=%p next=%p first->next=%p\n", 
              __func__, pthread_self(), first, next, first->next);
    } while(!CAS(&(head->head), first, next));
    return first;
}

У меня есть небольшая многопоточная тестовая программа, есть только 1 поток производителя, чтобы вставить узел в хвостовой список, и два потока потребителей, чтобы удалить узел из заголовка списка. Функции вставки / удаления показаны ниже: 

    void *list_insert(void *data)
{
    int i;
    printf("producer: id=%lx\n", pthread_self());

    for ( i = 0 ; i < 10 ; i++) {
        struct my_list_node *e = malloc(sizeof(struct my_list_node));
        e->a = SOMEID;
        llist_insert_tail(&global_list, &(e_array[i]->list));
        printf("new node p=%p next=%p\n", e_array[i], e_array[i]->list.next);
        ATOMIC_ADD64(&cn_added, 1);
    }
    ATOMIC_SUB(&cn_producer, 1);
    printf("Producer thread [%lx] exited! Still %d running...\n",pthread_self(), cn_producer);
    return 0;
}

void *list_remove(void *data)
{
    struct llist_node *pos = NULL;
    printf("consumer: id=%lx\n", pthread_self());
    while(cn_producer || !llist_empty(&global_list)) {
        struct my_list_node *e;
        pos = llist_remove_head(&global_list);
        if (pos) {
            e = llist_entry(pos, struct my_list_node, list);
            printf("tid=%lx removed %p\n", pthread_self(), pos);
            if (e->a != SOMEID) {
                printf("data wrong!!\n");
                exit(1);
            }
            ATOMIC_ADD64(&cn_deleted, 1);
        } else {
            sched_yield();
        }
    }
    printf("Consumer thread [%lx] exited %d\n", pthread_self(), cn_producer);
    return 0;
}

Испытание, показывающее постоянный сбой, например, производитель вставил 10 узлов, но потребитель выдвинул только 1/2 или 3 узла, один из типичных результатов, которые я получил, показал ниже: 

consumer: id=7f4d469e8700
consumer: id=7f4d461e7700
producer: id=7f4d459e6700
new node p=0x7f4d400008c0 next=(nil)
new node p=0x7f4d400008e0 next=(nil)
new node p=0x7f4d40000900 next=(nil)
new node p=0x7f4d40000920 next=(nil)
new node p=0x7f4d40000940 next=(nil)
new node p=0x7f4d40000960 next=(nil)
new node p=0x7f4d40000980 next=(nil)
new node p=0x7f4d400009a0 next=(nil)
new node p=0x7f4d400009c0 next=(nil)
new node p=0x7f4d400009e0 next=(nil)
Producer thread [7f4d459e6700] exited! Still 0 running...
llist_remove_head: tid=7f4d469e8700 first=0x7f4d400008c0 next=(nil) first->next=(nil)
llist_remove_head: tid=7f4d469e8700 head=(nil)
tid=7f4d469e8700 removed 0x7f4d400008c0
Consumer thread [7f4d469e8700] exited 0
llist_remove_head: tid=7f4d461e7700 first=0x7f4d400008c0 next=(nil) first->next=(nil)
Consumer thread [7f4d461e7700] exited 0

Как видно, потоки производителя вышли первыми и переключились на потоки потребителя, однако эта строка: llist_remove_head: tid=7f4d469e8700 first=0x7f4d400008c0 next=(nil) first->next=(nil) показывая, что один из потоков-потребителей пытается удалить первый узел, но его следующий указатель указывает на NULL, что НЕ должно иметь место, поскольку к настоящему моменту список полностью заполнен (с 10 узлами). Таким образом, есть состояние гонки, которое должно произойти, так как это список без блокировки, но я почесал голову и не мог понять, какое состояние гонки может вызвать этот вывод. Эта реализация похожа на https://github.com/darkautism/lfqueue Но я не могу понять, почему моя версия не работает.

1 Ответ

0 голосов
/ 25 июля 2018

Вы можете попробовать lfqueue

Прост в использовании, круглая конструкция без блокировки 

    int *ret;

    lfqueue_t results;

    lfqueue_init(&results);

    /** Wrap This scope in multithread testing **/
    int_data = (int*) malloc(sizeof(int));
    assert(int_data != NULL);
    *int_data = i++;
    /*Enqueue*/
    while (lfqueue_enq(&results, int_data) != 1) ;

    /*Dequeue*/
    while ( (ret = lfqueue_deq(&results)) == NULL);

    // printf("%d\n", *(int*) ret );
    free(ret);
    /** End **/

    lfqueue_clear(&results);

Существует lfstack , который поставляется вместе с тем же разработчиком.

...