Quoi de mieux pour une file d'attente de messages? mutex & cond ou mutex et sémaphore?

Question

J'implémente une file d'attente de messages C++ basée sur une file d'attente std ::. Comme j'avais besoin que les popers attendent dans une file d'attente vide, j'envisageais d'utiliser mutex pour l'exclusion mutuelle et de suspendre les threads dans la file d'attente vide, comme le fait glib avec gasyncqueue.

Cependant, il me semble qu'un sémaphore mutex & ferait l'affaire, je pense qu'il contient un nombre entier et cela semble être un nombre assez élevé pour atteindre les messages en attente. Les avantages du sémaphore sont que vous n'avez pas besoin de vérifier manuellement la condition chaque fois que vous revenez d'une attente, car vous êtes maintenant sûr que quelqu'un a inséré quelque chose (quand quelqu'un a inséré 2 éléments et vous êtes le deuxième thread arrivant) .

Lequel choisiriez-vous?

EDIT: Changé la question en réponse à Rogers @ Greg

Answer 1

Une seule sémaphores ne pas faire le travail - vous devez être comparer (sémaphores mutex +) et (mutex + variable de condition).

Il est assez facile de voir cela en essayant de la mettre en œuvre:

void push(T t) 
{ 
    queue.push(t); 
    sem.post(); 
} 

T pop() 
{ 
    sem.wait(); 
    T t = queue.top(); 
    queue.pop(); 
    return t; 
} 

Comme vous pouvez le voir, il n'y a pas d'exclusion mutuelle lorsque vous êtes en train de lecture/écriture à la file d'attente, même si la signalisation (de le sémaphore) est là. Plusieurs threads peuvent appeler push en même temps et casser la file d'attente, ou plusieurs threads peuvent appeler pop en même temps et le casser. Ou, un thread pourrait appeler pop et supprimer le premier élément de la file d'attente alors qu'un autre thread appelé push.

Vous devriez utiliser ce que vous pensez être plus facile à implémenter, je doute que les performances varient beaucoup (il peut être intéressant de mesurer cependant).

Answer 2

Si vous voulez permettre à plusieurs utilisateurs simultanément à un moment d'utiliser la file d'attente, vous devez utiliser sémaphores.

sema(10) // ten threads/process have the concurrent access. 

sema_lock(&sema_obj) 
queue 
sema_unlock(&sema_obj) 

Mutex «n'autorise» qu'un seul utilisateur à la fois.

pthread_mutex_lock(&mutex_obj) 
global_data; 
pthread_mutex_unlock(&mutex_obj) 

C'est la principale différence et vous devez décider quelle solution répond à vos besoins. Mais je choisirais l'approche mutex, parce que vous n'avez pas besoin de spécifier combien d'utilisateurs peuvent saisir votre ressource.

Answer 3

Personnellement, j'utilise un mutex pour sérialiser l'accès à la liste, et réveiller le consommateur en envoyant un octet sur un socket (produit par socketpair()). Cela peut être un peu moins efficace qu'un sémaphore ou une variable de condition, mais cela a l'avantage de permettre au consommateur de bloquer dans select()/poll(). De cette façon, le consommateur peut également attendre autre chose que la file d'attente de données, s'il le souhaite. Il vous permet également d'utiliser exactement le même code de mise en file d'attente sur presque tous les systèmes d'exploitation, puisque pratiquement chaque système d'exploitation prend en charge l'API BSD sockets.

psuedocode suit:

// Called by the producer. Adds a data item to the queue, and sends a byte 
// on the socket to notify the consumer, if necessary 
void PushToQueue(const DataItem & di) 
{ 
    mutex.Lock(); 
    bool sendSignal = (queue.size() == 0); 
    queue.push_back(di); 
    mutex.Unlock(); 
    if (sendSignal) producerSocket.SendAByteNonBlocking(); 
} 

// Called by consumer after consumerSocket selects as ready-for-read 
// Returns true if (di) was written to, or false if there wasn't anything to read after all 
// Consumer should call this in a loop until it returns false, and then 
// go back to sleep inside select() to wait for further data from the producer 
bool PopFromQueue(DataItem & di) 
{ 
    consumerSocket.ReadAsManyBytesAsPossibleWithoutBlockingAndThrowThemAway(); 
    mutex.Lock(); 
    bool ret = (queue.size() > 0); 
    if (ret) queue.pop_front(di); 
    mutex.Unlock(); 
    return ret; 
}