Tampon circulaire sans verrouillage

Question

Je suis en train de concevoir un système qui se connecte à un ou plusieurs flux de données, effectue une analyse des données et déclenche des événements en fonction du résultat. Dans une configuration typique de producteur/consommateur multithread, j'aurai plusieurs threads de producteur mettant des données dans une file d'attente, et plusieurs threads de consommateur lisant les données, et les consommateurs sont seulement intéressés par le dernier point de données plus un nombre n de points. Les threads du producteur devront se bloquer si le consommateur lent ne peut pas suivre, et bien sûr les threads du consommateur se bloqueront lorsqu'il n'y aura pas de mises à jour non traitées. L'utilisation d'une file d'attente concurrente typique avec un verrou de lecteur/écrivain fonctionnera bien mais le taux de données entrant pourrait être énorme, donc je voulais réduire mes frais de verrouillage, en particulier les verrous d'écriture pour les producteurs. Je pense qu'un tampon circulaire sans verrou est ce dont j'avais besoin.

Maintenant, deux questions :

1. Le tampon circulaire sans verrou est-il la solution ?

2. Si c'est le cas, avant que je ne crée mon propre système, connaissez-vous une implémentation publique qui répondrait à mes besoins ?

Toute indication sur l'implémentation d'un tampon circulaire sans verrou est la bienvenue.

BTW, je fais ça en C++ sur Linux.

Quelques informations supplémentaires :

Le temps de réponse est critique pour mon système. Idéalement, les fils consommateurs voudront voir les mises à jour arriver le plus rapidement possible, car un retard supplémentaire d'une milliseconde pourrait rendre le système sans valeur, ou avec une valeur bien moindre.

L'idée de conception vers laquelle je penche est un tampon circulaire semi-libre où le thread producteur met les données dans le tampon aussi vite qu'il le peut, appelons-le la tête du tampon A, sans blocage jusqu'à ce que le tampon soit plein, lorsque A rencontre la fin du tampon Z. Les threads consommateurs détiendront chacun deux pointeurs vers le tampon circulaire, P et P _n où P est la tête de mémoire tampon locale du thread et P _n est le nième élément après P. Chaque thread de consommateur fera avancer ses P et P _n une fois qu'il a fini de traiter le P actuel et que le pointeur de fin de tampon Z est avancé avec le P le plus lent _n . Lorsque P rattrape A, ce qui signifie qu'il n'y a plus de nouvelle mise à jour à traiter, le consommateur tourne et attend que A avance à nouveau. Si le thread du consommateur tourne trop longtemps, il peut être mis en veille et attendre une variable de condition, mais je suis d'accord avec le fait que le consommateur prenne un cycle CPU en attendant la mise à jour car cela n'augmente pas ma latence (j'aurai plus de cœurs CPU que de threads). Imaginez que vous ayez une piste circulaire, et que le producteur soit devant un groupe de consommateurs, la clé est de régler le système de sorte que le producteur soit généralement juste un peu en avance sur les consommateurs, et la plupart de ces opérations peuvent être effectuées en utilisant des techniques sans verrou. Je comprends qu'il n'est pas facile d'obtenir les détails de l'implémentation correcte... ok, très difficile, c'est pourquoi je veux apprendre des erreurs des autres avant de faire les miennes.

Answer 1

Voici comment je m'y prendrais :

• mappe la file d'attente dans un tableau
• garder l'état avec un index de la prochaine lecture et de la prochaine écriture.
• garder un vecteur binaire vide et plein

L'insertion consiste à utiliser un CAS avec un incrément et un roll over sur l'écriture suivante. Une fois que vous avez un slot, ajoutez votre valeur puis mettez le bit vide/plein qui lui correspond.

Les retraits nécessitent une vérification du bit précédent pour tester les sous-débits, mais à part cela, c'est la même chose que pour l'écriture, mais en utilisant l'index de lecture et en effaçant le bit vide/plein.

Soyez avertis,

1. Je ne suis pas un expert dans ce domaine.
2. Les opérations ASM atomiques semblent être très lentes lorsque je les ai utilisées, donc si vous vous retrouvez avec plusieurs d'entre elles, il serait plus rapide d'utiliser des verrous intégrés dans les fonctions d'insertion/suppression. La théorie est qu'une seule opération atomique pour saisir le verrou suivie de (très) quelques opérations ASM non atomiques pourrait être plus rapide que la même chose effectuée par plusieurs opérations atomiques. Mais pour que cela fonctionne, il faudrait une mise en ligne manuelle ou automatique pour qu'il n'y ait qu'un seul petit bloc d'ASM.

Answer 2

Vous pouvez essayer lfqueue

Il est simple à utiliser, il s'agit d'une conception circulaire sans verrouillage.

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);

Answer 3

Dans certaines situations, il n'est pas nécessaire d'utiliser le verrouillage pour éviter les conditions de course, notamment lorsqu'il n'y a qu'un seul producteur et consommateur.

Considérez ce paragraphe du LDD3 :

Lorsqu'il est soigneusement mis en œuvre, un tampon circulaire ne nécessite aucun verrouillage en l'absence de producteurs ou de consommateurs multiples. Le producteur est le seul thread qui est autorisé à modifier l'index d'écriture et l'emplacement du tableau vers lequel il pointe. Tant que l'auteur stocke une nouvelle valeur dans le tampon avant de mettre à jour l'index d'écriture, le lecteur verra toujours une vue cohérente. Le lecteur, à son tour, est le seul thread qui peut accéder à l'index de lecture et à la valeur qu'il indique. En veillant à ce que les deux pointeurs ne s'écrasent pas l'un l'autre, le producteur et le consommateur peuvent accéder à la mémoire tampon simultanément sans conditions de concurrence.

Answer 4

Si vous considérez comme une condition préalable que le tampon ne sera jamais plein, envisagez d'utiliser cet algorithme sans verrou :

capacity must be a power of 2
buffer = new T[capacity] ~ on different cache line
mask = capacity - 1
write_index ~ on different cache line
read_index ~ on different cache line

enqueue:
    write_i = write_index.fetch_add(1) & mask
    buffer[write_i] = element ~ release store

dequeue:
    read_i = read_index.fetch_add(1) & mask
    element
    while ((element = buffer[read_i] ~ acquire load) == NULL) {
        spin loop
    }
    buffer[read_i] = NULL ~ relaxed store
    return element