Правильно ли реализован lock-free битовый массив?

Question

[Василий Дёмин]

Есть битовый массив и несколько потоков, в данном случае два. Каждый поток получает номера операций, которые ему необходимо выполнить. Все потоки получают различные номера операций. Каждая из операций - массив типа int, оканчивающийся -1 и содержащий номера битов, которые необходимо установить в битовом массиве, если до установки бита он был неустановлен, то поток должен вывести номер данного бита. Вопрос: правильно ли реализован lock-free битовый массив в данном коде?

#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <stdatomic.h>
#include <threads.h>

#define BITS 2109       // Количество бит в битовом массиве
#define MAX_OPS 1000    // Максимальное количество операций
#define MAX_OP_SIZE 100 // Максимальный размер операции
static atomic_uchar bitset[(BITS>>3) + ((BITS&7) != 0)];

static int* ops[MAX_OPS];
static int num_ops;

// Структура для передачи потоку номеров операций, которые он должен выполнить
struct range { int from, to; };

// Функция инициализации массива операций случайными значениями
static inline void gen_ops() {
    num_ops = 300 + rand()%(MAX_OPS-300);
    for(int i = 0; i < num_ops; i++) {
        int op_size = 20 + rand()%(MAX_OP_SIZE-20);
        int *op = malloc(sizeof(int)*(op_size+1));
        if(!op) {
            perror("malloc");
            exit(EXIT_FAILURE);
        }
        for(int j = 0; j < op_size; j++)
            op[j] = rand()%BITS;
        op[op_size] = -1;
        ops[i] = op;
    }
}

static inline void free_ops() {
    for(int i = 0; i < num_ops; i++)
        free(ops[i]);
}

// Функция, выполняющаяся в потоках
static int worker(void *arg) {
    struct range *r = (struct range*)arg;
    for(int i = r->from; i < r->to; i++) {
        int *op = ops[i];
        for(;;) {
            int val = *op; // Получение номера бита, который необходимо установить
            if(val < 0)    // Если конец операции, выходим из цикла
                break;

            int or = 1 << (val&7);
            // Атомарно устанавливаем нужный бит и получаем предыдущее значение
            unsigned char byte = atomic_fetch_or(&bitset[val >> 3], or);
            if((byte & or) == 0) {
                // Если бит не был установлен до этого, выводим его номер
                printf("%d\n", val);
            }
            op++;
        }
    }
    return 0;
}

int main() {
    srand(time(NULL));
    for(int i = 0; i < sizeof(bitset); i++)
        atomic_init(&bitset[i], 0);

    gen_ops();

    thrd_t thr1, thr2;
    struct range r1 = {0, num_ops/2};
    struct range r2 = {r1.to, num_ops};
    thrd_create(&thr1, worker, &r1);
    thrd_create(&thr2, worker, &r2);
    thrd_join(thr1, NULL);
    thrd_join(thr2, NULL);

    free_ops();
}

Answer 1

[res2001]

Какого-то специфического lock-free алгоритма у вас в коде нет.
Достоточно для чтеня/записи использовать атомик операции и оно будет нормально работать, без гонки данных.