86 lines
3.0 KiB
C
86 lines
3.0 KiB
C
// ke_queue.c - ring-buffer message queue
|
|
#include "ke_queue.h"
|
|
#include "ke_thread.h"
|
|
#include "ke_sched.h"
|
|
#include "ob_object.h"
|
|
#include <string.h>
|
|
|
|
static void QueueDestroy(OBJECT* obj) { (void)obj; }
|
|
|
|
KSTATUS QueueCreate(QUEUE **out) {
|
|
if (!out) return STATUS_INVALID_PARAMETER;
|
|
QUEUE* q = (QUEUE *)ObAllocateObject(ObjectTypeQueue, sizeof(QUEUE));
|
|
if (!q) return STATUS_NO_MEMORY;
|
|
q->Header.Destroy = QueueDestroy;
|
|
q->Lock = (SPINLOCK)SPINLOCK_INIT;
|
|
q->Head = q->Tail = q->Count = 0;
|
|
LIST_INIT(&q->SendWaiters);
|
|
LIST_INIT(&q->RecvWaiters);
|
|
*out = q;
|
|
return STATUS_SUCCESS;
|
|
}
|
|
|
|
KSTATUS KeSendMessage(QUEUE* queue, const QUEUE_MESSAGE* msg, bool blocking) {
|
|
if (!queue || !msg) return STATUS_INVALID_PARAMETER;
|
|
THREAD* self = KeCurrentThread();
|
|
|
|
for (;;) {
|
|
uintptr_t flags = KeAcquireSpinlock(&queue->Lock);
|
|
if (queue->Count < QUEUE_DEPTH) {
|
|
queue->Ring[queue->Tail] = *msg;
|
|
queue->Tail = (queue->Tail + 1) % QUEUE_DEPTH;
|
|
queue->Count++;
|
|
|
|
// Wake a receiver if one is waiting.
|
|
if (!LIST_IS_EMPTY(&queue->RecvWaiters)) {
|
|
LIST_ENTRY* node = ListPopHead(&queue->RecvWaiters);
|
|
THREAD* t = CONTAINING_RECORD(node, THREAD, SchedulerLink);
|
|
KeReleaseSpinlock(&queue->Lock, flags);
|
|
KeUnblockThread(t);
|
|
} else {
|
|
KeReleaseSpinlock(&queue->Lock, flags);
|
|
}
|
|
return STATUS_SUCCESS;
|
|
}
|
|
if (!blocking) {
|
|
KeReleaseSpinlock(&queue->Lock, flags);
|
|
return STATUS_QUEUE_FULL;
|
|
}
|
|
ListInsertTail(&queue->SendWaiters, &self->SchedulerLink);
|
|
KeReleaseSpinlock(&queue->Lock, flags);
|
|
KeBlockThread(self, &queue->Header);
|
|
}
|
|
}
|
|
|
|
KSTATUS KeReceiveMessage(QUEUE* queue, QUEUE_MESSAGE* msg, bool blocking) {
|
|
if (!queue || !msg) return STATUS_INVALID_PARAMETER;
|
|
THREAD* self = KeCurrentThread();
|
|
|
|
for (;;) {
|
|
uintptr_t flags = KeAcquireSpinlock(&queue->Lock);
|
|
if (queue->Count > 0) {
|
|
*msg = queue->Ring[queue->Head];
|
|
queue->Head = (queue->Head + 1) % QUEUE_DEPTH;
|
|
queue->Count--;
|
|
|
|
// Wake a sender if one is waiting.
|
|
if (!LIST_IS_EMPTY(&queue->SendWaiters)) {
|
|
LIST_ENTRY* node = ListPopHead(&queue->SendWaiters);
|
|
THREAD* t = CONTAINING_RECORD(node, THREAD, SchedulerLink);
|
|
KeReleaseSpinlock(&queue->Lock, flags);
|
|
KeUnblockThread(t);
|
|
} else {
|
|
KeReleaseSpinlock(&queue->Lock, flags);
|
|
}
|
|
return STATUS_SUCCESS;
|
|
}
|
|
if (!blocking) {
|
|
KeReleaseSpinlock(&queue->Lock, flags);
|
|
return STATUS_QUEUE_EMPTY;
|
|
}
|
|
ListInsertTail(&queue->RecvWaiters, &self->SchedulerLink);
|
|
KeReleaseSpinlock(&queue->Lock, flags);
|
|
KeBlockThread(self, &queue->Header);
|
|
}
|
|
}
|