[−][src]Crate posixmq
Posix message queue wrapper with optional mio integration.
Posix message queues are like pipes, but message-oriented which makes them
safe to read by multiple processes. Messages are sorted based on an
additional priority parameter. Queues are not placed in the normal file
system, but uses a separate, flat namespace. Normal file permissions still
apply though.
For a longer introduction, see man mq_overview
or man mq
.
They are not all that useful, as only Linux and some BSDs implement them, and even there you might be limited to creating queues with a capacity of no more than 10 messages at a time.
Examples
Send a couple messages:
use posixmq::PosixMq; // open the message queue if it exists, or create it if it doesn't. // names should start with a slash and have no more slashes. let mq = PosixMq::create("/hello_posixmq").unwrap(); mq.send(0, b"message").unwrap(); // messages with equal priority will be received in order mq.send(0, b"queue").unwrap(); // but this message has higher priority and will be received first mq.send(10, b"Hello,").unwrap();
and receive them:
use posixmq::PosixMq; // open the queue read-only, or fail if it doesn't exist. let mq = PosixMq::open("/hello_posixmq").unwrap(); // delete the message queue when you don't need to open it again. // otherwise it will remain until the system is rebooted, consuming posixmq::unlink("/hello_posixmq").unwrap(); // the receive buffer must be at least as big as the biggest possible message, // or you will not be allowed to receive anything. let mut buf = vec![0; mq.attributes().max_msg_len]; assert_eq!(mq.receive(&mut buf).unwrap(), (10, "Hello,".len())); assert_eq!(mq.receive(&mut buf).unwrap(), (0, "message".len())); assert_eq!(mq.receive(&mut buf).unwrap(), (0, "queue".len())); assert_eq!(&buf[..5], b"queue"); // check that there are no more messages assert_eq!(mq.attributes().current_messages, 0); // note that acting on this value is race-prone. A better way to do this // would be to switch our descriptor to non-blocking mode, and check for // an error of type `ErrorKind::WouldBlock`.
With mio (and features = ["mio"]
):
// set up queue let mq = posixmq::OpenOptions::readwrite() .nonblocking() .capacity(3) .max_msg_len(100) .create_new() .open("/mio") .unwrap(); let _ = posixmq::unlink("/mio"); // spawn a thread to send something let mq = Arc::new(mq); let sender = mq.clone(); // clones the Arc; let sender = thread::spawn(move|| sender.send(0, b"Hello").unwrap() ); // set up mio and register let poll = Poll::new().unwrap(); poll.register(&*mq, Token(0), Ready::readable(), PollOpt::edge()).unwrap(); let mut events = Events::with_capacity(10); poll.poll(&mut events, None).unwrap(); for event in &events { if event.token() == Token(0) { loop { let mut buf = [0; 100]; match mq.receive(&mut buf) { Err(ref e) if e.kind() == ErrorKind::WouldBlock => break, Err(e) => panic!("Error receiving message: {}", e), Ok((priority, len)) => { assert_eq!(priority, 0); assert_eq!(&buf[..len], b"Hello"); } } } } }
See the examples/ directory for more.
Portability
While the p in POSIX stands for Portable, that is not a fitting description of their message queues; Support is spotty even among *nix OSes. Windows, macOS, OpenBSD, Android, ios, Rumprun and Emscripten doesn't support posix message queues at all.
Compatible operating systems and features
Linux | FreeBSD 11+ | NetBSD | DragonFlyBSD | Illumos | Fuchsia | |
---|---|---|---|---|---|---|
core features | Yes | Yes | buggy | Yes | No | Yes |
mio Evented | Yes | Yes | useless | No | No | No |
Sync | Yes | No | Yes | No | No | Yes |
FromRawFd +IntoRawFd | Yes | No | Yes | No | No | Yes |
AsRawFd | Yes | Yes | Yes | No | No | Yes |
(is |set )_cloexec() | Yes | Yes | Yes | No | No | Yes |
Tested? | Yes, CI | Yes, CI | Manually | No | No | Cross-compiles |
This library will fail to compile if the target OS doesn't support posix message queues at all.
Feature explanations:
FromRawFd+IntoRawFd
: For this to compile, the innermqd_t
type must anint
typedef, and bad things might happen if it doesn't represent a file descriptor. These impls are currently on by default and only disabled when known not to work.Sync
: not auto-implemented whenmqd_t
is a pointer, and I haven't convinced myself it's safe even though I expect it to be. When it's just anint
fd I assume the OS will treat different threads equivalent to different processes.AsRawFd
: similar toFromRawFd
andIntoRawFd
, but FreeBSD 11+ has a function which lets one get a file descriptor for amqd_t
. This is required for querying or changing cloexec, and also for reliably setting it.- mio
Evented
: The impl requires bothAsRawFd
and that mio compiless. This does not guarantee that the polling mechanism used by mio supports posix message queues though.
On Linux, message queues and their permissions can be viewed in
/dev/mqueue/
. The kernel can be compiled to not support posix message
queues, so it's not guaranteed to always work. (sch as on Adroid)
On FreeBSD, the kernel module responsible for posix message queues
is not loaded by default; Run kldload mqueuefs
as root to enable it.
To list queues, the file system must additionally be mounted first:
mount -t mqueuefs null $somewhere
.
Versions before 11 do not have the function used to get a file descriptor,
so this library will not compile there.
While Illumos / Solaris support posix message queues, this libray won't work there because the libc crate doesn't have bindings.
OS-dependent restrictions and default values
Not even limiting oneself to the core features is enough to guarantee portability!
Linux | FreeBSD | NetBSD | |
---|---|---|---|
max priority | 32767 | 63 | 31 |
default capacity | 10 | 10 | 32 |
default max_msg_len | 8192 | 1024 | 992 |
max capacity | 10* | 100 | 512 |
max max_msg_len | 8192* | 16384 | 16384 |
allows empty messages | Yes | Yes | No |
enforces name rules | Yes | Yes | No |
allows "/." and "/.." | No | No | Yes |
On Linux the listed size limits only apply to unprivileged processes.
As root there instead appears to be a combined limit on memory usage of the
form capacity*(max_msg_len+k)
, but is several times higher than 10*8192.
Differences from the C API
send()
andreceive()
tries again when EINTR /ErrorKind::Interrupted
is returned. (Consistent with normal Rust io)- Descriptors are by default opened with O_CLOEXEC. (Consistent with normal Rust io)
open()
and all other methods which takeAsRef<[u8]>
prepends '/' to the name if missing. (They allocate anyway, to append a terminating '\0')
Missing and planned features
mq_timedsend()
andmq_timedreceive()
wrappers.try_clone()
Iterator
-implementing struct that callsreceive()
- Listing queues and their owners using OS-specific interfaces (such as /dev/mqueue/ on Linux)
- tmpfile equivalent
- Querying and possibly changing limits and default values
- Struct that deletes the message queue when dropped
- Test or check more platforms on CI
- Support more OSes?
mq_notify()
?
Please open an issue if you want any of them.
Structs
Attributes | Contains information about the capacities and state of a posix message queue. |
OpenOptions | Flags and parameters which control how a |
PosixMq | A descriptor for an open posix message queue. |
Functions
name_from_bytes | Helper function for converting a |
unlink | Delete a posix message queue. |
unlink_c | Delete a posix message queue, without inspecting |