std.mqueue

POSIX Message Queues (WP-11): Kernel-verwaltete IPC-Queues mit priorisiertem Zugriff. Nachrichten werden nach Priorität (0 = niedrigste, MQ_PRIO_MAX = 32767) sortiert und gepuffert. Queue-Namen beginnen mit / und erscheinen unter /dev/mqueue/. Verbindungen werden als Dateideskriptoren (int64) verwaltet.

→ Standard Library · std.pipe · std.thread

import std.mqueue;
import std.alloc;
import std.io;

fn Producer(): void {
    var fd := MqOpen("/worker_queue",
                     MQ_OPEN_WRONLY | MQ_OPEN_CREATE, 420);

    var msg: pchar := "job:render_frame";
    MqSend(fd, msg, 16, 10);  // Priorität 10

    MqClose(fd);
}

fn Consumer(): void {
    var fd := MqOpen("/worker_queue", MQ_OPEN_RDONLY, 0);

    var buf: int64 := alloc(MQ_DEFAULT_MSGSIZE);
    var prio: int64 := alloc(8);
    var n := MqRecv(fd, buf, MQ_DEFAULT_MSGSIZE, prio);

    PrintLn("Empfangen (prio=" + IntToStr(peek64(prio)) + "): " + buf);

    free(buf, MQ_DEFAULT_MSGSIZE);
    free(prio, 8);
    MqClose(fd);
    MqUnlink("/worker_queue");
}

import std.mqueue;
import std.alloc;

fn SetupQueue(): int64 {
    // Queue für bis zu 64 kleine Steuernachrichten (max. 64 Bytes)
    return MqOpenWith("/ctrl",
                      MQ_OPEN_RDWR | MQ_OPEN_CREATE,
                      420,   // 0644
                      64,    // maxmsg
                      64);   // msgsize
}

import std.mqueue;
import std.alloc;
import std.io;

fn PrioDispatch(): void {
    var fd := MqOpenWith("/prio_q",
                         MQ_OPEN_RDWR | MQ_OPEN_CREATE,
                         420, 32, 128);

    // Nachrichten mit verschiedenen Prioritäten senden
    var low:  pchar := "backup";
    var high: pchar := "alert";
    var mid:  pchar := "status";
    MqSend(fd, low,  7,  1);   // Priorität 1  — niedrig
    MqSend(fd, mid,  7,  5);   // Priorität 5  — mittel
    MqSend(fd, high, 6, 20);   // Priorität 20 — hoch

    // Empfang erfolgt in Prioritätsreihenfolge: high → mid → low
    var buf: int64 := alloc(128);
    var prio: int64 := alloc(8);
    var i: int64 := 0;
    while (i < 3) {
        MqRecv(fd, buf, 128, prio);
        PrintLn(buf + " (prio=" + IntToStr(peek64(prio)) + ")");
        i := i + 1;
    }

    free(buf, 128);
    free(prio, 8);
    MqClose(fd);
    MqUnlink("/prio_q");
}

import std.mqueue;
import std.time;
import std.alloc;
import std.io;

fn TryReceive(name: pchar): bool {
    var fd := MqOpen(name, MQ_OPEN_RDONLY, 0);

    // Absoluter Timeout: jetzt + 500 ms
    var now    := TimeNowNs();
    var timeout := now + 500000000;

    var buf: int64 := alloc(MQ_DEFAULT_MSGSIZE);
    var prio: int64 := alloc(8);
    var n := MqTimedRecv(fd, buf, MQ_DEFAULT_MSGSIZE, prio, timeout);

    MqClose(fd);
    free(buf, MQ_DEFAULT_MSGSIZE);
    free(prio, 8);

    if (n < 0) {
        PrintLn("Timeout oder Fehler");
        return false;
    }
    return true;
}

import std.mqueue;
import std.io;

fn PrintQueueStatus(name: pchar): void {
    var fd := MqOpen(name, MQ_OPEN_RDONLY, 0);

    PrintLn("Max. Nachrichten:     " + IntToStr(MqGetMaxMsg(fd)));
    PrintLn("Aktuelle Nachrichten: " + IntToStr(MqGetCurMsgs(fd)));

    MqClose(fd);
}

• Queue-Namen müssen mit / beginnen und dürfen keinen weiteren / enthalten.
• bufLen bei MqRecv/MqTimedRecv muss ≥ mq_msgsize der Queue sein — sonst Fehler −EMSGSIZE.
• Queues überleben den Prozess: MqUnlink muss explizit aufgerufen werden.
• Unter Linux liegt das Limit für mq_maxmsg standardmäßig bei 10 (sichtbar in /proc/sys/fs/mqueue/msg_max). Größere Kapazitäten erfordern Root oder angepasste Systemgrenzen.
• Für Timeout-Werte: TimeNowNs() (aus std.time) liefert Nanosekunden seit CLOCK_REALTIME, was direkt als timeoutNs verwendet werden kann.

• std.pipe — Pipes (einfacheres IPC für Eltern-/Kindprozesse)
• std.thread — Threads mit Mutex/Bedingungsvariablen
• std.signals — UNIX-Signale als leichtgewichtige IPC-Alternative
• std.net.epoll — epoll kann mqueue-fds überwachen

Letzte Aktualisierung: 2026-06-05