seahag/mq.c

127 lines
4 KiB
C

/* _,.---._ .-._ .--.-. ,--.--------.
* _,..---._ ,-.' , - `. /==/ \ .-._/==/ //==/, - , -\
* /==/, - \ /==/_, , - \|==|, \/ /, |==\ -\\==\.-. - ,-./
* |==| _ _\==| .=. |==|- \| | \==\- \`--`\==\- \
* |==| .=. |==|_ : ;=: - |==| , | -| `--`-' \==\_ \
* |==|,| | -|==| , '=' |==| - _ | |==|- |
* |==| '=' /\==\ - ,_ /|==| /\ , | |==|, |
* |==|-, _`/ '.='. - .' /==/, | |- | /==/ -/
* `-.`.____.' `--`--'' `--`./ `--` `--`--`
* _ __ ,---. .-._ .=-.-. _,.----.
* .-`.' ,`..--.' \ /==/ \ .-._ /==/_ /.' .' - \
* /==/, - \==\-/\ \ |==|, \/ /, /==|, |/==/ , ,-'
* |==| _ .=. /==/-|_\ | |==|- \| ||==| ||==|- | .
* |==| , '=',\==\, - \ |==| , | -||==|- ||==|_ `-' \
* |==|- '..'/==/ - ,| |==| - _ ||==| ,||==| _ , |
* |==|, | /==/- /\ - \|==| /\ , ||==|- |\==\. /
* /==/ - | \==\ _.\=\.-'/==/, | |- |/==/. / `-.`.___.-'
* `--`---' `--` `--`./ `--``--`-`
*
* @(#)Copyright (c) 2023, Nathan D. Fisher.
*
* This is free software. It comes with NO WARRANTY.
* Permission to use, modify and distribute this source code
* is granted subject to the following conditions.
* 1/ that the above copyright notice and this notice
* are preserved in all copies and that due credit be given
* to the author.
* 2/ that any changes to this code are clearly commented
* as such so that the author does not get blamed for bugs
* other than his own.
*/
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h> // free, malloc
#include "haggis.h"
#include "mq.h"
haggis_msg* haggis_msg_init(haggis_message_type tag, haggis_message_body body) {
haggis_msg *msg = calloc(1, sizeof(haggis_msg));
if (msg == NULL)
return NULL;
msg->next = msg->prev = NULL;
msg->tag = tag;
msg->body = body;
return msg;
}
void haggis_msg_deinit(haggis_msg *self) {
switch (self->tag) {
case DevNodeSkipped:
case NodeCreated:
case NodeExtracted:
free(self->body.f_name);
break;
case EndOfArchive:
case ArchiveError:
break;
}
free(self);
}
int haggis_mq_init(haggis_mq *self) {
int ret;
self->head = NULL;
self->tail = NULL;
ret = pthread_mutex_init(&self->mutex, NULL);
if (ret != 0)
return ret;
return pthread_cond_init(&self->cond, NULL);
}
int haggis_mq_push(haggis_mq *self, haggis_msg *msg) {
pthread_mutex_lock(&self->mutex);
if (self->tail == NULL) {
self->tail = self->head = msg;
} else {
msg->next = self->tail;
self->tail->prev = msg;
self->tail = msg;
}
self->count++;
pthread_cond_broadcast(&self->cond);
pthread_mutex_unlock(&self->mutex);
return 0;
}
haggis_msg* haggis_mq_pop(haggis_mq *self) {
haggis_msg *msg;
while (self->count == 0)
pthread_cond_wait(&self->cond, &self->mutex);
pthread_mutex_lock(&self->mutex);
self->count--;
msg = self->head;
if (msg->tag == EndOfArchive)
return msg;
if (self->tail == self->head) {
self->tail = self->head = NULL;
} else {
self->head = self->head->prev;
}
pthread_mutex_unlock(&self->mutex);
msg->prev = msg->next = NULL;
return msg;
}
haggis_msg* haggis_mq_try_pop(haggis_mq *self) {
haggis_msg *msg;
if (self->count == 0)
return NULL;
pthread_mutex_lock(&self->mutex);
self->count--;
msg = self->head;
if (msg->tag == EndOfArchive)
return msg;
if (self->tail == self->head) {
self->tail = self->head = NULL;
} else {
self->head = self->head->prev;
}
pthread_mutex_unlock(&self->mutex);
msg->prev = msg->next = NULL;
return msg;
}