mirror of
https://github.com/justinian/jsix.git
synced 2025-12-10 00:14:32 -08:00
[kernel] Fix mailbox bugs
This commit contains a number of related mailbox issues: - Add extra parameters to mailbox_respond_receive to allow both the number of bytes/handles passed in, and the size of the byte/handle buffers to be passed in. - Don't delete mailbox messages on receipt if the caller is waiting on reply - Correctly pass status messages along with a mailbox::replyer object - Actually wake the calling thread in the mailbox::replyer dtor - Make sure to release locks _before_ calling thread::wake() on blocked threads, as that may cause them to be scheduled ahead of the current thread.
This commit is contained in:
@@ -41,6 +41,7 @@ event::wake_observer()
|
||||
uint64_t value = read();
|
||||
if (value) {
|
||||
m_queue.pop_next_unlocked();
|
||||
lock.release();
|
||||
t->wake(value);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,7 +11,8 @@ namespace obj {
|
||||
static_assert(mailbox::message::slab_size % sizeof(mailbox::message) == 0,
|
||||
"mailbox message size does not fit cleanly into N pages.");
|
||||
|
||||
constexpr uint64_t no_message = 1;
|
||||
constexpr uint64_t no_message = 0;
|
||||
constexpr uint64_t has_message = 1;
|
||||
|
||||
mailbox::mailbox() :
|
||||
kobject(kobject::type::mailbox),
|
||||
@@ -52,26 +53,31 @@ mailbox::send(message *msg)
|
||||
m_messages.push_back(msg);
|
||||
|
||||
thread *t = m_queue.pop_next();
|
||||
if (t) t->wake();
|
||||
|
||||
lock.release();
|
||||
if (t) t->wake(has_message);
|
||||
}
|
||||
|
||||
bool
|
||||
mailbox::call(message *msg)
|
||||
{
|
||||
uint16_t reply_tag = next_reply_tag();
|
||||
|
||||
util::scoped_lock lock {m_message_lock};
|
||||
|
||||
if (!++m_next_reply_tag) ++m_next_reply_tag;
|
||||
msg->reply_tag = m_next_reply_tag;
|
||||
|
||||
msg->reply_tag = reply_tag;
|
||||
thread ¤t = thread::current();
|
||||
m_pending.insert(m_next_reply_tag, {¤t, msg});
|
||||
m_pending.insert(reply_tag, {¤t, msg});
|
||||
|
||||
m_messages.push_back(msg);
|
||||
|
||||
thread *t = m_queue.pop_next();
|
||||
if (t) t->wake();
|
||||
|
||||
return (current.block() != no_message);
|
||||
lock.release();
|
||||
if (t) t->wake(has_message);
|
||||
|
||||
uint64_t result = current.block();
|
||||
return result == has_message;
|
||||
}
|
||||
|
||||
bool
|
||||
@@ -113,11 +119,13 @@ mailbox::reply(uint16_t reply_tag)
|
||||
message *msg = p->msg;
|
||||
m_pending.erase(reply_tag);
|
||||
|
||||
return {msg, caller};
|
||||
return {msg, caller, has_message};
|
||||
}
|
||||
|
||||
mailbox::replyer::~replyer()
|
||||
{
|
||||
if (caller)
|
||||
caller->wake(status);
|
||||
}
|
||||
|
||||
} // namespace obj
|
||||
|
||||
@@ -71,6 +71,10 @@ public:
|
||||
replyer reply(uint16_t reply_tag);
|
||||
|
||||
private:
|
||||
inline uint16_t next_reply_tag() {
|
||||
return (__atomic_add_fetch(&m_next_reply_tag, 1, __ATOMIC_SEQ_CST) << 1) | 1;
|
||||
}
|
||||
|
||||
bool m_closed;
|
||||
uint16_t m_next_reply_tag;
|
||||
|
||||
@@ -104,14 +108,14 @@ struct mailbox::message :
|
||||
class mailbox::replyer
|
||||
{
|
||||
public:
|
||||
replyer() : msg {nullptr}, caller {nullptr} {}
|
||||
replyer(mailbox::message *m, thread *c) : msg {m}, caller {c} {}
|
||||
replyer(replyer &&o) : msg {o.msg}, caller {o.caller} {
|
||||
o.msg = nullptr; o.caller = nullptr;
|
||||
replyer() : msg {nullptr}, caller {nullptr}, status {0} {}
|
||||
replyer(mailbox::message *m, thread *c, uint64_t s) : msg {m}, caller {c}, status {s} {}
|
||||
replyer(replyer &&o) : msg {o.msg}, caller {o.caller}, status {o.status} {
|
||||
o.msg = nullptr; o.caller = nullptr; o.status = 0;
|
||||
}
|
||||
|
||||
replyer & operator=(replyer &&o) {
|
||||
msg = o.msg; caller = o.caller;
|
||||
msg = o.msg; caller = o.caller; status = o.status;
|
||||
o.msg = nullptr; o.caller = nullptr;
|
||||
return *this;
|
||||
}
|
||||
|
||||
@@ -47,7 +47,7 @@ prep_send(
|
||||
|
||||
msg->handle_count = handle_count;
|
||||
for (unsigned i = 0; i < handle_count; ++i) {
|
||||
handle *h = get_handle<kobject>(handles[i]);
|
||||
handle const *h = get_handle<kobject>(handles[i]);
|
||||
if (!h)
|
||||
return j6_err_invalid_arg;
|
||||
msg->handles[i] = *h;
|
||||
@@ -78,10 +78,10 @@ prep_receive(
|
||||
|
||||
*handle_count = msg->handle_count;
|
||||
process &proc = process::current();
|
||||
for (size_t i = 0; i < msg->handle_count; ++i)
|
||||
for (size_t i = 0; i < msg->handle_count; ++i) {
|
||||
handles[i] = proc.add_handle(msg->handles[i]);
|
||||
|
||||
delete msg;
|
||||
msg->handles[i] = {};
|
||||
}
|
||||
}
|
||||
|
||||
j6_status_t
|
||||
@@ -142,6 +142,8 @@ mailbox_receive(
|
||||
data, data_len,
|
||||
handles, handle_count);
|
||||
|
||||
if (*reply_tag == 0)
|
||||
delete msg;
|
||||
return j6_status_ok;
|
||||
}
|
||||
|
||||
@@ -217,17 +219,23 @@ mailbox_respond_receive(
|
||||
uint64_t * tag,
|
||||
void * data,
|
||||
size_t * data_len,
|
||||
size_t data_in,
|
||||
j6_handle_t * handles,
|
||||
size_t * handle_count,
|
||||
size_t handles_in,
|
||||
uint16_t * reply_tag,
|
||||
uint64_t * badge,
|
||||
uint64_t flags)
|
||||
{
|
||||
j6_status_t s = mailbox_respond(self, *tag, data, *data_len, handles, *handle_count, *reply_tag);
|
||||
j6_status_t s = mailbox_respond(self, *tag, data, data_in, handles, handles_in, *reply_tag);
|
||||
if (s != j6_status_ok)
|
||||
return s;
|
||||
|
||||
return mailbox_receive(self, tag, data, data_len, handles, handle_count, reply_tag, badge, flags);
|
||||
s = mailbox_receive(self, tag, data, data_len, handles, handle_count, reply_tag, badge, flags);
|
||||
if (s != j6_status_ok)
|
||||
return s;
|
||||
|
||||
return j6_status_ok;
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user