Re: [PATCH v4] xenbus: support large messages
Hello,
Sorry, it seems I missed that mail :/
Added my tag below.
BTW, I didn't see the mb() fix between rsp_cons+= and reading rsp_prod
on the Linux side?
Samuel
Juergen Gross, le mer. 24 nov. 2021 08:00:55 +0100, a ecrit:
> Ping?
>
> On 04.10.21 11:40, Juergen Gross wrote:
> > Today the implementation of the xenbus protocol in Mini-OS will only
> > allow to transfer the complete message to or from the ring page buffer.
> > This is limiting the maximum message size to lower values as the xenbus
> > protocol normally would allow.
> >
> > Change that by allowing to transfer the xenbus message in chunks as
> > soon as they are available.
> >
> > Avoid crashing Mini-OS in case of illegal data read from the ring
> > buffer.
> >
> > Signed-off-by: Juergen Gross
Reviewed-by: Samuel Thibault
> > ---
> > V2:
> > - drop redundant if (Samuel Thibault)
> > - move rmb() (Samuel Thibault)
> > V3:
> > - correct notification test (Samuel Thibault)
> > V4:
> > - more memory barriers (Samuel Thibault)
> > ---
> > xenbus/xenbus.c | 210
> > 1 file changed, 122 insertions(+), 88 deletions(-)
> >
> > diff --git a/xenbus/xenbus.c b/xenbus/xenbus.c
> > index 23de61e..b687678 100644
> > --- a/xenbus/xenbus.c
> > +++ b/xenbus/xenbus.c
> > @@ -29,6 +29,7 @@
> > #include
> > #include
> > #include
> > +#include
> > #define min(x,y) ({ \
> > typeof(x) tmpx = (x); \
> > @@ -46,6 +47,7 @@
> > static struct xenstore_domain_interface *xenstore_buf;
> > static DECLARE_WAIT_QUEUE_HEAD(xb_waitq);
> > DECLARE_WAIT_QUEUE_HEAD(xenbus_watch_queue);
> > +static __DECLARE_SEMAPHORE_GENERIC(xb_write_sem, 1);
> > xenbus_event_queue xenbus_events;
> > static struct watch {
> > @@ -231,75 +233,103 @@ char *xenbus_wait_for_state_change(const char* path,
> > XenbusState *state, xenbus_
> > }
> > +static void xenbus_read_data(char *buf, unsigned int len)
> > +{
> > +unsigned int off = 0;
> > +unsigned int prod, cons;
> > +unsigned int size;
> > +
> > +while (off != len)
> > +{
> > +wait_event(xb_waitq, xenstore_buf->rsp_prod !=
> > xenstore_buf->rsp_cons);
> > +
> > +prod = xenstore_buf->rsp_prod;
> > +cons = xenstore_buf->rsp_cons;
> > +DEBUG("Rsp_cons %d, rsp_prod %d.\n", cons, prod);
> > +size = min(len - off, prod - cons);
> > +
> > +rmb(); /* Make sure data read from ring is ordered with
> > rsp_prod. */
> > +memcpy_from_ring(xenstore_buf->rsp, buf + off,
> > + MASK_XENSTORE_IDX(cons), size);
> > +off += size;
> > +mb();/* memcpy() and rsp_cons update must not be reordered. */
> > +xenstore_buf->rsp_cons += size;
> > +mb();/* rsp_cons must be visible before we look at rsp_prod. */
> > +if (xenstore_buf->rsp_prod - cons >= XENSTORE_RING_SIZE)
> > +notify_remote_via_evtchn(xenbus_evtchn);
> > +}
> > +}
> > +
> > static void xenbus_thread_func(void *ign)
> > {
> > struct xsd_sockmsg msg;
> > -unsigned prod = xenstore_buf->rsp_prod;
> > +char *data;
> > for (;;) {
> > -wait_event(xb_waitq, prod != xenstore_buf->rsp_prod);
> > -while (1) {
> > -prod = xenstore_buf->rsp_prod;
> > -DEBUG("Rsp_cons %d, rsp_prod %d.\n", xenstore_buf->rsp_cons,
> > - xenstore_buf->rsp_prod);
> > -if (xenstore_buf->rsp_prod - xenstore_buf->rsp_cons <
> > sizeof(msg))
> > -break;
> > -rmb();
> > -memcpy_from_ring(xenstore_buf->rsp, ,
> > - MASK_XENSTORE_IDX(xenstore_buf->rsp_cons),
> > - sizeof(msg));
> > -DEBUG("Msg len %d, %d avail, id %d.\n", msg.len + sizeof(msg),
> > - xenstore_buf->rsp_prod - xenstore_buf->rsp_cons,
> > msg.req_id);
> > -
> > -if (xenstore_buf->rsp_prod - xenstore_buf->rsp_cons <
> > -sizeof(msg) + msg.len)
> > -break;
> > -
> > -DEBUG("Message is good.\n");
> > -
> > -if (msg.type == XS_WATCH_EVENT) {
> > -struct xenbus_event *event = malloc(sizeof(*event) +
> > msg.len);
> > -xenbus_event_queue *events = NULL;
> > -char *data = (char*)event + sizeof(*event);
> > -struct watch *watch;
> > -
> > -memcpy_from_ring(xenstore_buf->rsp, data,
> > -MASK_XENSTORE_IDX(xenstore_buf->rsp_cons +
> > sizeof(msg)),
> > -msg.len);
> > -
> > -event->path = data;
> > -event->token = event->path + strlen(event->path) + 1;
> > -
> > -mb();
> > -xenstore_buf->rsp_cons += msg.len + sizeof(msg);
> > -
> > -for (watch = watches; watch; watch = watch->next)
> > -if
Re: [PATCH v4] xenbus: support large messages
Ping?
On 04.10.21 11:40, Juergen Gross wrote:
Today the implementation of the xenbus protocol in Mini-OS will only
allow to transfer the complete message to or from the ring page buffer.
This is limiting the maximum message size to lower values as the xenbus
protocol normally would allow.
Change that by allowing to transfer the xenbus message in chunks as
soon as they are available.
Avoid crashing Mini-OS in case of illegal data read from the ring
buffer.
Signed-off-by: Juergen Gross
---
V2:
- drop redundant if (Samuel Thibault)
- move rmb() (Samuel Thibault)
V3:
- correct notification test (Samuel Thibault)
V4:
- more memory barriers (Samuel Thibault)
---
xenbus/xenbus.c | 210
1 file changed, 122 insertions(+), 88 deletions(-)
diff --git a/xenbus/xenbus.c b/xenbus/xenbus.c
index 23de61e..b687678 100644
--- a/xenbus/xenbus.c
+++ b/xenbus/xenbus.c
@@ -29,6 +29,7 @@
#include
#include
#include
+#include
#define min(x,y) ({ \
typeof(x) tmpx = (x); \
@@ -46,6 +47,7 @@
static struct xenstore_domain_interface *xenstore_buf;
static DECLARE_WAIT_QUEUE_HEAD(xb_waitq);
DECLARE_WAIT_QUEUE_HEAD(xenbus_watch_queue);
+static __DECLARE_SEMAPHORE_GENERIC(xb_write_sem, 1);
xenbus_event_queue xenbus_events;
static struct watch {
@@ -231,75 +233,103 @@ char *xenbus_wait_for_state_change(const char* path,
XenbusState *state, xenbus_
}
+static void xenbus_read_data(char *buf, unsigned int len)
+{
+unsigned int off = 0;
+unsigned int prod, cons;
+unsigned int size;
+
+while (off != len)
+{
+wait_event(xb_waitq, xenstore_buf->rsp_prod != xenstore_buf->rsp_cons);
+
+prod = xenstore_buf->rsp_prod;
+cons = xenstore_buf->rsp_cons;
+DEBUG("Rsp_cons %d, rsp_prod %d.\n", cons, prod);
+size = min(len - off, prod - cons);
+
+rmb(); /* Make sure data read from ring is ordered with rsp_prod. */
+memcpy_from_ring(xenstore_buf->rsp, buf + off,
+ MASK_XENSTORE_IDX(cons), size);
+off += size;
+mb();/* memcpy() and rsp_cons update must not be reordered. */
+xenstore_buf->rsp_cons += size;
+mb();/* rsp_cons must be visible before we look at rsp_prod. */
+if (xenstore_buf->rsp_prod - cons >= XENSTORE_RING_SIZE)
+notify_remote_via_evtchn(xenbus_evtchn);
+}
+}
+
static void xenbus_thread_func(void *ign)
{
struct xsd_sockmsg msg;
-unsigned prod = xenstore_buf->rsp_prod;
+char *data;
for (;;) {
-wait_event(xb_waitq, prod != xenstore_buf->rsp_prod);
-while (1) {
-prod = xenstore_buf->rsp_prod;
-DEBUG("Rsp_cons %d, rsp_prod %d.\n", xenstore_buf->rsp_cons,
- xenstore_buf->rsp_prod);
-if (xenstore_buf->rsp_prod - xenstore_buf->rsp_cons < sizeof(msg))
-break;
-rmb();
-memcpy_from_ring(xenstore_buf->rsp, ,
- MASK_XENSTORE_IDX(xenstore_buf->rsp_cons),
- sizeof(msg));
-DEBUG("Msg len %d, %d avail, id %d.\n", msg.len + sizeof(msg),
- xenstore_buf->rsp_prod - xenstore_buf->rsp_cons, msg.req_id);
-
-if (xenstore_buf->rsp_prod - xenstore_buf->rsp_cons <
-sizeof(msg) + msg.len)
-break;
-
-DEBUG("Message is good.\n");
-
-if (msg.type == XS_WATCH_EVENT) {
-struct xenbus_event *event = malloc(sizeof(*event) + msg.len);
-xenbus_event_queue *events = NULL;
-char *data = (char*)event + sizeof(*event);
-struct watch *watch;
-
-memcpy_from_ring(xenstore_buf->rsp, data,
-MASK_XENSTORE_IDX(xenstore_buf->rsp_cons + sizeof(msg)),
-msg.len);
-
-event->path = data;
-event->token = event->path + strlen(event->path) + 1;
-
-mb();
-xenstore_buf->rsp_cons += msg.len + sizeof(msg);
-
-for (watch = watches; watch; watch = watch->next)
-if (!strcmp(watch->token, event->token)) {
-events = watch->events;
-break;
-}
-
-if (events) {
-event->next = *events;
-*events = event;
-wake_up(_watch_queue);
-} else {
-printk("unexpected watch token %s\n", event->token);
-free(event);
+xenbus_read_data((char *), sizeof(msg));
+DEBUG("Msg len %d, %d avail, id %d.\n", msg.len + sizeof(msg),
+ xenstore_buf->rsp_prod - xenstore_buf->rsp_cons, msg.req_id);
+
+if (msg.len > XENSTORE_PAYLOAD_MAX) {
+printk("Xenstore violates protocol, message longer
[PATCH v4] xenbus: support large messages
Today the implementation of the xenbus protocol in Mini-OS will only
allow to transfer the complete message to or from the ring page buffer.
This is limiting the maximum message size to lower values as the xenbus
protocol normally would allow.
Change that by allowing to transfer the xenbus message in chunks as
soon as they are available.
Avoid crashing Mini-OS in case of illegal data read from the ring
buffer.
Signed-off-by: Juergen Gross
---
V2:
- drop redundant if (Samuel Thibault)
- move rmb() (Samuel Thibault)
V3:
- correct notification test (Samuel Thibault)
V4:
- more memory barriers (Samuel Thibault)
---
xenbus/xenbus.c | 210
1 file changed, 122 insertions(+), 88 deletions(-)
diff --git a/xenbus/xenbus.c b/xenbus/xenbus.c
index 23de61e..b687678 100644
--- a/xenbus/xenbus.c
+++ b/xenbus/xenbus.c
@@ -29,6 +29,7 @@
#include
#include
#include
+#include
#define min(x,y) ({ \
typeof(x) tmpx = (x); \
@@ -46,6 +47,7 @@
static struct xenstore_domain_interface *xenstore_buf;
static DECLARE_WAIT_QUEUE_HEAD(xb_waitq);
DECLARE_WAIT_QUEUE_HEAD(xenbus_watch_queue);
+static __DECLARE_SEMAPHORE_GENERIC(xb_write_sem, 1);
xenbus_event_queue xenbus_events;
static struct watch {
@@ -231,75 +233,103 @@ char *xenbus_wait_for_state_change(const char* path,
XenbusState *state, xenbus_
}
+static void xenbus_read_data(char *buf, unsigned int len)
+{
+unsigned int off = 0;
+unsigned int prod, cons;
+unsigned int size;
+
+while (off != len)
+{
+wait_event(xb_waitq, xenstore_buf->rsp_prod != xenstore_buf->rsp_cons);
+
+prod = xenstore_buf->rsp_prod;
+cons = xenstore_buf->rsp_cons;
+DEBUG("Rsp_cons %d, rsp_prod %d.\n", cons, prod);
+size = min(len - off, prod - cons);
+
+rmb(); /* Make sure data read from ring is ordered with rsp_prod. */
+memcpy_from_ring(xenstore_buf->rsp, buf + off,
+ MASK_XENSTORE_IDX(cons), size);
+off += size;
+mb();/* memcpy() and rsp_cons update must not be reordered. */
+xenstore_buf->rsp_cons += size;
+mb();/* rsp_cons must be visible before we look at rsp_prod. */
+if (xenstore_buf->rsp_prod - cons >= XENSTORE_RING_SIZE)
+notify_remote_via_evtchn(xenbus_evtchn);
+}
+}
+
static void xenbus_thread_func(void *ign)
{
struct xsd_sockmsg msg;
-unsigned prod = xenstore_buf->rsp_prod;
+char *data;
for (;;) {
-wait_event(xb_waitq, prod != xenstore_buf->rsp_prod);
-while (1) {
-prod = xenstore_buf->rsp_prod;
-DEBUG("Rsp_cons %d, rsp_prod %d.\n", xenstore_buf->rsp_cons,
- xenstore_buf->rsp_prod);
-if (xenstore_buf->rsp_prod - xenstore_buf->rsp_cons < sizeof(msg))
-break;
-rmb();
-memcpy_from_ring(xenstore_buf->rsp, ,
- MASK_XENSTORE_IDX(xenstore_buf->rsp_cons),
- sizeof(msg));
-DEBUG("Msg len %d, %d avail, id %d.\n", msg.len + sizeof(msg),
- xenstore_buf->rsp_prod - xenstore_buf->rsp_cons, msg.req_id);
-
-if (xenstore_buf->rsp_prod - xenstore_buf->rsp_cons <
-sizeof(msg) + msg.len)
-break;
-
-DEBUG("Message is good.\n");
-
-if (msg.type == XS_WATCH_EVENT) {
-struct xenbus_event *event = malloc(sizeof(*event) + msg.len);
-xenbus_event_queue *events = NULL;
-char *data = (char*)event + sizeof(*event);
-struct watch *watch;
-
-memcpy_from_ring(xenstore_buf->rsp, data,
-MASK_XENSTORE_IDX(xenstore_buf->rsp_cons + sizeof(msg)),
-msg.len);
-
-event->path = data;
-event->token = event->path + strlen(event->path) + 1;
-
-mb();
-xenstore_buf->rsp_cons += msg.len + sizeof(msg);
-
-for (watch = watches; watch; watch = watch->next)
-if (!strcmp(watch->token, event->token)) {
-events = watch->events;
-break;
-}
-
-if (events) {
-event->next = *events;
-*events = event;
-wake_up(_watch_queue);
-} else {
-printk("unexpected watch token %s\n", event->token);
-free(event);
+xenbus_read_data((char *), sizeof(msg));
+DEBUG("Msg len %d, %d avail, id %d.\n", msg.len + sizeof(msg),
+ xenstore_buf->rsp_prod - xenstore_buf->rsp_cons, msg.req_id);
+
+if (msg.len > XENSTORE_PAYLOAD_MAX) {
+printk("Xenstore violates protocol, message longer than
allowed.\n");
+return;
+}
+
+if (msg.type
