Message exchange using AmigaOS-style message ports and message headers works only between threads. Messages are identified by and queued using direct pointers to the message headers, which are intrusively prepended to the intended message payload; additionally, message payloads frequently include pointers to ancillary data structures, basically rendering the msg_length field basically useless.
By making messages a proper kernel-managed resource, we can treat messages similarly to virtual files, exposing a precisely managed window into the sender’s address space. Messages encapsulated this way retain much of the desirable developer experience that the original AmigaOS is well known for. Although message payloads must now be copied, this approach reduces the number of copies to a minimum of just one.
NOTE. The text that follows is hypothetical and forward looking. The design contained herein is preliminary. It’s also written in C, even though the system call interfaces will be designed around the RISC-V assembly language.
To receive messages, a task must first create a message channel. The recommended process for creating a new message channel is to first allocate a signal (see AllocSignal()) for it to use when receiving a new message. Next, create the channel proper (see NewChannel()), setting optional attributes as desired by the application.
The following code illustrates how to create a new channel with a custom-assigned signal bit:
int errcode;
uint8_t my_sig_bit = AllocSignal(&errcode);
if(errcode != 0) {
// No more signal bits
}
ChannelAttrs chan_attrs = {
.sig_bit = my_sig_bit,
};
uint64_t chanID = NewChannel(&chan_attrs, CHATTR_SIGBIT, &errcode);
if(errcode != 0) {
// Channel creation failed for some reason
}If successful, this will return a number uniquely identifying the channel in the VM/OS environment. The channel will be configured to signal the task that created it with the signal bit specified.
NOTE. You could create the channel and leave CHATTR_SIGBIT clear, at which time VExec will try to allocate a sigbit on your behalf. However, to find out what bit it allocated, you’ll later need to query the channel’s attributes. You might as well just allocate the signal bit yourself and set it manually upon new channel creation. This also allows several different channels to share a common signal bit.
Once a channel is created, it can be published so that other tasks can now locate it by name.
chan_attrs.name = "ServerChannel";
chan_attrs.is_public = 1;
SetChanAttrs(chanID, &chan_attrs, CHATTR_NAME | CHATTR_ISPUBLIC, &errcode);
if(errcode != 0) {
// Channel cannot be made public for some reason.
}At this point, the server code can wait on the channel in the normal way.
uint64_t channel_sigmask = 1 << my_sig_bit;
// Assume this_mask and that_mask have already been initialized elsewhere.
sigs = Wait(channel_sigmask | this_mask | that_mask);
if(sigs & my_channel_event) {
handle_my_channel(chanID);
}
if(sigs & this_mask) {
handle_this_thing(...etc...);
}
if(sigs & that_mask) {
handle_that_thing(...etc...);
}Once we know that the channel has received an event, we know at least one message has arrived since we last checked its status. We can then grab and process messages off of it, one by one, until the channel has been drained.
void handle_my_channel(uint64_t chanID) {
MsgID msg;
int errcode;
for(;;) {
msg = GetChannelMsg(chanID, &errcode);
if(errcode == 0) {
handle_nonlocal_message(msg);
} else {
// Most likely, we just emptied the channel of messages.
break;
}
}
}We’ll discuss handling of messages in the next section.
You can dispose of a channel, assuming it has been completely emptied.
errcode = DisposeChannel(chanID);
if(errcode != 0) {
// Most likely, the channel isn't empty.
}So far, the basic work-flow is pretty much identical to in-process message ports and traditional messages, except that instead of pointers, we’re now dealing with opaque handles for the various data objects. The biggest difference, however, will come from processing messages; recall that, so far, we still don’t have access to the message’s content yet.
To recover the content of a message, you must read the message. Unlike classical AmigaOS, where you can just reference a message’s structure elements directly, we cannot do that in VM/OS. The message data still resides in the sender’s address space at this point.
To read a message, you can use ReadChanMsg(). This function is responsible for copying the message data out of the sender’s address space and directly into your own address space. Note that the message itself maintains a cursor, indicating from where it should start reading. When you first receive the message, the cursor is reset to offset zero.
void handle_nonlocal_message(uint64_t msg) {
struct MyMessageHeader header;
struct Body1 b1;
struct Body2 b2;
int errcode;
int64_t actual = ReadChanMsg(msg, &header, sizeof(header));
if(actual != sizeof(header)) {
// Message that was sent is too small even for a proper header.
goto msg_doone;
}
switch(header.request_code) {
case DO_REQUEST_1:
actual = ReadChanMsg(msg, &b1, sizeof(b1));
if(actual == sizeof(b1)) {
// Process request 1 here.
}
goto msg_done;
case DO_REQUEST_2:
actual = ReadChanMsg(msg, &b2, sizeof(b2));
if(actual == sizeof(b2)) {
// Process request 1 here.
}
goto msg_done;
default:
// Unknown message type; just reply to it.
}
msg_done:
errcode = ReplyChanMsg(msg);
if(errcode != 0) {
// Message reply failed; even so, it is no longer eligible for use
// by this task anymore.
}
}Of course, ReadChanMsg() assumes you can read the message sequentially, from start to end. That’s usually what you want to do; however, if you need to seek elsewhere and read, you can do so with ReadChanMsgAt(msg, buffer, size, offset). This sets the cursor to offset, and then falls into ReadChanMsg().
If you have data you wish to send back to the sender, which is frequently the case, you can do so with WriteChanMsg() and WriteChanMsgAt(), which behave similarly to their read-counterparts, except that it copies data back into the sender’s address space.
ReplyChanMsg() sends the message back to its sender, by way of its reply channel. NOTE: Once a message has been replied, it can no longer be read from or written to (unless the sender re-sends the message, of course).
Creating a message is somewhat similar to receiving one. You have a (set of) data structure(s), and then you must create a new message to represent the memory those structures occupy.
The NewMsgQuick() function is usually sufficient; it takes a single buffer and a reply channel ID to create a new message. However, if you have more complicated requirements, you can represent the structure using an I/O vector, which is used for scatter/gather operations.
For example, if we represent a Unix-style filesystem call as a message to a file server somewhere, we might see logic like so:
// Example implementation of write().
// I am going to skip error handling for brevity's sake.
size_t write(int fd, void *buffer, size_t length) {
struct WriteRequest wr = {
.fd = fd,
.req_length = length,
.actual_length = 0, // will be filled in later
.errcode = 0, // will be filled in later
};
// Because no data copies happen until the last possible moment,
// we can directly reference the supplied memory buffer in our IOVec.
// The file server will "see" this message as a single WriteRequest-
// prefixed buffer of length sizeof(wr)+length bytes.
struct io_vec iov[2] = {
[0] {.io_addr = &wr, .io_len = sizeof(wr) },
[1] {.io_addr = buffer, .io_len = length },
};
msg = NewMsg(our_reply_chan_id, &iov, 2, &errcode);
SendChanMsg(our_fileserver_id, msg);
// arm I/O response timeout timer here...
uint64_t sigs = Wait(our_reply_chan_mask | timer_mask);
if(sigs & our_reply_chan_mask) {
GetChanMsg(our_reply_chan_id, &errcode); // we know it's our msg.
DisposeMsg(msg);
errno = wr.errcode;
return wr.actual_length;
}
if(sigs & timer_mask) {
// I/O timeout happened; clean up and "cancel" the message, etc.
// Return EAGAIN or EINTR something.
}
}The following code is a more-or-less complete server implementation. Some details have been elided, both for brevity and because a C environment for VM/OS doesn’t yet exist. This code is preliminary and subject to change once concrete implementations are written.
enum Ops {
op_add, op_subtract, op_shutdown
};
struct MsgReq {
enum Ops function;
int64_t accumulator;
int64_t operand;
};
void
main(void) {
int errcode;
// Allocate a signal, so that we can be notified when
// a message arrives on our channel.
uint8_t server_sigbit = AllocSignal(&errcode);
if(errcode != 0) {
printf("Unable to allocate signal.\n");
goto end;
}
uint64_t server_channel_mask = 1 << server_sigbit;
// Create a public channel named TallyServer. Configure
// it so that, when a message arrives, we (this task) is
// notified via the signal we allocated above.
ChannelAttrs chattrs = { 0, };
chattrs.sigbit = server_sigbit;
chattrs.name = "TallyServer";
chattrs.is_public = true;
uint64_t chanID = NewChannel(
&chattrs,
// Set only these attrs; ignore or assume defaults for the rest.
CHATTRF_SIGBIT | CHATTRF_NAME | CHATTRF_ISPUBLIC,
&errcode
);
if(errcode != 0) {
printf("Unable to create public channel");
goto no_channel;
}
printf("Welcome to Tally Server. Now entering event loop.\n");
for(bool still_running = true; still_true; ) {
// We only need to wait on a single event source, so we ignore
// return code (it'll either be 0 or server_channel_mask).
Wait(server_channel_mask);
// If we're here, we know that at least one message is awaiting
// us on the server channel.
for(;;) {
msg = GetChanMsg(chanID, &errcode);
if(errcode == GCME_CHAN_EMPTY)
break;
if(errcode != 0) {
printf("Unable to GetChanMsg.");
goto no_chan_msg;
}
struct MsgReq mr;
uint64_t actual = ReadChanMsg(msg, &mr, sizeof(mr));
switch(mr.function) {
case op_shutdown:
still_running = false;
prinf("Someone is asking us to shutdown.\n");
break;
case op_add:
mr.accumulator += mr.operand;
break;
case op_subtract:
mr.accumulator -= mr.operand;
break;
}
// After calculating the results, return it to the sender.
// Note that we use WriteChanMsgAt() here because the message
// cursor is already set at the end of the message, thanks to
// the prior read.
actual = WriteChanMsgAt(msg, &mr, sizeof(mr), 0);
ReplyChanMsg(msg);
}
}
no_chan_msg:
DisposeChannel(chanID);
no_channel:
FreeSignal(server_sigbit);
end:
}enum Ops {
op_add, op_subtract, op_shutdown
};
struct MsgReq {
enum Ops function;
int64_t accumulator;
int64_t operand;
};
void
main(int argc, char* argv[]) {
// Create our reply channel. After we send a message,
// it will eventually appear on this channel.
uint8_t reply_sigbit = AllocSignal(&errcode);
if(errcode != 0) {
printf("Cannot allocate a signal.\n");
goto end;
}
uint64_t reply_mask = 1 << reply_sigbit;
struct ChannelAttrs chatters = { 0, };
chattrs.sig_bit = reply_sigbit;
uint64_t replyID = NewChannel(&chattrs, CHATTRF_SIGBIT, &errcode);
if(errcode != 0) {
printf("Unable to create reply channel.\n");
goto no_reply_channel;
}
// Locate the server; this is the program who will perform our
// arithmetic for us, since C just isn't up to the task. ;)
uint64_t serverID = FindChannel("TallyServer", &errcode);
if(errcode != 0) {
printf("Cannot find Tally Server channel\n");
goto bad_operation;
}
// Build the message we want to send.
struct MsgReq mr;
if(argc < 4) {
printf("Syntax: %s accumulator/N function/K operand/N\n", argv[0]);
goto bad_operation;
}
switch(argv[2][0]) {
case '+': mr.function = op_add; break;
case '-': mr.function = op_subtract; break;
case 'Q': mr.function = op_shutdown; break;
default:
printf("I don't know what operation you asked for.\n");
goto bad_operation;
}
mr.accumulator = atoi(argv[1]);
mr.operand = atoi(argv[3]);
// Send the message. When building the message, be sure to set the
// reply channel ID, so it knows where to go when it comes home!
uint64_t msg = NewMsgQuick(replyID, &mr, sizeof(mr), &errcode);
if(errcode != 0) {
printf("Cannot create message from simple buffer\n");
goto bad_operation;
}
SendChanMsg(serverID, msg);
// At this point, the message is away and queued onto the server's channel.
// NO DATA HAS BEEN COPIED YET. Thus, if you need to send more messages,
// you must create more buffers as they're needed. The memory _this_
// message refers to must remain stable until the receiver has copied the
// data for itself.
//
// We can now do whatever we want. For this simple example, we choose to
// just wait for a response by waiting on the reply port.
Wait(reply_mask);
// We complete the handshake by retrieving the sent message.
uint64_t reply_msg_id = GetChanMsg(replyID, &errcode);
if(errcode != 0) {
printf("Uh oh; something went wrong when collecting our reply.\n");
goto no_reply;
}
// Print out the results from our server.
printf("The result is: %lld\n", mr.accumulator);
// At this point, we could re-use the buffer/msg pair.
no_reply:
DisposeMsg(msg);
bad_operation:
DisposeChannel(replyID);
no_reply_channel:
FreeSignal(reply_sigbit);
end:
}