[PATCH v7 0/3] Add a vhost RPMsg API

[PATCH v7 0/3] Add a vhost RPMsg API

[Guennadi Liakhovetski]

Hi,

Next update:

v7:
- remove documentation update to be send separately
- address comments from Mathieu Poirier (thanks)

v6:
- rename include/linux/virtio_rpmsg.h -> include/linux/rpmsg/virtio.h

v5:
- don't hard-code message layout

v4:
- add endianness conversions to comply with the VirtIO standard

v3:
- address several checkpatch warnings
- address comments from Mathieu Poirier

v2:
- update patch #5 with a correct vhost_dev_init() prototype
- drop patch #6 - it depends on a different patch, that is currently
  an RFC
- address comments from Pierre-Louis Bossart:
  * remove "default n" from Kconfig

Linux supports RPMsg over VirtIO for "remote processor" / AMP use
cases. It can however also be used for virtualisation scenarios,
e.g. when using KVM to run Linux on both the host and the guests.
This patch set adds a wrapper API to facilitate writing vhost
drivers for such RPMsg-based solutions. The first use case is an
audio DSP virtualisation project, currently under development, ready
for review and submission, available at
https://github.com/thesofproject/linux/pull/1501/commits

Thanks
Guennadi

Guennadi Liakhovetski (3):
  vhost: convert VHOST_VSOCK_SET_RUNNING to a generic ioctl
  rpmsg: move common structures and defines to headers
  vhost: add an RPMsg API

 drivers/rpmsg/virtio_rpmsg_bus.c |  78 +------
 drivers/vhost/Kconfig            |   7 +
 drivers/vhost/Makefile           |   3 +
 drivers/vhost/rpmsg.c            | 370 +++++++++++++++++++++++++++++++
 drivers/vhost/vhost_rpmsg.h      |  74 +++++++
 include/linux/rpmsg/virtio.h     |  83 +++++++
 include/uapi/linux/rpmsg.h       |   3 +
 include/uapi/linux/vhost.h       |   4 +-
 8 files changed, 545 insertions(+), 77 deletions(-)
 create mode 100644 drivers/vhost/rpmsg.c
 create mode 100644 drivers/vhost/vhost_rpmsg.h
 create mode 100644 include/linux/rpmsg/virtio.h

-- 
2.28.0

[PATCH v7 0/3] Add a vhost RPMsg API

[Guennadi Liakhovetski]

Hi,

Next update:

v7:
- remove documentation update to be send separately
- address comments from Mathieu Poirier (thanks)

v6:
- rename include/linux/virtio_rpmsg.h -> include/linux/rpmsg/virtio.h

v5:
- don't hard-code message layout

v4:
- add endianness conversions to comply with the VirtIO standard

v3:
- address several checkpatch warnings
- address comments from Mathieu Poirier

v2:
- update patch #5 with a correct vhost_dev_init() prototype
- drop patch #6 - it depends on a different patch, that is currently
  an RFC
- address comments from Pierre-Louis Bossart:
  * remove "default n" from Kconfig

Linux supports RPMsg over VirtIO for "remote processor" / AMP use
cases. It can however also be used for virtualisation scenarios,
e.g. when using KVM to run Linux on both the host and the guests.
This patch set adds a wrapper API to facilitate writing vhost
drivers for such RPMsg-based solutions. The first use case is an
audio DSP virtualisation project, currently under development, ready
for review and submission, available at
https://github.com/thesofproject/linux/pull/1501/commits

Thanks
Guennadi

Guennadi Liakhovetski (3):
  vhost: convert VHOST_VSOCK_SET_RUNNING to a generic ioctl
  rpmsg: move common structures and defines to headers
  vhost: add an RPMsg API

 drivers/rpmsg/virtio_rpmsg_bus.c |  78 +------
 drivers/vhost/Kconfig            |   7 +
 drivers/vhost/Makefile           |   3 +
 drivers/vhost/rpmsg.c            | 370 +++++++++++++++++++++++++++++++
 drivers/vhost/vhost_rpmsg.h      |  74 +++++++
 include/linux/rpmsg/virtio.h     |  83 +++++++
 include/uapi/linux/rpmsg.h       |   3 +
 include/uapi/linux/vhost.h       |   4 +-
 8 files changed, 545 insertions(+), 77 deletions(-)
 create mode 100644 drivers/vhost/rpmsg.c
 create mode 100644 drivers/vhost/vhost_rpmsg.h
 create mode 100644 include/linux/rpmsg/virtio.h

-- 
2.28.0

_______________________________________________
Virtualization mailing list
Virtualization@lists.linux-foundation.org
https://lists.linuxfoundation.org/mailman/listinfo/virtualization

[PATCH v7 1/3] vhost: convert VHOST_VSOCK_SET_RUNNING to a generic ioctl

[Guennadi Liakhovetski]

VHOST_VSOCK_SET_RUNNING is used by the vhost vsock driver to perform
crucial VirtQueue initialisation, like assigning .private fields and
calling vhost_vq_init_access(), and clean up. However, this ioctl is
actually extremely useful for any vhost driver, that doesn't have a
side channel to inform it of a status change, e.g. upon a guest
reboot. This patch makes that ioctl generic, while preserving its
numeric value and also keeping the original alias.

Signed-off-by: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
---
 include/uapi/linux/vhost.h | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/include/uapi/linux/vhost.h b/include/uapi/linux/vhost.h
index 75232185324a..11a4948b6216 100644
--- a/include/uapi/linux/vhost.h
+++ b/include/uapi/linux/vhost.h
@@ -97,6 +97,8 @@
 #define VHOST_SET_BACKEND_FEATURES _IOW(VHOST_VIRTIO, 0x25, __u64)
 #define VHOST_GET_BACKEND_FEATURES _IOR(VHOST_VIRTIO, 0x26, __u64)
 
+#define VHOST_SET_RUNNING _IOW(VHOST_VIRTIO, 0x61, int)
+
 /* VHOST_NET specific defines */
 
 /* Attach virtio net ring to a raw socket, or tap device.
@@ -118,7 +120,7 @@
 /* VHOST_VSOCK specific defines */
 
 #define VHOST_VSOCK_SET_GUEST_CID	_IOW(VHOST_VIRTIO, 0x60, __u64)
-#define VHOST_VSOCK_SET_RUNNING		_IOW(VHOST_VIRTIO, 0x61, int)
+#define VHOST_VSOCK_SET_RUNNING		VHOST_SET_RUNNING
 
 /* VHOST_VDPA specific defines */
 
-- 
2.28.0

[PATCH v7 1/3] vhost: convert VHOST_VSOCK_SET_RUNNING to a generic ioctl

[Guennadi Liakhovetski]

VHOST_VSOCK_SET_RUNNING is used by the vhost vsock driver to perform
crucial VirtQueue initialisation, like assigning .private fields and
calling vhost_vq_init_access(), and clean up. However, this ioctl is
actually extremely useful for any vhost driver, that doesn't have a
side channel to inform it of a status change, e.g. upon a guest
reboot. This patch makes that ioctl generic, while preserving its
numeric value and also keeping the original alias.

Signed-off-by: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
---
 include/uapi/linux/vhost.h | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/include/uapi/linux/vhost.h b/include/uapi/linux/vhost.h
index 75232185324a..11a4948b6216 100644
--- a/include/uapi/linux/vhost.h
+++ b/include/uapi/linux/vhost.h
@@ -97,6 +97,8 @@
 #define VHOST_SET_BACKEND_FEATURES _IOW(VHOST_VIRTIO, 0x25, __u64)
 #define VHOST_GET_BACKEND_FEATURES _IOR(VHOST_VIRTIO, 0x26, __u64)
 
+#define VHOST_SET_RUNNING _IOW(VHOST_VIRTIO, 0x61, int)
+
 /* VHOST_NET specific defines */
 
 /* Attach virtio net ring to a raw socket, or tap device.
@@ -118,7 +120,7 @@
 /* VHOST_VSOCK specific defines */
 
 #define VHOST_VSOCK_SET_GUEST_CID	_IOW(VHOST_VIRTIO, 0x60, __u64)
-#define VHOST_VSOCK_SET_RUNNING		_IOW(VHOST_VIRTIO, 0x61, int)
+#define VHOST_VSOCK_SET_RUNNING		VHOST_SET_RUNNING
 
 /* VHOST_VDPA specific defines */
 
-- 
2.28.0

_______________________________________________
Virtualization mailing list
Virtualization@lists.linux-foundation.org
https://lists.linuxfoundation.org/mailman/listinfo/virtualization

[PATCH v7 2/3] rpmsg: move common structures and defines to headers

[Guennadi Liakhovetski]

virtio_rpmsg_bus.c keeps RPMsg protocol structure declarations and
common defines like the ones, needed for name-space announcements,
internal. Move them to common headers instead.

Signed-off-by: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
Reviewed-by: Mathieu Poirier <mathieu.poirier@linaro.org>
---
 drivers/rpmsg/virtio_rpmsg_bus.c | 78 +-----------------------------
 include/linux/rpmsg/virtio.h     | 83 ++++++++++++++++++++++++++++++++
 include/uapi/linux/rpmsg.h       |  3 ++
 3 files changed, 88 insertions(+), 76 deletions(-)
 create mode 100644 include/linux/rpmsg/virtio.h

diff --git a/drivers/rpmsg/virtio_rpmsg_bus.c b/drivers/rpmsg/virtio_rpmsg_bus.c
index 9006fc7f73d0..f39c426f9c5e 100644
--- a/drivers/rpmsg/virtio_rpmsg_bus.c
+++ b/drivers/rpmsg/virtio_rpmsg_bus.c
@@ -19,6 +19,7 @@
 #include <linux/mutex.h>
 #include <linux/of_device.h>
 #include <linux/rpmsg.h>
+#include <linux/rpmsg/virtio.h>
 #include <linux/scatterlist.h>
 #include <linux/slab.h>
 #include <linux/sched.h>
@@ -27,6 +28,7 @@
 #include <linux/virtio_ids.h>
 #include <linux/virtio_config.h>
 #include <linux/wait.h>
+#include <uapi/linux/rpmsg.h>
 
 #include "rpmsg_internal.h"
 
@@ -70,58 +72,6 @@ struct virtproc_info {
 	struct rpmsg_endpoint *ns_ept;
 };
 
-/* The feature bitmap for virtio rpmsg */
-#define VIRTIO_RPMSG_F_NS	0 /* RP supports name service notifications */
-
-/**
- * struct rpmsg_hdr - common header for all rpmsg messages
- * @src: source address
- * @dst: destination address
- * @reserved: reserved for future use
- * @len: length of payload (in bytes)
- * @flags: message flags
- * @data: @len bytes of message payload data
- *
- * Every message sent(/received) on the rpmsg bus begins with this header.
- */
-struct rpmsg_hdr {
-	__virtio32 src;
-	__virtio32 dst;
-	__virtio32 reserved;
-	__virtio16 len;
-	__virtio16 flags;
-	u8 data[];
-} __packed;
-
-/**
- * struct rpmsg_ns_msg - dynamic name service announcement message
- * @name: name of remote service that is published
- * @addr: address of remote service that is published
- * @flags: indicates whether service is created or destroyed
- *
- * This message is sent across to publish a new service, or announce
- * about its removal. When we receive these messages, an appropriate
- * rpmsg channel (i.e device) is created/destroyed. In turn, the ->probe()
- * or ->remove() handler of the appropriate rpmsg driver will be invoked
- * (if/as-soon-as one is registered).
- */
-struct rpmsg_ns_msg {
-	char name[RPMSG_NAME_SIZE];
-	__virtio32 addr;
-	__virtio32 flags;
-} __packed;
-
-/**
- * enum rpmsg_ns_flags - dynamic name service announcement flags
- *
- * @RPMSG_NS_CREATE: a new remote service was just created
- * @RPMSG_NS_DESTROY: a known remote service was just destroyed
- */
-enum rpmsg_ns_flags {
-	RPMSG_NS_CREATE		= 0,
-	RPMSG_NS_DESTROY	= 1,
-};
-
 /**
  * @vrp: the remote processor this channel belongs to
  */
@@ -134,27 +84,6 @@ struct virtio_rpmsg_channel {
 #define to_virtio_rpmsg_channel(_rpdev) \
 	container_of(_rpdev, struct virtio_rpmsg_channel, rpdev)
 
-/*
- * We're allocating buffers of 512 bytes each for communications. The
- * number of buffers will be computed from the number of buffers supported
- * by the vring, upto a maximum of 512 buffers (256 in each direction).
- *
- * Each buffer will have 16 bytes for the msg header and 496 bytes for
- * the payload.
- *
- * This will utilize a maximum total space of 256KB for the buffers.
- *
- * We might also want to add support for user-provided buffers in time.
- * This will allow bigger buffer size flexibility, and can also be used
- * to achieve zero-copy messaging.
- *
- * Note that these numbers are purely a decision of this driver - we
- * can change this without changing anything in the firmware of the remote
- * processor.
- */
-#define MAX_RPMSG_NUM_BUFS	(512)
-#define MAX_RPMSG_BUF_SIZE	(512)
-
 /*
  * Local addresses are dynamically allocated on-demand.
  * We do not dynamically assign addresses from the low 1024 range,
@@ -162,9 +91,6 @@ struct virtio_rpmsg_channel {
  */
 #define RPMSG_RESERVED_ADDRESSES	(1024)
 
-/* Address 53 is reserved for advertising remote services */
-#define RPMSG_NS_ADDR			(53)
-
 static void virtio_rpmsg_destroy_ept(struct rpmsg_endpoint *ept);
 static int virtio_rpmsg_send(struct rpmsg_endpoint *ept, void *data, int len);
 static int virtio_rpmsg_sendto(struct rpmsg_endpoint *ept, void *data, int len,
diff --git a/include/linux/rpmsg/virtio.h b/include/linux/rpmsg/virtio.h
new file mode 100644
index 000000000000..3ede1a4a68a3
--- /dev/null
+++ b/include/linux/rpmsg/virtio.h
@@ -0,0 +1,83 @@
+/* SPDX-License-Identifier: GPL-2.0 */
+
+#ifndef _LINUX_RPMSG_VIRTIO_H
+#define _LINUX_RPMSG_VIRTIO_H
+
+#include <linux/mod_devicetable.h>
+#include <linux/types.h>
+#include <linux/virtio_types.h>
+
+/**
+ * struct rpmsg_hdr - common header for all rpmsg messages
+ * @src: source address
+ * @dst: destination address
+ * @reserved: reserved for future use
+ * @len: length of payload (in bytes)
+ * @flags: message flags
+ * @data: @len bytes of message payload data
+ *
+ * Every message sent(/received) on the rpmsg bus begins with this header.
+ */
+struct rpmsg_hdr {
+	__virtio32 src;
+	__virtio32 dst;
+	__virtio32 reserved;
+	__virtio16 len;
+	__virtio16 flags;
+	u8 data[];
+} __packed;
+
+/**
+ * struct rpmsg_ns_msg - dynamic name service announcement message
+ * @name: name of remote service that is published
+ * @addr: address of remote service that is published
+ * @flags: indicates whether service is created or destroyed
+ *
+ * This message is sent across to publish a new service, or announce
+ * about its removal. When we receive these messages, an appropriate
+ * rpmsg channel (i.e device) is created/destroyed. In turn, the ->probe()
+ * or ->remove() handler of the appropriate rpmsg driver will be invoked
+ * (if/as-soon-as one is registered).
+ */
+struct rpmsg_ns_msg {
+	char name[RPMSG_NAME_SIZE];
+	__virtio32 addr;
+	__virtio32 flags;
+} __packed;
+
+/**
+ * enum rpmsg_ns_flags - dynamic name service announcement flags
+ *
+ * @RPMSG_NS_CREATE: a new remote service was just created
+ * @RPMSG_NS_DESTROY: a known remote service was just destroyed
+ */
+enum rpmsg_ns_flags {
+	RPMSG_NS_CREATE		= 0,
+	RPMSG_NS_DESTROY	= 1,
+};
+
+/*
+ * We're allocating buffers of 512 bytes each for communications. The
+ * number of buffers will be computed from the number of buffers supported
+ * by the vring, upto a maximum of 512 buffers (256 in each direction).
+ *
+ * Each buffer will have 16 bytes for the msg header and 496 bytes for
+ * the payload.
+ *
+ * This will utilize a maximum total space of 256KB for the buffers.
+ *
+ * We might also want to add support for user-provided buffers in time.
+ * This will allow bigger buffer size flexibility, and can also be used
+ * to achieve zero-copy messaging.
+ *
+ * Note that these numbers are purely a decision of this driver - we
+ * can change this without changing anything in the firmware of the remote
+ * processor.
+ */
+#define MAX_RPMSG_NUM_BUFS	512
+#define MAX_RPMSG_BUF_SIZE	512
+
+/* Address 53 is reserved for advertising remote services */
+#define RPMSG_NS_ADDR		53
+
+#endif
diff --git a/include/uapi/linux/rpmsg.h b/include/uapi/linux/rpmsg.h
index e14c6dab4223..d669c04ef289 100644
--- a/include/uapi/linux/rpmsg.h
+++ b/include/uapi/linux/rpmsg.h
@@ -24,4 +24,7 @@ struct rpmsg_endpoint_info {
 #define RPMSG_CREATE_EPT_IOCTL	_IOW(0xb5, 0x1, struct rpmsg_endpoint_info)
 #define RPMSG_DESTROY_EPT_IOCTL	_IO(0xb5, 0x2)
 
+/* The feature bitmap for virtio rpmsg */
+#define VIRTIO_RPMSG_F_NS	0 /* RP supports name service notifications */
+
 #endif
-- 
2.28.0

[PATCH v7 2/3] rpmsg: move common structures and defines to headers

[Guennadi Liakhovetski]

virtio_rpmsg_bus.c keeps RPMsg protocol structure declarations and
common defines like the ones, needed for name-space announcements,
internal. Move them to common headers instead.

Signed-off-by: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
Reviewed-by: Mathieu Poirier <mathieu.poirier@linaro.org>
---
 drivers/rpmsg/virtio_rpmsg_bus.c | 78 +-----------------------------
 include/linux/rpmsg/virtio.h     | 83 ++++++++++++++++++++++++++++++++
 include/uapi/linux/rpmsg.h       |  3 ++
 3 files changed, 88 insertions(+), 76 deletions(-)
 create mode 100644 include/linux/rpmsg/virtio.h

diff --git a/drivers/rpmsg/virtio_rpmsg_bus.c b/drivers/rpmsg/virtio_rpmsg_bus.c
index 9006fc7f73d0..f39c426f9c5e 100644
--- a/drivers/rpmsg/virtio_rpmsg_bus.c
+++ b/drivers/rpmsg/virtio_rpmsg_bus.c
@@ -19,6 +19,7 @@
 #include <linux/mutex.h>
 #include <linux/of_device.h>
 #include <linux/rpmsg.h>
+#include <linux/rpmsg/virtio.h>
 #include <linux/scatterlist.h>
 #include <linux/slab.h>
 #include <linux/sched.h>
@@ -27,6 +28,7 @@
 #include <linux/virtio_ids.h>
 #include <linux/virtio_config.h>
 #include <linux/wait.h>
+#include <uapi/linux/rpmsg.h>
 
 #include "rpmsg_internal.h"
 
@@ -70,58 +72,6 @@ struct virtproc_info {
 	struct rpmsg_endpoint *ns_ept;
 };
 
-/* The feature bitmap for virtio rpmsg */
-#define VIRTIO_RPMSG_F_NS	0 /* RP supports name service notifications */
-
-/**
- * struct rpmsg_hdr - common header for all rpmsg messages
- * @src: source address
- * @dst: destination address
- * @reserved: reserved for future use
- * @len: length of payload (in bytes)
- * @flags: message flags
- * @data: @len bytes of message payload data
- *
- * Every message sent(/received) on the rpmsg bus begins with this header.
- */
-struct rpmsg_hdr {
-	__virtio32 src;
-	__virtio32 dst;
-	__virtio32 reserved;
-	__virtio16 len;
-	__virtio16 flags;
-	u8 data[];
-} __packed;
-
-/**
- * struct rpmsg_ns_msg - dynamic name service announcement message
- * @name: name of remote service that is published
- * @addr: address of remote service that is published
- * @flags: indicates whether service is created or destroyed
- *
- * This message is sent across to publish a new service, or announce
- * about its removal. When we receive these messages, an appropriate
- * rpmsg channel (i.e device) is created/destroyed. In turn, the ->probe()
- * or ->remove() handler of the appropriate rpmsg driver will be invoked
- * (if/as-soon-as one is registered).
- */
-struct rpmsg_ns_msg {
-	char name[RPMSG_NAME_SIZE];
-	__virtio32 addr;
-	__virtio32 flags;
-} __packed;
-
-/**
- * enum rpmsg_ns_flags - dynamic name service announcement flags
- *
- * @RPMSG_NS_CREATE: a new remote service was just created
- * @RPMSG_NS_DESTROY: a known remote service was just destroyed
- */
-enum rpmsg_ns_flags {
-	RPMSG_NS_CREATE		= 0,
-	RPMSG_NS_DESTROY	= 1,
-};
-
 /**
  * @vrp: the remote processor this channel belongs to
  */
@@ -134,27 +84,6 @@ struct virtio_rpmsg_channel {
 #define to_virtio_rpmsg_channel(_rpdev) \
 	container_of(_rpdev, struct virtio_rpmsg_channel, rpdev)
 
-/*
- * We're allocating buffers of 512 bytes each for communications. The
- * number of buffers will be computed from the number of buffers supported
- * by the vring, upto a maximum of 512 buffers (256 in each direction).
- *
- * Each buffer will have 16 bytes for the msg header and 496 bytes for
- * the payload.
- *
- * This will utilize a maximum total space of 256KB for the buffers.
- *
- * We might also want to add support for user-provided buffers in time.
- * This will allow bigger buffer size flexibility, and can also be used
- * to achieve zero-copy messaging.
- *
- * Note that these numbers are purely a decision of this driver - we
- * can change this without changing anything in the firmware of the remote
- * processor.
- */
-#define MAX_RPMSG_NUM_BUFS	(512)
-#define MAX_RPMSG_BUF_SIZE	(512)
-
 /*
  * Local addresses are dynamically allocated on-demand.
  * We do not dynamically assign addresses from the low 1024 range,
@@ -162,9 +91,6 @@ struct virtio_rpmsg_channel {
  */
 #define RPMSG_RESERVED_ADDRESSES	(1024)
 
-/* Address 53 is reserved for advertising remote services */
-#define RPMSG_NS_ADDR			(53)
-
 static void virtio_rpmsg_destroy_ept(struct rpmsg_endpoint *ept);
 static int virtio_rpmsg_send(struct rpmsg_endpoint *ept, void *data, int len);
 static int virtio_rpmsg_sendto(struct rpmsg_endpoint *ept, void *data, int len,
diff --git a/include/linux/rpmsg/virtio.h b/include/linux/rpmsg/virtio.h
new file mode 100644
index 000000000000..3ede1a4a68a3
--- /dev/null
+++ b/include/linux/rpmsg/virtio.h
@@ -0,0 +1,83 @@
+/* SPDX-License-Identifier: GPL-2.0 */
+
+#ifndef _LINUX_RPMSG_VIRTIO_H
+#define _LINUX_RPMSG_VIRTIO_H
+
+#include <linux/mod_devicetable.h>
+#include <linux/types.h>
+#include <linux/virtio_types.h>
+
+/**
+ * struct rpmsg_hdr - common header for all rpmsg messages
+ * @src: source address
+ * @dst: destination address
+ * @reserved: reserved for future use
+ * @len: length of payload (in bytes)
+ * @flags: message flags
+ * @data: @len bytes of message payload data
+ *
+ * Every message sent(/received) on the rpmsg bus begins with this header.
+ */
+struct rpmsg_hdr {
+	__virtio32 src;
+	__virtio32 dst;
+	__virtio32 reserved;
+	__virtio16 len;
+	__virtio16 flags;
+	u8 data[];
+} __packed;
+
+/**
+ * struct rpmsg_ns_msg - dynamic name service announcement message
+ * @name: name of remote service that is published
+ * @addr: address of remote service that is published
+ * @flags: indicates whether service is created or destroyed
+ *
+ * This message is sent across to publish a new service, or announce
+ * about its removal. When we receive these messages, an appropriate
+ * rpmsg channel (i.e device) is created/destroyed. In turn, the ->probe()
+ * or ->remove() handler of the appropriate rpmsg driver will be invoked
+ * (if/as-soon-as one is registered).
+ */
+struct rpmsg_ns_msg {
+	char name[RPMSG_NAME_SIZE];
+	__virtio32 addr;
+	__virtio32 flags;
+} __packed;
+
+/**
+ * enum rpmsg_ns_flags - dynamic name service announcement flags
+ *
+ * @RPMSG_NS_CREATE: a new remote service was just created
+ * @RPMSG_NS_DESTROY: a known remote service was just destroyed
+ */
+enum rpmsg_ns_flags {
+	RPMSG_NS_CREATE		= 0,
+	RPMSG_NS_DESTROY	= 1,
+};
+
+/*
+ * We're allocating buffers of 512 bytes each for communications. The
+ * number of buffers will be computed from the number of buffers supported
+ * by the vring, upto a maximum of 512 buffers (256 in each direction).
+ *
+ * Each buffer will have 16 bytes for the msg header and 496 bytes for
+ * the payload.
+ *
+ * This will utilize a maximum total space of 256KB for the buffers.
+ *
+ * We might also want to add support for user-provided buffers in time.
+ * This will allow bigger buffer size flexibility, and can also be used
+ * to achieve zero-copy messaging.
+ *
+ * Note that these numbers are purely a decision of this driver - we
+ * can change this without changing anything in the firmware of the remote
+ * processor.
+ */
+#define MAX_RPMSG_NUM_BUFS	512
+#define MAX_RPMSG_BUF_SIZE	512
+
+/* Address 53 is reserved for advertising remote services */
+#define RPMSG_NS_ADDR		53
+
+#endif
diff --git a/include/uapi/linux/rpmsg.h b/include/uapi/linux/rpmsg.h
index e14c6dab4223..d669c04ef289 100644
--- a/include/uapi/linux/rpmsg.h
+++ b/include/uapi/linux/rpmsg.h
@@ -24,4 +24,7 @@ struct rpmsg_endpoint_info {
 #define RPMSG_CREATE_EPT_IOCTL	_IOW(0xb5, 0x1, struct rpmsg_endpoint_info)
 #define RPMSG_DESTROY_EPT_IOCTL	_IO(0xb5, 0x2)
 
+/* The feature bitmap for virtio rpmsg */
+#define VIRTIO_RPMSG_F_NS	0 /* RP supports name service notifications */
+
 #endif
-- 
2.28.0

_______________________________________________
Virtualization mailing list
Virtualization@lists.linux-foundation.org
https://lists.linuxfoundation.org/mailman/listinfo/virtualization

[PATCH v7 3/3] vhost: add an RPMsg API

[Guennadi Liakhovetski]

Linux supports running the RPMsg protocol over the VirtIO transport
protocol, but currently there is only support for VirtIO clients and
no support for VirtIO servers. This patch adds a vhost-based RPMsg
server implementation, which makes it possible to use RPMsg over
VirtIO between guest VMs and the host.

Signed-off-by: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
---
 drivers/vhost/Kconfig       |   7 +
 drivers/vhost/Makefile      |   3 +
 drivers/vhost/rpmsg.c       | 370 ++++++++++++++++++++++++++++++++++++
 drivers/vhost/vhost_rpmsg.h |  74 ++++++++
 4 files changed, 454 insertions(+)
 create mode 100644 drivers/vhost/rpmsg.c
 create mode 100644 drivers/vhost/vhost_rpmsg.h

diff --git a/drivers/vhost/Kconfig b/drivers/vhost/Kconfig
index 587fbae06182..ee1a19b7ab3d 100644
--- a/drivers/vhost/Kconfig
+++ b/drivers/vhost/Kconfig
@@ -38,6 +38,13 @@ config VHOST_NET
 	  To compile this driver as a module, choose M here: the module will
 	  be called vhost_net.
 
+config VHOST_RPMSG
+	tristate
+	select VHOST
+	help
+	  Vhost RPMsg API allows vhost drivers to communicate with VirtIO
+	  drivers on guest VMs, using the RPMsg over VirtIO protocol.
+
 config VHOST_SCSI
 	tristate "VHOST_SCSI TCM fabric driver"
 	depends on TARGET_CORE && EVENTFD
diff --git a/drivers/vhost/Makefile b/drivers/vhost/Makefile
index f3e1897cce85..9cf459d59f97 100644
--- a/drivers/vhost/Makefile
+++ b/drivers/vhost/Makefile
@@ -2,6 +2,9 @@
 obj-$(CONFIG_VHOST_NET) += vhost_net.o
 vhost_net-y := net.o
 
+obj-$(CONFIG_VHOST_RPMSG) += vhost_rpmsg.o
+vhost_rpmsg-y := rpmsg.o
+
 obj-$(CONFIG_VHOST_SCSI) += vhost_scsi.o
 vhost_scsi-y := scsi.o
 
diff --git a/drivers/vhost/rpmsg.c b/drivers/vhost/rpmsg.c
new file mode 100644
index 000000000000..0ddee5b5f017
--- /dev/null
+++ b/drivers/vhost/rpmsg.c
@@ -0,0 +1,370 @@
+// SPDX-License-Identifier: GPL-2.0-only
+/*
+ * Copyright(c) 2020 Intel Corporation. All rights reserved.
+ *
+ * Author: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
+ *
+ * Vhost RPMsg VirtIO interface provides a set of functions to be used on the
+ * host side as a counterpart to the guest side RPMsg VirtIO API, provided by
+ * drivers/rpmsg/virtio_rpmsg_bus.c. This API can be used by any vhost driver to
+ * handle RPMsg specific virtqueue processing.
+ * Vhost drivers, using this API will use their own VirtIO device IDs, that
+ * should then also be added to the ID table in virtio_rpmsg_bus.c
+ */
+
+#include <linux/compat.h>
+#include <linux/file.h>
+#include <linux/miscdevice.h>
+#include <linux/module.h>
+#include <linux/mutex.h>
+#include <linux/rpmsg/virtio.h>
+#include <linux/vhost.h>
+#include <uapi/linux/rpmsg.h>
+
+#include "vhost.h"
+#include "vhost_rpmsg.h"
+
+/*
+ * All virtio-rpmsg virtual queue kicks always come with just one buffer -
+ * either input or output, but we can also handle split messages
+ */
+static int vhost_rpmsg_get_msg(struct vhost_virtqueue *vq, unsigned int *cnt)
+{
+	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
+	unsigned int out, in;
+	int head = vhost_get_vq_desc(vq, vq->iov, ARRAY_SIZE(vq->iov), &out, &in,
+				     NULL, NULL);
+	if (head < 0) {
+		vq_err(vq, "%s(): error %d getting buffer\n",
+		       __func__, head);
+		return head;
+	}
+
+	/* Nothing new? */
+	if (head == vq->num)
+		return head;
+
+	if (vq == &vr->vq[VIRTIO_RPMSG_RESPONSE]) {
+		if (out) {
+			vq_err(vq, "%s(): invalid %d output in response queue\n",
+			       __func__, out);
+			goto return_buf;
+		}
+
+		*cnt = in;
+	}
+
+	if (vq == &vr->vq[VIRTIO_RPMSG_REQUEST]) {
+		if (in) {
+			vq_err(vq, "%s(): invalid %d input in request queue\n",
+		       __func__, in);
+			goto return_buf;
+		}
+
+		*cnt = out;
+	}
+
+	return head;
+
+return_buf:
+	vhost_add_used(vq, head, 0);
+
+	return -EINVAL;
+}
+
+static const struct vhost_rpmsg_ept *vhost_rpmsg_ept_find(struct vhost_rpmsg *vr, int addr)
+{
+	unsigned int i;
+
+	for (i = 0; i < vr->n_epts; i++)
+		if (vr->ept[i].addr == addr)
+			return vr->ept + i;
+
+	return NULL;
+}
+
+/*
+ * if len < 0, then for reading a request, the complete virtual queue buffer
+ * size is prepared, for sending a response, the length in the iterator is used
+ */
+int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
+			   unsigned int qid, ssize_t len)
+	__acquires(vq->mutex)
+{
+	struct vhost_virtqueue *vq = vr->vq + qid;
+	unsigned int cnt;
+	ssize_t ret;
+	size_t tmp;
+
+	if (qid >= VIRTIO_RPMSG_NUM_OF_VQS)
+		return -EINVAL;
+
+	iter->vq = vq;
+
+	mutex_lock(&vq->mutex);
+	vhost_disable_notify(&vr->dev, vq);
+
+	iter->head = vhost_rpmsg_get_msg(vq, &cnt);
+	if (iter->head == vq->num)
+		iter->head = -EAGAIN;
+
+	if (iter->head < 0) {
+		ret = iter->head;
+		goto unlock;
+	}
+
+	tmp = iov_length(vq->iov, cnt);
+	if (tmp < sizeof(iter->rhdr)) {
+		vq_err(vq, "%s(): size %zu too small\n", __func__, tmp);
+		ret = -ENOBUFS;
+		goto return_buf;
+	}
+
+	switch (qid) {
+	case VIRTIO_RPMSG_REQUEST:
+		if (len >= 0) {
+			if (tmp < sizeof(iter->rhdr) + len) {
+				ret = -ENOBUFS;
+				goto return_buf;
+			}
+
+			tmp = len + sizeof(iter->rhdr);
+		}
+
+		/* len is now the size of the payload */
+		iov_iter_init(&iter->iov_iter, WRITE, vq->iov, cnt, tmp);
+
+		/* Read the RPMSG header with endpoint addresses */
+		tmp = copy_from_iter(&iter->rhdr, sizeof(iter->rhdr), &iter->iov_iter);
+		if (tmp != sizeof(iter->rhdr)) {
+			vq_err(vq, "%s(): got %zu instead of %zu\n", __func__,
+			       tmp, sizeof(iter->rhdr));
+			ret = -EIO;
+			goto return_buf;
+		}
+
+		iter->ept = vhost_rpmsg_ept_find(vr, vhost32_to_cpu(vq, iter->rhdr.dst));
+		if (!iter->ept) {
+			vq_err(vq, "%s(): no endpoint with address %d\n",
+			       __func__, vhost32_to_cpu(vq, iter->rhdr.dst));
+			ret = -ENOENT;
+			goto return_buf;
+		}
+
+		/* Let the endpoint read the payload */
+		if (iter->ept->read) {
+			ret = iter->ept->read(vr, iter);
+			if (ret < 0)
+				goto return_buf;
+
+			iter->rhdr.len = cpu_to_vhost16(vq, ret);
+		} else {
+			iter->rhdr.len = 0;
+		}
+
+		/* Prepare for the response phase */
+		iter->rhdr.dst = iter->rhdr.src;
+		iter->rhdr.src = cpu_to_vhost32(vq, iter->ept->addr);
+
+		break;
+	case VIRTIO_RPMSG_RESPONSE:
+		if (!iter->ept && iter->rhdr.dst != cpu_to_vhost32(vq, RPMSG_NS_ADDR)) {
+			/*
+			 * Usually the iterator is configured when processing a
+			 * message on the request queue, but it's also possible
+			 * to send a message on the response queue without a
+			 * preceding request, in that case the iterator must
+			 * contain source and destination addresses.
+			 */
+			iter->ept = vhost_rpmsg_ept_find(vr, vhost32_to_cpu(vq, iter->rhdr.src));
+			if (!iter->ept) {
+				ret = -ENOENT;
+				goto return_buf;
+			}
+		}
+
+		if (len >= 0) {
+			if (tmp < sizeof(iter->rhdr) + len) {
+				ret = -ENOBUFS;
+				goto return_buf;
+			}
+
+			iter->rhdr.len = cpu_to_vhost16(vq, len);
+			tmp = len + sizeof(iter->rhdr);
+		}
+
+		/* len is now the size of the payload */
+		iov_iter_init(&iter->iov_iter, READ, vq->iov, cnt, tmp);
+
+		/* Write the RPMSG header with endpoint addresses */
+		tmp = copy_to_iter(&iter->rhdr, sizeof(iter->rhdr), &iter->iov_iter);
+		if (tmp != sizeof(iter->rhdr)) {
+			ret = -EIO;
+			goto return_buf;
+		}
+
+		/* Let the endpoint write the payload */
+		if (iter->ept && iter->ept->write) {
+			ret = iter->ept->write(vr, iter);
+			if (ret < 0)
+				goto return_buf;
+		}
+
+		break;
+	}
+
+	return 0;
+
+return_buf:
+	vhost_add_used(vq, iter->head, 0);
+unlock:
+	vhost_enable_notify(&vr->dev, vq);
+	mutex_unlock(&vq->mutex);
+
+	return ret;
+}
+EXPORT_SYMBOL_GPL(vhost_rpmsg_start_lock);
+
+size_t vhost_rpmsg_copy(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
+			void *data, size_t size)
+{
+	/*
+	 * We could check for excess data, but copy_{to,from}_iter() don't do
+	 * that either
+	 */
+	if (iter->vq == vr->vq + VIRTIO_RPMSG_RESPONSE)
+		return copy_to_iter(data, size, &iter->iov_iter);
+
+	return copy_from_iter(data, size, &iter->iov_iter);
+}
+EXPORT_SYMBOL_GPL(vhost_rpmsg_copy);
+
+int vhost_rpmsg_finish_unlock(struct vhost_rpmsg *vr,
+			      struct vhost_rpmsg_iter *iter)
+	__releases(vq->mutex)
+{
+	if (iter->head >= 0)
+		vhost_add_used_and_signal(iter->vq->dev, iter->vq, iter->head,
+					  vhost16_to_cpu(iter->vq, iter->rhdr.len) +
+					  sizeof(iter->rhdr));
+
+	vhost_enable_notify(&vr->dev, iter->vq);
+	mutex_unlock(&iter->vq->mutex);
+
+	return iter->head;
+}
+EXPORT_SYMBOL_GPL(vhost_rpmsg_finish_unlock);
+
+/*
+ * Return false to terminate the external loop only if we fail to obtain either
+ * a request or a response buffer
+ */
+static bool handle_rpmsg_req_single(struct vhost_rpmsg *vr,
+				    struct vhost_virtqueue *vq)
+{
+	struct vhost_rpmsg_iter iter;
+	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_REQUEST, -EINVAL);
+	if (!ret)
+		ret = vhost_rpmsg_finish_unlock(vr, &iter);
+	if (ret < 0) {
+		if (ret != -EAGAIN)
+			vq_err(vq, "%s(): RPMSG processing failed %d\n",
+			       __func__, ret);
+		return false;
+	}
+
+	if (!iter.ept->write)
+		return true;
+
+	ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, -EINVAL);
+	if (!ret)
+		ret = vhost_rpmsg_finish_unlock(vr, &iter);
+	if (ret < 0) {
+		vq_err(vq, "%s(): RPMSG finalising failed %d\n", __func__, ret);
+		return false;
+	}
+
+	return true;
+}
+
+static void handle_rpmsg_req_kick(struct vhost_work *work)
+{
+	struct vhost_virtqueue *vq = container_of(work, struct vhost_virtqueue,
+						  poll.work);
+	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
+
+	while (handle_rpmsg_req_single(vr, vq))
+		;
+}
+
+/*
+ * initialise two virtqueues with an array of endpoints,
+ * request and response callbacks
+ */
+void vhost_rpmsg_init(struct vhost_rpmsg *vr, const struct vhost_rpmsg_ept *ept,
+		      unsigned int n_epts)
+{
+	unsigned int i;
+
+	for (i = 0; i < ARRAY_SIZE(vr->vq); i++)
+		vr->vq_p[i] = &vr->vq[i];
+
+	/* vq[0]: host -> guest, vq[1]: host <- guest */
+	vr->vq[VIRTIO_RPMSG_REQUEST].handle_kick = handle_rpmsg_req_kick;
+	vr->vq[VIRTIO_RPMSG_RESPONSE].handle_kick = NULL;
+
+	vr->ept = ept;
+	vr->n_epts = n_epts;
+
+	vhost_dev_init(&vr->dev, vr->vq_p, VIRTIO_RPMSG_NUM_OF_VQS,
+		       UIO_MAXIOV, 0, 0, true, NULL);
+}
+EXPORT_SYMBOL_GPL(vhost_rpmsg_init);
+
+void vhost_rpmsg_destroy(struct vhost_rpmsg *vr)
+{
+	if (vhost_dev_has_owner(&vr->dev))
+		vhost_poll_flush(&vr->vq[VIRTIO_RPMSG_REQUEST].poll);
+
+	vhost_dev_cleanup(&vr->dev);
+}
+EXPORT_SYMBOL_GPL(vhost_rpmsg_destroy);
+
+/* send namespace */
+int vhost_rpmsg_ns_announce(struct vhost_rpmsg *vr, const char *name, unsigned int src)
+{
+	struct vhost_virtqueue *vq = &vr->vq[VIRTIO_RPMSG_RESPONSE];
+	struct vhost_rpmsg_iter iter = {
+		.rhdr = {
+			.src = 0,
+			.dst = cpu_to_vhost32(vq, RPMSG_NS_ADDR),
+		},
+	};
+	struct rpmsg_ns_msg ns = {
+		.addr = cpu_to_vhost32(vq, src),
+		.flags = cpu_to_vhost32(vq, RPMSG_NS_CREATE), /* for rpmsg_ns_cb() */
+	};
+	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, sizeof(ns));
+
+	if (ret < 0)
+		return ret;
+
+	strlcpy(ns.name, name, sizeof(ns.name));
+
+	ret = vhost_rpmsg_copy(vr, &iter, &ns, sizeof(ns));
+	if (ret != sizeof(ns))
+		vq_err(iter.vq, "%s(): added %d instead of %zu bytes\n",
+		       __func__, ret, sizeof(ns));
+
+	ret = vhost_rpmsg_finish_unlock(vr, &iter);
+	if (ret < 0)
+		vq_err(iter.vq, "%s(): namespace announcement failed: %d\n",
+		       __func__, ret);
+
+	return ret;
+}
+EXPORT_SYMBOL_GPL(vhost_rpmsg_ns_announce);
+
+MODULE_LICENSE("GPL v2");
+MODULE_AUTHOR("Intel, Inc.");
+MODULE_DESCRIPTION("Vhost RPMsg API");
diff --git a/drivers/vhost/vhost_rpmsg.h b/drivers/vhost/vhost_rpmsg.h
new file mode 100644
index 000000000000..c020ea14cd16
--- /dev/null
+++ b/drivers/vhost/vhost_rpmsg.h
@@ -0,0 +1,74 @@
+/* SPDX-License-Identifier: GPL-2.0 */
+/*
+ * Copyright(c) 2020 Intel Corporation. All rights reserved.
+ *
+ * Author: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
+ */
+
+#ifndef VHOST_RPMSG_H
+#define VHOST_RPMSG_H
+
+#include <linux/rpmsg/virtio.h>
+#include <linux/uio.h>
+
+#include "vhost.h"
+
+/* RPMsg uses two VirtQueues: one for each direction */
+enum {
+	VIRTIO_RPMSG_RESPONSE,	/* RPMsg response (host->guest) buffers */
+	VIRTIO_RPMSG_REQUEST,	/* RPMsg request (guest->host) buffers */
+	/* Keep last */
+	VIRTIO_RPMSG_NUM_OF_VQS,
+};
+
+struct vhost_rpmsg_ept;
+
+struct vhost_rpmsg_iter {
+	struct iov_iter iov_iter;
+	struct rpmsg_hdr rhdr;
+	struct vhost_virtqueue *vq;
+	const struct vhost_rpmsg_ept *ept;
+	int head;
+	void *priv;
+};
+
+struct vhost_rpmsg {
+	struct vhost_dev dev;
+	struct vhost_virtqueue vq[VIRTIO_RPMSG_NUM_OF_VQS];
+	struct vhost_virtqueue *vq_p[VIRTIO_RPMSG_NUM_OF_VQS];
+	const struct vhost_rpmsg_ept *ept;
+	unsigned int n_epts;
+};
+
+struct vhost_rpmsg_ept {
+	ssize_t (*read)(struct vhost_rpmsg *, struct vhost_rpmsg_iter *);
+	ssize_t (*write)(struct vhost_rpmsg *, struct vhost_rpmsg_iter *);
+	int addr;
+};
+
+static inline size_t vhost_rpmsg_iter_len(const struct vhost_rpmsg_iter *iter)
+{
+	return iter->rhdr.len;
+}
+
+#define VHOST_RPMSG_ITER(_vq, _src, _dst) {			\
+	.rhdr = {						\
+			.src = cpu_to_vhost32(_vq, _src),	\
+			.dst = cpu_to_vhost32(_vq, _dst),	\
+		},						\
+	}
+
+void vhost_rpmsg_init(struct vhost_rpmsg *vr, const struct vhost_rpmsg_ept *ept,
+		      unsigned int n_epts);
+void vhost_rpmsg_destroy(struct vhost_rpmsg *vr);
+int vhost_rpmsg_ns_announce(struct vhost_rpmsg *vr, const char *name,
+			    unsigned int src);
+int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr,
+			   struct vhost_rpmsg_iter *iter,
+			   unsigned int qid, ssize_t len);
+size_t vhost_rpmsg_copy(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
+			void *data, size_t size);
+int vhost_rpmsg_finish_unlock(struct vhost_rpmsg *vr,
+			      struct vhost_rpmsg_iter *iter);
+
+#endif
-- 
2.28.0

_______________________________________________
Virtualization mailing list
Virtualization@lists.linux-foundation.org
https://lists.linuxfoundation.org/mailman/listinfo/virtualization

Re: [PATCH v7 3/3] vhost: add an RPMsg API

[Vincent Whitchurch]

On Thu, Sep 10, 2020 at 01:13:51PM +0200, Guennadi Liakhovetski wrote:
> +int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> +			   unsigned int qid, ssize_t len)
> +	__acquires(vq->mutex)
> +{
> +	struct vhost_virtqueue *vq = vr->vq + qid;
> +	unsigned int cnt;
> +	ssize_t ret;
> +	size_t tmp;
> +
> +	if (qid >= VIRTIO_RPMSG_NUM_OF_VQS)
> +		return -EINVAL;
> +
> +	iter->vq = vq;
> +
> +	mutex_lock(&vq->mutex);
> +	vhost_disable_notify(&vr->dev, vq);
> +
> +	iter->head = vhost_rpmsg_get_msg(vq, &cnt);
> +	if (iter->head == vq->num)
> +		iter->head = -EAGAIN;
> +
> +	if (iter->head < 0) {
> +		ret = iter->head;
> +		goto unlock;
> +	}
> +
[...]
> +
> +return_buf:
> +	vhost_add_used(vq, iter->head, 0);
> +unlock:
> +	vhost_enable_notify(&vr->dev, vq);
> +	mutex_unlock(&vq->mutex);
> +
> +	return ret;
> +}

There is a race condition here.  New buffers could have been added while
notifications were disabled (between vhost_disable_notify() and
vhost_enable_notify()), so the other vhost drivers check the return
value of vhost_enable_notify() and rerun their work loops if it returns
true.  This driver doesn't do that so it stops processing requests if
that condition hits.

Something like the below seems to fix it but the correct fix could maybe
involve changing this API to account for this case so that it looks more
like the code in other vhost drivers.

diff --git a/drivers/vhost/rpmsg.c b/drivers/vhost/rpmsg.c
index 7c753258d42..673dd4ec865 100644
--- a/drivers/vhost/rpmsg.c
+++ b/drivers/vhost/rpmsg.c
@@ -302,8 +302,14 @@ static void handle_rpmsg_req_kick(struct vhost_work *work)
 	struct vhost_virtqueue *vq = container_of(work, struct vhost_virtqueue,
 						  poll.work);
 	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
+	struct vhost_virtqueue *reqvq = vr->vq + VIRTIO_RPMSG_REQUEST;
 
-	while (handle_rpmsg_req_single(vr, vq))
+	/*
+	 * The !vhost_vq_avail_empty() check is needed since the vhost_rpmsg*
+	 * APIs don't check the return value of vhost_enable_notify() and retry
+	 * if there were buffers added while notifications were disabled.
+	 */
+	while (handle_rpmsg_req_single(vr, vq) || !vhost_vq_avail_empty(reqvq->dev, reqvq))
 		;
 }
 

Re: [PATCH v7 3/3] vhost: add an RPMsg API

[Vincent Whitchurch]

On Thu, Sep 10, 2020 at 01:13:51PM +0200, Guennadi Liakhovetski wrote:
> +int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> +			   unsigned int qid, ssize_t len)
> +	__acquires(vq->mutex)
> +{
> +	struct vhost_virtqueue *vq = vr->vq + qid;
> +	unsigned int cnt;
> +	ssize_t ret;
> +	size_t tmp;
> +
> +	if (qid >= VIRTIO_RPMSG_NUM_OF_VQS)
> +		return -EINVAL;
> +
> +	iter->vq = vq;
> +
> +	mutex_lock(&vq->mutex);
> +	vhost_disable_notify(&vr->dev, vq);
> +
> +	iter->head = vhost_rpmsg_get_msg(vq, &cnt);
> +	if (iter->head == vq->num)
> +		iter->head = -EAGAIN;
> +
> +	if (iter->head < 0) {
> +		ret = iter->head;
> +		goto unlock;
> +	}
> +
[...]
> +
> +return_buf:
> +	vhost_add_used(vq, iter->head, 0);
> +unlock:
> +	vhost_enable_notify(&vr->dev, vq);
> +	mutex_unlock(&vq->mutex);
> +
> +	return ret;
> +}

There is a race condition here.  New buffers could have been added while
notifications were disabled (between vhost_disable_notify() and
vhost_enable_notify()), so the other vhost drivers check the return
value of vhost_enable_notify() and rerun their work loops if it returns
true.  This driver doesn't do that so it stops processing requests if
that condition hits.

Something like the below seems to fix it but the correct fix could maybe
involve changing this API to account for this case so that it looks more
like the code in other vhost drivers.

diff --git a/drivers/vhost/rpmsg.c b/drivers/vhost/rpmsg.c
index 7c753258d42..673dd4ec865 100644
--- a/drivers/vhost/rpmsg.c
+++ b/drivers/vhost/rpmsg.c
@@ -302,8 +302,14 @@ static void handle_rpmsg_req_kick(struct vhost_work *work)
 	struct vhost_virtqueue *vq = container_of(work, struct vhost_virtqueue,
 						  poll.work);
 	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
+	struct vhost_virtqueue *reqvq = vr->vq + VIRTIO_RPMSG_REQUEST;
 
-	while (handle_rpmsg_req_single(vr, vq))
+	/*
+	 * The !vhost_vq_avail_empty() check is needed since the vhost_rpmsg*
+	 * APIs don't check the return value of vhost_enable_notify() and retry
+	 * if there were buffers added while notifications were disabled.
+	 */
+	while (handle_rpmsg_req_single(vr, vq) || !vhost_vq_avail_empty(reqvq->dev, reqvq))
 		;
 }
 
_______________________________________________
Virtualization mailing list
Virtualization@lists.linux-foundation.org
https://lists.linuxfoundation.org/mailman/listinfo/virtualization

Re: [PATCH v7 3/3] vhost: add an RPMsg API

[Guennadi Liakhovetski]

Hi Vincent,

On Thu, Sep 17, 2020 at 10:55:59AM +0200, Vincent Whitchurch wrote:
> On Thu, Sep 10, 2020 at 01:13:51PM +0200, Guennadi Liakhovetski wrote:
> > +int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> > +			   unsigned int qid, ssize_t len)
> > +	__acquires(vq->mutex)
> > +{
> > +	struct vhost_virtqueue *vq = vr->vq + qid;
> > +	unsigned int cnt;
> > +	ssize_t ret;
> > +	size_t tmp;
> > +
> > +	if (qid >= VIRTIO_RPMSG_NUM_OF_VQS)
> > +		return -EINVAL;
> > +
> > +	iter->vq = vq;
> > +
> > +	mutex_lock(&vq->mutex);
> > +	vhost_disable_notify(&vr->dev, vq);
> > +
> > +	iter->head = vhost_rpmsg_get_msg(vq, &cnt);
> > +	if (iter->head == vq->num)
> > +		iter->head = -EAGAIN;
> > +
> > +	if (iter->head < 0) {
> > +		ret = iter->head;
> > +		goto unlock;
> > +	}
> > +
> [...]
> > +
> > +return_buf:
> > +	vhost_add_used(vq, iter->head, 0);
> > +unlock:
> > +	vhost_enable_notify(&vr->dev, vq);
> > +	mutex_unlock(&vq->mutex);
> > +
> > +	return ret;
> > +}
> 
> There is a race condition here.  New buffers could have been added while
> notifications were disabled (between vhost_disable_notify() and
> vhost_enable_notify()), so the other vhost drivers check the return
> value of vhost_enable_notify() and rerun their work loops if it returns
> true.  This driver doesn't do that so it stops processing requests if
> that condition hits.

You're right, thanks for spotting this!

> Something like the below seems to fix it but the correct fix could maybe
> involve changing this API to account for this case so that it looks more
> like the code in other vhost drivers.

I'll try to use your proposed code, we'll see if it turns out incorrect.

> diff --git a/drivers/vhost/rpmsg.c b/drivers/vhost/rpmsg.c
> index 7c753258d42..673dd4ec865 100644
> --- a/drivers/vhost/rpmsg.c
> +++ b/drivers/vhost/rpmsg.c
> @@ -302,8 +302,14 @@ static void handle_rpmsg_req_kick(struct vhost_work *work)
>  	struct vhost_virtqueue *vq = container_of(work, struct vhost_virtqueue,
>  						  poll.work);
>  	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
> +	struct vhost_virtqueue *reqvq = vr->vq + VIRTIO_RPMSG_REQUEST;

This is only called on the request queue, so, we can just use vq here.

>  
> -	while (handle_rpmsg_req_single(vr, vq))
> +	/*
> +	 * The !vhost_vq_avail_empty() check is needed since the vhost_rpmsg*
> +	 * APIs don't check the return value of vhost_enable_notify() and retry
> +	 * if there were buffers added while notifications were disabled.
> +	 */
> +	while (handle_rpmsg_req_single(vr, vq) || !vhost_vq_avail_empty(reqvq->dev, reqvq))
>  		;
>  }

Thanks
Guennadi
_______________________________________________
Virtualization mailing list
Virtualization@lists.linux-foundation.org
https://lists.linuxfoundation.org/mailman/listinfo/virtualization

Re: [PATCH v7 3/3] vhost: add an RPMsg API

[Mathieu Poirier]

On Thu, Sep 10, 2020 at 01:13:51PM +0200, Guennadi Liakhovetski wrote:
> Linux supports running the RPMsg protocol over the VirtIO transport
> protocol, but currently there is only support for VirtIO clients and
> no support for VirtIO servers. This patch adds a vhost-based RPMsg
> server implementation, which makes it possible to use RPMsg over
> VirtIO between guest VMs and the host.

I now get the client/server concept you are describing above but that happened
only after a lot of mental gymnastics.  If you drop the whole client/server
concept and concentrate on what this patch does, things will go better.  I would
personally go with what you have in the Kconfig: 

> +	  Vhost RPMsg API allows vhost drivers to communicate with VirtIO
> +	  drivers on guest VMs, using the RPMsg over VirtIO protocol.

It is concise but describes exactly what this patch provide.

> 
> Signed-off-by: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> ---
>  drivers/vhost/Kconfig       |   7 +
>  drivers/vhost/Makefile      |   3 +
>  drivers/vhost/rpmsg.c       | 370 ++++++++++++++++++++++++++++++++++++
>  drivers/vhost/vhost_rpmsg.h |  74 ++++++++
>  4 files changed, 454 insertions(+)
>  create mode 100644 drivers/vhost/rpmsg.c
>  create mode 100644 drivers/vhost/vhost_rpmsg.h
> 
> diff --git a/drivers/vhost/Kconfig b/drivers/vhost/Kconfig
> index 587fbae06182..ee1a19b7ab3d 100644
> --- a/drivers/vhost/Kconfig
> +++ b/drivers/vhost/Kconfig
> @@ -38,6 +38,13 @@ config VHOST_NET
>  	  To compile this driver as a module, choose M here: the module will
>  	  be called vhost_net.
>  
> +config VHOST_RPMSG
> +	tristate
> +	select VHOST
> +	help
> +	  Vhost RPMsg API allows vhost drivers to communicate with VirtIO
> +	  drivers on guest VMs, using the RPMsg over VirtIO protocol.
> +

I suppose you intend this to be selectable from another config option?

>  config VHOST_SCSI
>  	tristate "VHOST_SCSI TCM fabric driver"
>  	depends on TARGET_CORE && EVENTFD
> diff --git a/drivers/vhost/Makefile b/drivers/vhost/Makefile
> index f3e1897cce85..9cf459d59f97 100644
> --- a/drivers/vhost/Makefile
> +++ b/drivers/vhost/Makefile
> @@ -2,6 +2,9 @@
>  obj-$(CONFIG_VHOST_NET) += vhost_net.o
>  vhost_net-y := net.o
>  
> +obj-$(CONFIG_VHOST_RPMSG) += vhost_rpmsg.o
> +vhost_rpmsg-y := rpmsg.o
> +
>  obj-$(CONFIG_VHOST_SCSI) += vhost_scsi.o
>  vhost_scsi-y := scsi.o
>  
> diff --git a/drivers/vhost/rpmsg.c b/drivers/vhost/rpmsg.c
> new file mode 100644
> index 000000000000..0ddee5b5f017
> --- /dev/null
> +++ b/drivers/vhost/rpmsg.c
> @@ -0,0 +1,370 @@
> +// SPDX-License-Identifier: GPL-2.0-only
> +/*
> + * Copyright(c) 2020 Intel Corporation. All rights reserved.
> + *
> + * Author: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> + *
> + * Vhost RPMsg VirtIO interface provides a set of functions to be used on the
> + * host side as a counterpart to the guest side RPMsg VirtIO API, provided by
> + * drivers/rpmsg/virtio_rpmsg_bus.c. This API can be used by any vhost driver to
> + * handle RPMsg specific virtqueue processing.
> + * Vhost drivers, using this API will use their own VirtIO device IDs, that
> + * should then also be added to the ID table in virtio_rpmsg_bus.c
> + */
> +
> +#include <linux/compat.h>
> +#include <linux/file.h>
> +#include <linux/miscdevice.h>

As far as I can tell the above two are not needed.

> +#include <linux/module.h>
> +#include <linux/mutex.h>
> +#include <linux/rpmsg/virtio.h>
> +#include <linux/vhost.h>
> +#include <uapi/linux/rpmsg.h>
> +
> +#include "vhost.h"
> +#include "vhost_rpmsg.h"
> +
> +/*
> + * All virtio-rpmsg virtual queue kicks always come with just one buffer -
> + * either input or output, but we can also handle split messages
> + */
> +static int vhost_rpmsg_get_msg(struct vhost_virtqueue *vq, unsigned int *cnt)
> +{
> +	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
> +	unsigned int out, in;
> +	int head = vhost_get_vq_desc(vq, vq->iov, ARRAY_SIZE(vq->iov), &out, &in,
> +				     NULL, NULL);
> +	if (head < 0) {
> +		vq_err(vq, "%s(): error %d getting buffer\n",
> +		       __func__, head);
> +		return head;
> +	}
> +
> +	/* Nothing new? */
> +	if (head == vq->num)
> +		return head;
> +
> +	if (vq == &vr->vq[VIRTIO_RPMSG_RESPONSE]) {
> +		if (out) {
> +			vq_err(vq, "%s(): invalid %d output in response queue\n",
> +			       __func__, out);
> +			goto return_buf;
> +		}
> +
> +		*cnt = in;
> +	}
> +
> +	if (vq == &vr->vq[VIRTIO_RPMSG_REQUEST]) {
> +		if (in) {
> +			vq_err(vq, "%s(): invalid %d input in request queue\n",
> +		       __func__, in);
> +			goto return_buf;
> +		}
> +
> +		*cnt = out;
> +	}
> +
> +	return head;
> +
> +return_buf:
> +	vhost_add_used(vq, head, 0);
> +
> +	return -EINVAL;
> +}
> +
> +static const struct vhost_rpmsg_ept *vhost_rpmsg_ept_find(struct vhost_rpmsg *vr, int addr)
> +{
> +	unsigned int i;
> +
> +	for (i = 0; i < vr->n_epts; i++)
> +		if (vr->ept[i].addr == addr)
> +			return vr->ept + i;
> +
> +	return NULL;
> +}
> +
> +/*
> + * if len < 0, then for reading a request, the complete virtual queue buffer
> + * size is prepared, for sending a response, the length in the iterator is used
> + */
> +int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> +			   unsigned int qid, ssize_t len)
> +	__acquires(vq->mutex)
> +{
> +	struct vhost_virtqueue *vq = vr->vq + qid;
> +	unsigned int cnt;
> +	ssize_t ret;
> +	size_t tmp;
> +
> +	if (qid >= VIRTIO_RPMSG_NUM_OF_VQS)
> +		return -EINVAL;
> +
> +	iter->vq = vq;
> +
> +	mutex_lock(&vq->mutex);
> +	vhost_disable_notify(&vr->dev, vq);
> +
> +	iter->head = vhost_rpmsg_get_msg(vq, &cnt);
> +	if (iter->head == vq->num)
> +		iter->head = -EAGAIN;
> +
> +	if (iter->head < 0) {
> +		ret = iter->head;
> +		goto unlock;
> +	}
> +
> +	tmp = iov_length(vq->iov, cnt);
> +	if (tmp < sizeof(iter->rhdr)) {
> +		vq_err(vq, "%s(): size %zu too small\n", __func__, tmp);
> +		ret = -ENOBUFS;
> +		goto return_buf;
> +	}
> +
> +	switch (qid) {
> +	case VIRTIO_RPMSG_REQUEST:
> +		if (len >= 0) {
> +			if (tmp < sizeof(iter->rhdr) + len) {
> +				ret = -ENOBUFS;
> +				goto return_buf;
> +			}
> +
> +			tmp = len + sizeof(iter->rhdr);
> +		}
> +
> +		/* len is now the size of the payload */
> +		iov_iter_init(&iter->iov_iter, WRITE, vq->iov, cnt, tmp);
> +
> +		/* Read the RPMSG header with endpoint addresses */
> +		tmp = copy_from_iter(&iter->rhdr, sizeof(iter->rhdr), &iter->iov_iter);
> +		if (tmp != sizeof(iter->rhdr)) {
> +			vq_err(vq, "%s(): got %zu instead of %zu\n", __func__,
> +			       tmp, sizeof(iter->rhdr));
> +			ret = -EIO;
> +			goto return_buf;
> +		}
> +
> +		iter->ept = vhost_rpmsg_ept_find(vr, vhost32_to_cpu(vq, iter->rhdr.dst));
> +		if (!iter->ept) {
> +			vq_err(vq, "%s(): no endpoint with address %d\n",
> +			       __func__, vhost32_to_cpu(vq, iter->rhdr.dst));
> +			ret = -ENOENT;
> +			goto return_buf;
> +		}
> +
> +		/* Let the endpoint read the payload */
> +		if (iter->ept->read) {
> +			ret = iter->ept->read(vr, iter);
> +			if (ret < 0)
> +				goto return_buf;
> +
> +			iter->rhdr.len = cpu_to_vhost16(vq, ret);
> +		} else {
> +			iter->rhdr.len = 0;
> +		}
> +
> +		/* Prepare for the response phase */
> +		iter->rhdr.dst = iter->rhdr.src;
> +		iter->rhdr.src = cpu_to_vhost32(vq, iter->ept->addr);

I'm a little puzzled here - what will the response look like?  And why is it
prepared here?  From what I can see doing so introduces coupling with function
handle_rpmsg_req_single().  I think confirmation of reception should be handled
by endpoints rather than in the core. 

> +
> +		break;
> +	case VIRTIO_RPMSG_RESPONSE:
> +		if (!iter->ept && iter->rhdr.dst != cpu_to_vhost32(vq, RPMSG_NS_ADDR)) {
> +			/*
> +			 * Usually the iterator is configured when processing a
> +			 * message on the request queue, but it's also possible
> +			 * to send a message on the response queue without a
> +			 * preceding request, in that case the iterator must
> +			 * contain source and destination addresses.
> +			 */
> +			iter->ept = vhost_rpmsg_ept_find(vr, vhost32_to_cpu(vq, iter->rhdr.src));
> +			if (!iter->ept) {
> +				ret = -ENOENT;
> +				goto return_buf;
> +			}
> +		}
> +
> +		if (len >= 0) {
> +			if (tmp < sizeof(iter->rhdr) + len) {
> +				ret = -ENOBUFS;
> +				goto return_buf;
> +			}
> +
> +			iter->rhdr.len = cpu_to_vhost16(vq, len);
> +			tmp = len + sizeof(iter->rhdr);
> +		}
> +
> +		/* len is now the size of the payload */
> +		iov_iter_init(&iter->iov_iter, READ, vq->iov, cnt, tmp);
> +
> +		/* Write the RPMSG header with endpoint addresses */
> +		tmp = copy_to_iter(&iter->rhdr, sizeof(iter->rhdr), &iter->iov_iter);
> +		if (tmp != sizeof(iter->rhdr)) {
> +			ret = -EIO;
> +			goto return_buf;
> +		}
> +
> +		/* Let the endpoint write the payload */

I would specifically mention that namespace payloads are taken care of by
vhost_rpmsg_ns_announce().  That makes it easier for people to connect the dots. 

> +		if (iter->ept && iter->ept->write) {
> +			ret = iter->ept->write(vr, iter);
> +			if (ret < 0)
> +				goto return_buf;
> +		}
> +
> +		break;
> +	}
> +
> +	return 0;
> +
> +return_buf:
> +	vhost_add_used(vq, iter->head, 0);
> +unlock:
> +	vhost_enable_notify(&vr->dev, vq);
> +	mutex_unlock(&vq->mutex);
> +
> +	return ret;
> +}
> +EXPORT_SYMBOL_GPL(vhost_rpmsg_start_lock);
> +
> +size_t vhost_rpmsg_copy(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> +			void *data, size_t size)
> +{
> +	/*
> +	 * We could check for excess data, but copy_{to,from}_iter() don't do
> +	 * that either
> +	 */
> +	if (iter->vq == vr->vq + VIRTIO_RPMSG_RESPONSE)
> +		return copy_to_iter(data, size, &iter->iov_iter);
> +
> +	return copy_from_iter(data, size, &iter->iov_iter);
> +}
> +EXPORT_SYMBOL_GPL(vhost_rpmsg_copy);
> +
> +int vhost_rpmsg_finish_unlock(struct vhost_rpmsg *vr,
> +			      struct vhost_rpmsg_iter *iter)
> +	__releases(vq->mutex)
> +{
> +	if (iter->head >= 0)
> +		vhost_add_used_and_signal(iter->vq->dev, iter->vq, iter->head,
> +					  vhost16_to_cpu(iter->vq, iter->rhdr.len) +
> +					  sizeof(iter->rhdr));
> +
> +	vhost_enable_notify(&vr->dev, iter->vq);
> +	mutex_unlock(&iter->vq->mutex);
> +
> +	return iter->head;
> +}
> +EXPORT_SYMBOL_GPL(vhost_rpmsg_finish_unlock);
> +
> +/*
> + * Return false to terminate the external loop only if we fail to obtain either
> + * a request or a response buffer
> + */
> +static bool handle_rpmsg_req_single(struct vhost_rpmsg *vr,
> +				    struct vhost_virtqueue *vq)
> +{
> +	struct vhost_rpmsg_iter iter;
> +	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_REQUEST, -EINVAL);
> +	if (!ret)
> +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> +	if (ret < 0) {
> +		if (ret != -EAGAIN)
> +			vq_err(vq, "%s(): RPMSG processing failed %d\n",
> +			       __func__, ret);
> +		return false;
> +	}
> +
> +	if (!iter.ept->write)
> +		return true;
> +
> +	ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, -EINVAL);
> +	if (!ret)
> +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> +	if (ret < 0) {
> +		vq_err(vq, "%s(): RPMSG finalising failed %d\n", __func__, ret);
> +		return false;
> +	}

As I said before dealing with the "response" queue here seems to be introducing
coupling with vhost_rpmsg_start_lock()...  Endpoints should be doing that.

> +
> +	return true;
> +}
> +
> +static void handle_rpmsg_req_kick(struct vhost_work *work)
> +{
> +	struct vhost_virtqueue *vq = container_of(work, struct vhost_virtqueue,
> +						  poll.work);
> +	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
> +
> +	while (handle_rpmsg_req_single(vr, vq))
> +		;
> +}
> +
> +/*
> + * initialise two virtqueues with an array of endpoints,
> + * request and response callbacks
> + */
> +void vhost_rpmsg_init(struct vhost_rpmsg *vr, const struct vhost_rpmsg_ept *ept,
> +		      unsigned int n_epts)
> +{
> +	unsigned int i;
> +
> +	for (i = 0; i < ARRAY_SIZE(vr->vq); i++)
> +		vr->vq_p[i] = &vr->vq[i];
> +
> +	/* vq[0]: host -> guest, vq[1]: host <- guest */
> +	vr->vq[VIRTIO_RPMSG_REQUEST].handle_kick = handle_rpmsg_req_kick;
> +	vr->vq[VIRTIO_RPMSG_RESPONSE].handle_kick = NULL;

The comment depicts vq[0] followed by vq[1] but the code initialise vq[1] before
vq[0], which is wildly confusing.  At the very least this should be: 

	vr->vq[VIRTIO_RPMSG_RESPONSE].handle_kick = NULL;
	vr->vq[VIRTIO_RPMSG_REQUEST].handle_kick = handle_rpmsg_req_kick;

And even better:

        /* See configuration of *vq_cbs[] in rpmsg_probe()  */
	vr->vq[VIRTIO_RPMSG_TX].handle_kick = NULL;
	vr->vq[VIRTIO_RPMSG_RX].handle_kick = handle_rpmsg_req_kick;

> +
> +	vr->ept = ept;
> +	vr->n_epts = n_epts;
> +
> +	vhost_dev_init(&vr->dev, vr->vq_p, VIRTIO_RPMSG_NUM_OF_VQS,
> +		       UIO_MAXIOV, 0, 0, true, NULL);
> +}
> +EXPORT_SYMBOL_GPL(vhost_rpmsg_init);
> +
> +void vhost_rpmsg_destroy(struct vhost_rpmsg *vr)
> +{
> +	if (vhost_dev_has_owner(&vr->dev))
> +		vhost_poll_flush(&vr->vq[VIRTIO_RPMSG_REQUEST].poll);
> +
> +	vhost_dev_cleanup(&vr->dev);
> +}
> +EXPORT_SYMBOL_GPL(vhost_rpmsg_destroy);
> +
> +/* send namespace */
> +int vhost_rpmsg_ns_announce(struct vhost_rpmsg *vr, const char *name, unsigned int src)
> +{
> +	struct vhost_virtqueue *vq = &vr->vq[VIRTIO_RPMSG_RESPONSE];
> +	struct vhost_rpmsg_iter iter = {
> +		.rhdr = {
> +			.src = 0,
> +			.dst = cpu_to_vhost32(vq, RPMSG_NS_ADDR),
> +		},
> +	};
> +	struct rpmsg_ns_msg ns = {
> +		.addr = cpu_to_vhost32(vq, src),
> +		.flags = cpu_to_vhost32(vq, RPMSG_NS_CREATE), /* for rpmsg_ns_cb() */
> +	};

Here we have to assume the source can be found in the endpoints registered in
vhost_rpmsg_init().  I would put a check to make sure that is the case and
return an error otherwise. 

> +	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, sizeof(ns));
> +
> +	if (ret < 0)
> +		return ret;
> +
> +	strlcpy(ns.name, name, sizeof(ns.name));
> +
> +	ret = vhost_rpmsg_copy(vr, &iter, &ns, sizeof(ns));
> +	if (ret != sizeof(ns))
> +		vq_err(iter.vq, "%s(): added %d instead of %zu bytes\n",
> +		       __func__, ret, sizeof(ns));
> +
> +	ret = vhost_rpmsg_finish_unlock(vr, &iter);
> +	if (ret < 0)
> +		vq_err(iter.vq, "%s(): namespace announcement failed: %d\n",
> +		       __func__, ret);
> +
> +	return ret;
> +}
> +EXPORT_SYMBOL_GPL(vhost_rpmsg_ns_announce);
> +
> +MODULE_LICENSE("GPL v2");
> +MODULE_AUTHOR("Intel, Inc.");
> +MODULE_DESCRIPTION("Vhost RPMsg API");
> diff --git a/drivers/vhost/vhost_rpmsg.h b/drivers/vhost/vhost_rpmsg.h
> new file mode 100644
> index 000000000000..c020ea14cd16
> --- /dev/null
> +++ b/drivers/vhost/vhost_rpmsg.h
> @@ -0,0 +1,74 @@
> +/* SPDX-License-Identifier: GPL-2.0 */
> +/*
> + * Copyright(c) 2020 Intel Corporation. All rights reserved.
> + *
> + * Author: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> + */
> +
> +#ifndef VHOST_RPMSG_H
> +#define VHOST_RPMSG_H
> +
> +#include <linux/rpmsg/virtio.h>
> +#include <linux/uio.h>
> +
> +#include "vhost.h"
> +
> +/* RPMsg uses two VirtQueues: one for each direction */
> +enum {
> +	VIRTIO_RPMSG_RESPONSE,	/* RPMsg response (host->guest) buffers */
> +	VIRTIO_RPMSG_REQUEST,	/* RPMsg request (guest->host) buffers */

As I said above things would be much clearer if this was VIRTIO_RPMSG_TX and
VIRTIO_RPMSG_RX.

I won't be commenting on the mechanic needed to access and send information on
the virtqueues as it is completely foreign to me.  Other than the above I think
this is going somewhere. 

Thanks,
Mathieu

> +	/* Keep last */
> +	VIRTIO_RPMSG_NUM_OF_VQS,
> +};
> +
> +struct vhost_rpmsg_ept;
> +
> +struct vhost_rpmsg_iter {
> +	struct iov_iter iov_iter;
> +	struct rpmsg_hdr rhdr;
> +	struct vhost_virtqueue *vq;
> +	const struct vhost_rpmsg_ept *ept;
> +	int head;
> +	void *priv;
> +};
> +
> +struct vhost_rpmsg {
> +	struct vhost_dev dev;
> +	struct vhost_virtqueue vq[VIRTIO_RPMSG_NUM_OF_VQS];
> +	struct vhost_virtqueue *vq_p[VIRTIO_RPMSG_NUM_OF_VQS];
> +	const struct vhost_rpmsg_ept *ept;
> +	unsigned int n_epts;
> +};
> +
> +struct vhost_rpmsg_ept {
> +	ssize_t (*read)(struct vhost_rpmsg *, struct vhost_rpmsg_iter *);
> +	ssize_t (*write)(struct vhost_rpmsg *, struct vhost_rpmsg_iter *);
> +	int addr;
> +};
> +
> +static inline size_t vhost_rpmsg_iter_len(const struct vhost_rpmsg_iter *iter)
> +{
> +	return iter->rhdr.len;
> +}
> +
> +#define VHOST_RPMSG_ITER(_vq, _src, _dst) {			\
> +	.rhdr = {						\
> +			.src = cpu_to_vhost32(_vq, _src),	\
> +			.dst = cpu_to_vhost32(_vq, _dst),	\
> +		},						\
> +	}
> +
> +void vhost_rpmsg_init(struct vhost_rpmsg *vr, const struct vhost_rpmsg_ept *ept,
> +		      unsigned int n_epts);
> +void vhost_rpmsg_destroy(struct vhost_rpmsg *vr);
> +int vhost_rpmsg_ns_announce(struct vhost_rpmsg *vr, const char *name,
> +			    unsigned int src);
> +int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr,
> +			   struct vhost_rpmsg_iter *iter,
> +			   unsigned int qid, ssize_t len);
> +size_t vhost_rpmsg_copy(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> +			void *data, size_t size);
> +int vhost_rpmsg_finish_unlock(struct vhost_rpmsg *vr,
> +			      struct vhost_rpmsg_iter *iter);
> +
> +#endif
> -- 
> 2.28.0
> 

Re: [PATCH v7 3/3] vhost: add an RPMsg API

[Guennadi Liakhovetski]

Hi Mathieu,

On Thu, Sep 17, 2020 at 04:01:38PM -0600, Mathieu Poirier wrote:
> On Thu, Sep 10, 2020 at 01:13:51PM +0200, Guennadi Liakhovetski wrote:
> > Linux supports running the RPMsg protocol over the VirtIO transport
> > protocol, but currently there is only support for VirtIO clients and
> > no support for VirtIO servers. This patch adds a vhost-based RPMsg
> > server implementation, which makes it possible to use RPMsg over
> > VirtIO between guest VMs and the host.
> 
> I now get the client/server concept you are describing above but that happened
> only after a lot of mental gymnastics.  If you drop the whole client/server
> concept and concentrate on what this patch does, things will go better.  I would
> personally go with what you have in the Kconfig: 
> 
> > +	  Vhost RPMsg API allows vhost drivers to communicate with VirtIO
> > +	  drivers on guest VMs, using the RPMsg over VirtIO protocol.
> 
> It is concise but describes exactly what this patch provide.

Ok, thanks, will try to improve.

> > Signed-off-by: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > ---
> >  drivers/vhost/Kconfig       |   7 +
> >  drivers/vhost/Makefile      |   3 +
> >  drivers/vhost/rpmsg.c       | 370 ++++++++++++++++++++++++++++++++++++
> >  drivers/vhost/vhost_rpmsg.h |  74 ++++++++
> >  4 files changed, 454 insertions(+)
> >  create mode 100644 drivers/vhost/rpmsg.c
> >  create mode 100644 drivers/vhost/vhost_rpmsg.h
> > 
> > diff --git a/drivers/vhost/Kconfig b/drivers/vhost/Kconfig
> > index 587fbae06182..ee1a19b7ab3d 100644
> > --- a/drivers/vhost/Kconfig
> > +++ b/drivers/vhost/Kconfig
> > @@ -38,6 +38,13 @@ config VHOST_NET
> >  	  To compile this driver as a module, choose M here: the module will
> >  	  be called vhost_net.
> >  
> > +config VHOST_RPMSG
> > +	tristate
> > +	select VHOST
> > +	help
> > +	  Vhost RPMsg API allows vhost drivers to communicate with VirtIO
> > +	  drivers on guest VMs, using the RPMsg over VirtIO protocol.
> > +
> 
> I suppose you intend this to be selectable from another config option?

yes.

> >  config VHOST_SCSI
> >  	tristate "VHOST_SCSI TCM fabric driver"
> >  	depends on TARGET_CORE && EVENTFD
> > diff --git a/drivers/vhost/Makefile b/drivers/vhost/Makefile
> > index f3e1897cce85..9cf459d59f97 100644
> > --- a/drivers/vhost/Makefile
> > +++ b/drivers/vhost/Makefile
> > @@ -2,6 +2,9 @@
> >  obj-$(CONFIG_VHOST_NET) += vhost_net.o
> >  vhost_net-y := net.o
> >  
> > +obj-$(CONFIG_VHOST_RPMSG) += vhost_rpmsg.o
> > +vhost_rpmsg-y := rpmsg.o
> > +
> >  obj-$(CONFIG_VHOST_SCSI) += vhost_scsi.o
> >  vhost_scsi-y := scsi.o
> >  
> > diff --git a/drivers/vhost/rpmsg.c b/drivers/vhost/rpmsg.c
> > new file mode 100644
> > index 000000000000..0ddee5b5f017
> > --- /dev/null
> > +++ b/drivers/vhost/rpmsg.c
> > @@ -0,0 +1,370 @@
> > +// SPDX-License-Identifier: GPL-2.0-only
> > +/*
> > + * Copyright(c) 2020 Intel Corporation. All rights reserved.
> > + *
> > + * Author: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > + *
> > + * Vhost RPMsg VirtIO interface provides a set of functions to be used on the
> > + * host side as a counterpart to the guest side RPMsg VirtIO API, provided by
> > + * drivers/rpmsg/virtio_rpmsg_bus.c. This API can be used by any vhost driver to
> > + * handle RPMsg specific virtqueue processing.
> > + * Vhost drivers, using this API will use their own VirtIO device IDs, that
> > + * should then also be added to the ID table in virtio_rpmsg_bus.c
> > + */
> > +
> > +#include <linux/compat.h>
> > +#include <linux/file.h>
> > +#include <linux/miscdevice.h>
> 
> As far as I can tell the above two are not needed.

Look like left-over, will remove.

> > +#include <linux/module.h>
> > +#include <linux/mutex.h>
> > +#include <linux/rpmsg/virtio.h>
> > +#include <linux/vhost.h>
> > +#include <uapi/linux/rpmsg.h>
> > +
> > +#include "vhost.h"
> > +#include "vhost_rpmsg.h"
> > +
> > +/*
> > + * All virtio-rpmsg virtual queue kicks always come with just one buffer -
> > + * either input or output, but we can also handle split messages
> > + */
> > +static int vhost_rpmsg_get_msg(struct vhost_virtqueue *vq, unsigned int *cnt)
> > +{
> > +	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
> > +	unsigned int out, in;
> > +	int head = vhost_get_vq_desc(vq, vq->iov, ARRAY_SIZE(vq->iov), &out, &in,
> > +				     NULL, NULL);
> > +	if (head < 0) {
> > +		vq_err(vq, "%s(): error %d getting buffer\n",
> > +		       __func__, head);
> > +		return head;
> > +	}
> > +
> > +	/* Nothing new? */
> > +	if (head == vq->num)
> > +		return head;
> > +
> > +	if (vq == &vr->vq[VIRTIO_RPMSG_RESPONSE]) {
> > +		if (out) {
> > +			vq_err(vq, "%s(): invalid %d output in response queue\n",
> > +			       __func__, out);
> > +			goto return_buf;
> > +		}
> > +
> > +		*cnt = in;
> > +	}
> > +
> > +	if (vq == &vr->vq[VIRTIO_RPMSG_REQUEST]) {
> > +		if (in) {
> > +			vq_err(vq, "%s(): invalid %d input in request queue\n",
> > +		       __func__, in);
> > +			goto return_buf;
> > +		}
> > +
> > +		*cnt = out;
> > +	}
> > +
> > +	return head;
> > +
> > +return_buf:
> > +	vhost_add_used(vq, head, 0);
> > +
> > +	return -EINVAL;
> > +}
> > +
> > +static const struct vhost_rpmsg_ept *vhost_rpmsg_ept_find(struct vhost_rpmsg *vr, int addr)
> > +{
> > +	unsigned int i;
> > +
> > +	for (i = 0; i < vr->n_epts; i++)
> > +		if (vr->ept[i].addr == addr)
> > +			return vr->ept + i;
> > +
> > +	return NULL;
> > +}
> > +
> > +/*
> > + * if len < 0, then for reading a request, the complete virtual queue buffer
> > + * size is prepared, for sending a response, the length in the iterator is used
> > + */
> > +int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> > +			   unsigned int qid, ssize_t len)
> > +	__acquires(vq->mutex)
> > +{
> > +	struct vhost_virtqueue *vq = vr->vq + qid;
> > +	unsigned int cnt;
> > +	ssize_t ret;
> > +	size_t tmp;
> > +
> > +	if (qid >= VIRTIO_RPMSG_NUM_OF_VQS)
> > +		return -EINVAL;
> > +
> > +	iter->vq = vq;
> > +
> > +	mutex_lock(&vq->mutex);
> > +	vhost_disable_notify(&vr->dev, vq);
> > +
> > +	iter->head = vhost_rpmsg_get_msg(vq, &cnt);
> > +	if (iter->head == vq->num)
> > +		iter->head = -EAGAIN;
> > +
> > +	if (iter->head < 0) {
> > +		ret = iter->head;
> > +		goto unlock;
> > +	}
> > +
> > +	tmp = iov_length(vq->iov, cnt);
> > +	if (tmp < sizeof(iter->rhdr)) {
> > +		vq_err(vq, "%s(): size %zu too small\n", __func__, tmp);
> > +		ret = -ENOBUFS;
> > +		goto return_buf;
> > +	}
> > +
> > +	switch (qid) {
> > +	case VIRTIO_RPMSG_REQUEST:
> > +		if (len >= 0) {
> > +			if (tmp < sizeof(iter->rhdr) + len) {
> > +				ret = -ENOBUFS;
> > +				goto return_buf;
> > +			}
> > +
> > +			tmp = len + sizeof(iter->rhdr);
> > +		}
> > +
> > +		/* len is now the size of the payload */
> > +		iov_iter_init(&iter->iov_iter, WRITE, vq->iov, cnt, tmp);
> > +
> > +		/* Read the RPMSG header with endpoint addresses */
> > +		tmp = copy_from_iter(&iter->rhdr, sizeof(iter->rhdr), &iter->iov_iter);
> > +		if (tmp != sizeof(iter->rhdr)) {
> > +			vq_err(vq, "%s(): got %zu instead of %zu\n", __func__,
> > +			       tmp, sizeof(iter->rhdr));
> > +			ret = -EIO;
> > +			goto return_buf;
> > +		}
> > +
> > +		iter->ept = vhost_rpmsg_ept_find(vr, vhost32_to_cpu(vq, iter->rhdr.dst));
> > +		if (!iter->ept) {
> > +			vq_err(vq, "%s(): no endpoint with address %d\n",
> > +			       __func__, vhost32_to_cpu(vq, iter->rhdr.dst));
> > +			ret = -ENOENT;
> > +			goto return_buf;
> > +		}
> > +
> > +		/* Let the endpoint read the payload */
> > +		if (iter->ept->read) {
> > +			ret = iter->ept->read(vr, iter);
> > +			if (ret < 0)
> > +				goto return_buf;
> > +
> > +			iter->rhdr.len = cpu_to_vhost16(vq, ret);
> > +		} else {
> > +			iter->rhdr.len = 0;
> > +		}
> > +
> > +		/* Prepare for the response phase */
> > +		iter->rhdr.dst = iter->rhdr.src;
> > +		iter->rhdr.src = cpu_to_vhost32(vq, iter->ept->addr);
> 
> I'm a little puzzled here - what will the response look like?  And why is it
> prepared here?  From what I can see doing so introduces coupling with function
> handle_rpmsg_req_single().  I think confirmation of reception should be handled
> by endpoints rather than in the core. 

RPMsg always contain a header, so we keep the header in the iterator. If the 
caller wants to reply to the request, the easiest way to do that is to reuse the 
iterator. In that case obviously you have to swap source and destination 
addresses. This can be done either in the request handler of the API, or by the 
caller, or in the API response handler. It would be silly to have the user do 
that, that would be repeated code. But I agree, it's a bit unclean to modify the 
header before returning it to the user, without knowing, whether the user will 
use it, in which case it might be surprised to see most fields from the request 
unchanged and only addresses swapped. I'll move this to response with a check 
for a reused iterator.

> > +
> > +		break;
> > +	case VIRTIO_RPMSG_RESPONSE:
> > +		if (!iter->ept && iter->rhdr.dst != cpu_to_vhost32(vq, RPMSG_NS_ADDR)) {
> > +			/*
> > +			 * Usually the iterator is configured when processing a
> > +			 * message on the request queue, but it's also possible
> > +			 * to send a message on the response queue without a
> > +			 * preceding request, in that case the iterator must
> > +			 * contain source and destination addresses.
> > +			 */
> > +			iter->ept = vhost_rpmsg_ept_find(vr, vhost32_to_cpu(vq, iter->rhdr.src));
> > +			if (!iter->ept) {
> > +				ret = -ENOENT;
> > +				goto return_buf;
> > +			}
> > +		}
> > +
> > +		if (len >= 0) {
> > +			if (tmp < sizeof(iter->rhdr) + len) {
> > +				ret = -ENOBUFS;
> > +				goto return_buf;
> > +			}
> > +
> > +			iter->rhdr.len = cpu_to_vhost16(vq, len);
> > +			tmp = len + sizeof(iter->rhdr);
> > +		}
> > +
> > +		/* len is now the size of the payload */
> > +		iov_iter_init(&iter->iov_iter, READ, vq->iov, cnt, tmp);
> > +
> > +		/* Write the RPMSG header with endpoint addresses */
> > +		tmp = copy_to_iter(&iter->rhdr, sizeof(iter->rhdr), &iter->iov_iter);
> > +		if (tmp != sizeof(iter->rhdr)) {
> > +			ret = -EIO;
> > +			goto return_buf;
> > +		}
> > +
> > +		/* Let the endpoint write the payload */
> 
> I would specifically mention that namespace payloads are taken care of by
> vhost_rpmsg_ns_announce().  That makes it easier for people to connect the dots. 

Ok

> > +		if (iter->ept && iter->ept->write) {
> > +			ret = iter->ept->write(vr, iter);
> > +			if (ret < 0)
> > +				goto return_buf;
> > +		}
> > +
> > +		break;
> > +	}
> > +
> > +	return 0;
> > +
> > +return_buf:
> > +	vhost_add_used(vq, iter->head, 0);
> > +unlock:
> > +	vhost_enable_notify(&vr->dev, vq);
> > +	mutex_unlock(&vq->mutex);
> > +
> > +	return ret;
> > +}
> > +EXPORT_SYMBOL_GPL(vhost_rpmsg_start_lock);
> > +
> > +size_t vhost_rpmsg_copy(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> > +			void *data, size_t size)
> > +{
> > +	/*
> > +	 * We could check for excess data, but copy_{to,from}_iter() don't do
> > +	 * that either
> > +	 */
> > +	if (iter->vq == vr->vq + VIRTIO_RPMSG_RESPONSE)
> > +		return copy_to_iter(data, size, &iter->iov_iter);
> > +
> > +	return copy_from_iter(data, size, &iter->iov_iter);
> > +}
> > +EXPORT_SYMBOL_GPL(vhost_rpmsg_copy);
> > +
> > +int vhost_rpmsg_finish_unlock(struct vhost_rpmsg *vr,
> > +			      struct vhost_rpmsg_iter *iter)
> > +	__releases(vq->mutex)
> > +{
> > +	if (iter->head >= 0)
> > +		vhost_add_used_and_signal(iter->vq->dev, iter->vq, iter->head,
> > +					  vhost16_to_cpu(iter->vq, iter->rhdr.len) +
> > +					  sizeof(iter->rhdr));
> > +
> > +	vhost_enable_notify(&vr->dev, iter->vq);
> > +	mutex_unlock(&iter->vq->mutex);
> > +
> > +	return iter->head;
> > +}
> > +EXPORT_SYMBOL_GPL(vhost_rpmsg_finish_unlock);
> > +
> > +/*
> > + * Return false to terminate the external loop only if we fail to obtain either
> > + * a request or a response buffer
> > + */
> > +static bool handle_rpmsg_req_single(struct vhost_rpmsg *vr,
> > +				    struct vhost_virtqueue *vq)
> > +{
> > +	struct vhost_rpmsg_iter iter;
> > +	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_REQUEST, -EINVAL);
> > +	if (!ret)
> > +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > +	if (ret < 0) {
> > +		if (ret != -EAGAIN)
> > +			vq_err(vq, "%s(): RPMSG processing failed %d\n",
> > +			       __func__, ret);
> > +		return false;
> > +	}
> > +
> > +	if (!iter.ept->write)
> > +		return true;
> > +
> > +	ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, -EINVAL);
> > +	if (!ret)
> > +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > +	if (ret < 0) {
> > +		vq_err(vq, "%s(): RPMSG finalising failed %d\n", __func__, ret);
> > +		return false;
> > +	}
> 
> As I said before dealing with the "response" queue here seems to be introducing
> coupling with vhost_rpmsg_start_lock()...  Endpoints should be doing that.

Sorry, could you elaborate a bit, what do you mean by coupling?

> > +
> > +	return true;
> > +}
> > +
> > +static void handle_rpmsg_req_kick(struct vhost_work *work)
> > +{
> > +	struct vhost_virtqueue *vq = container_of(work, struct vhost_virtqueue,
> > +						  poll.work);
> > +	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
> > +
> > +	while (handle_rpmsg_req_single(vr, vq))
> > +		;
> > +}
> > +
> > +/*
> > + * initialise two virtqueues with an array of endpoints,
> > + * request and response callbacks
> > + */
> > +void vhost_rpmsg_init(struct vhost_rpmsg *vr, const struct vhost_rpmsg_ept *ept,
> > +		      unsigned int n_epts)
> > +{
> > +	unsigned int i;
> > +
> > +	for (i = 0; i < ARRAY_SIZE(vr->vq); i++)
> > +		vr->vq_p[i] = &vr->vq[i];
> > +
> > +	/* vq[0]: host -> guest, vq[1]: host <- guest */
> > +	vr->vq[VIRTIO_RPMSG_REQUEST].handle_kick = handle_rpmsg_req_kick;
> > +	vr->vq[VIRTIO_RPMSG_RESPONSE].handle_kick = NULL;
> 
> The comment depicts vq[0] followed by vq[1] but the code initialise vq[1] before
> vq[0], which is wildly confusing.  At the very least this should be: 

Nobody should care which of those is 0 and which is 1 :-) But indeed you have a point, 
that the protocol isn't strictly request-response based, the host can also send 
messages to the guest without preceding requests. So, TX / RX should be a better fit.

> 
> 	vr->vq[VIRTIO_RPMSG_RESPONSE].handle_kick = NULL;
> 	vr->vq[VIRTIO_RPMSG_REQUEST].handle_kick = handle_rpmsg_req_kick;
> 
> And even better:
> 
>         /* See configuration of *vq_cbs[] in rpmsg_probe()  */
> 	vr->vq[VIRTIO_RPMSG_TX].handle_kick = NULL;
> 	vr->vq[VIRTIO_RPMSG_RX].handle_kick = handle_rpmsg_req_kick;
> 
> > +
> > +	vr->ept = ept;
> > +	vr->n_epts = n_epts;
> > +
> > +	vhost_dev_init(&vr->dev, vr->vq_p, VIRTIO_RPMSG_NUM_OF_VQS,
> > +		       UIO_MAXIOV, 0, 0, true, NULL);
> > +}
> > +EXPORT_SYMBOL_GPL(vhost_rpmsg_init);
> > +
> > +void vhost_rpmsg_destroy(struct vhost_rpmsg *vr)
> > +{
> > +	if (vhost_dev_has_owner(&vr->dev))
> > +		vhost_poll_flush(&vr->vq[VIRTIO_RPMSG_REQUEST].poll);
> > +
> > +	vhost_dev_cleanup(&vr->dev);
> > +}
> > +EXPORT_SYMBOL_GPL(vhost_rpmsg_destroy);
> > +
> > +/* send namespace */
> > +int vhost_rpmsg_ns_announce(struct vhost_rpmsg *vr, const char *name, unsigned int src)
> > +{
> > +	struct vhost_virtqueue *vq = &vr->vq[VIRTIO_RPMSG_RESPONSE];
> > +	struct vhost_rpmsg_iter iter = {
> > +		.rhdr = {
> > +			.src = 0,
> > +			.dst = cpu_to_vhost32(vq, RPMSG_NS_ADDR),
> > +		},
> > +	};
> > +	struct rpmsg_ns_msg ns = {
> > +		.addr = cpu_to_vhost32(vq, src),
> > +		.flags = cpu_to_vhost32(vq, RPMSG_NS_CREATE), /* for rpmsg_ns_cb() */
> > +	};
> 
> Here we have to assume the source can be found in the endpoints registered in
> vhost_rpmsg_init().  I would put a check to make sure that is the case and
> return an error otherwise. 

Ok, will do.

> > +	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, sizeof(ns));
> > +
> > +	if (ret < 0)
> > +		return ret;
> > +
> > +	strlcpy(ns.name, name, sizeof(ns.name));
> > +
> > +	ret = vhost_rpmsg_copy(vr, &iter, &ns, sizeof(ns));
> > +	if (ret != sizeof(ns))
> > +		vq_err(iter.vq, "%s(): added %d instead of %zu bytes\n",
> > +		       __func__, ret, sizeof(ns));
> > +
> > +	ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > +	if (ret < 0)
> > +		vq_err(iter.vq, "%s(): namespace announcement failed: %d\n",
> > +		       __func__, ret);
> > +
> > +	return ret;
> > +}
> > +EXPORT_SYMBOL_GPL(vhost_rpmsg_ns_announce);
> > +
> > +MODULE_LICENSE("GPL v2");
> > +MODULE_AUTHOR("Intel, Inc.");
> > +MODULE_DESCRIPTION("Vhost RPMsg API");
> > diff --git a/drivers/vhost/vhost_rpmsg.h b/drivers/vhost/vhost_rpmsg.h
> > new file mode 100644
> > index 000000000000..c020ea14cd16
> > --- /dev/null
> > +++ b/drivers/vhost/vhost_rpmsg.h
> > @@ -0,0 +1,74 @@
> > +/* SPDX-License-Identifier: GPL-2.0 */
> > +/*
> > + * Copyright(c) 2020 Intel Corporation. All rights reserved.
> > + *
> > + * Author: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > + */
> > +
> > +#ifndef VHOST_RPMSG_H
> > +#define VHOST_RPMSG_H
> > +
> > +#include <linux/rpmsg/virtio.h>
> > +#include <linux/uio.h>
> > +
> > +#include "vhost.h"
> > +
> > +/* RPMsg uses two VirtQueues: one for each direction */
> > +enum {
> > +	VIRTIO_RPMSG_RESPONSE,	/* RPMsg response (host->guest) buffers */
> > +	VIRTIO_RPMSG_REQUEST,	/* RPMsg request (guest->host) buffers */
> 
> As I said above things would be much clearer if this was VIRTIO_RPMSG_TX and
> VIRTIO_RPMSG_RX.

Ack.

> I won't be commenting on the mechanic needed to access and send information on
> the virtqueues as it is completely foreign to me.  Other than the above I think
> this is going somewhere. 

I'll wait for your clarifications about "coupling" and send a v8.

Thanks for the comments so far
Guennadi

> Thanks,
> Mathieu
> 
> > +	/* Keep last */
> > +	VIRTIO_RPMSG_NUM_OF_VQS,
> > +};
> > +
> > +struct vhost_rpmsg_ept;
> > +
> > +struct vhost_rpmsg_iter {
> > +	struct iov_iter iov_iter;
> > +	struct rpmsg_hdr rhdr;
> > +	struct vhost_virtqueue *vq;
> > +	const struct vhost_rpmsg_ept *ept;
> > +	int head;
> > +	void *priv;
> > +};
> > +
> > +struct vhost_rpmsg {
> > +	struct vhost_dev dev;
> > +	struct vhost_virtqueue vq[VIRTIO_RPMSG_NUM_OF_VQS];
> > +	struct vhost_virtqueue *vq_p[VIRTIO_RPMSG_NUM_OF_VQS];
> > +	const struct vhost_rpmsg_ept *ept;
> > +	unsigned int n_epts;
> > +};
> > +
> > +struct vhost_rpmsg_ept {
> > +	ssize_t (*read)(struct vhost_rpmsg *, struct vhost_rpmsg_iter *);
> > +	ssize_t (*write)(struct vhost_rpmsg *, struct vhost_rpmsg_iter *);
> > +	int addr;
> > +};
> > +
> > +static inline size_t vhost_rpmsg_iter_len(const struct vhost_rpmsg_iter *iter)
> > +{
> > +	return iter->rhdr.len;
> > +}
> > +
> > +#define VHOST_RPMSG_ITER(_vq, _src, _dst) {			\
> > +	.rhdr = {						\
> > +			.src = cpu_to_vhost32(_vq, _src),	\
> > +			.dst = cpu_to_vhost32(_vq, _dst),	\
> > +		},						\
> > +	}
> > +
> > +void vhost_rpmsg_init(struct vhost_rpmsg *vr, const struct vhost_rpmsg_ept *ept,
> > +		      unsigned int n_epts);
> > +void vhost_rpmsg_destroy(struct vhost_rpmsg *vr);
> > +int vhost_rpmsg_ns_announce(struct vhost_rpmsg *vr, const char *name,
> > +			    unsigned int src);
> > +int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr,
> > +			   struct vhost_rpmsg_iter *iter,
> > +			   unsigned int qid, ssize_t len);
> > +size_t vhost_rpmsg_copy(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> > +			void *data, size_t size);
> > +int vhost_rpmsg_finish_unlock(struct vhost_rpmsg *vr,
> > +			      struct vhost_rpmsg_iter *iter);
> > +
> > +#endif
> > -- 
> > 2.28.0
> > 

Re: [PATCH v7 3/3] vhost: add an RPMsg API

[Guennadi Liakhovetski]

Hi Mathieu,

On Thu, Sep 17, 2020 at 04:01:38PM -0600, Mathieu Poirier wrote:
> On Thu, Sep 10, 2020 at 01:13:51PM +0200, Guennadi Liakhovetski wrote:
> > Linux supports running the RPMsg protocol over the VirtIO transport
> > protocol, but currently there is only support for VirtIO clients and
> > no support for VirtIO servers. This patch adds a vhost-based RPMsg
> > server implementation, which makes it possible to use RPMsg over
> > VirtIO between guest VMs and the host.
> 
> I now get the client/server concept you are describing above but that happened
> only after a lot of mental gymnastics.  If you drop the whole client/server
> concept and concentrate on what this patch does, things will go better.  I would
> personally go with what you have in the Kconfig: 
> 
> > +	  Vhost RPMsg API allows vhost drivers to communicate with VirtIO
> > +	  drivers on guest VMs, using the RPMsg over VirtIO protocol.
> 
> It is concise but describes exactly what this patch provide.

Ok, thanks, will try to improve.

> > Signed-off-by: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > ---
> >  drivers/vhost/Kconfig       |   7 +
> >  drivers/vhost/Makefile      |   3 +
> >  drivers/vhost/rpmsg.c       | 370 ++++++++++++++++++++++++++++++++++++
> >  drivers/vhost/vhost_rpmsg.h |  74 ++++++++
> >  4 files changed, 454 insertions(+)
> >  create mode 100644 drivers/vhost/rpmsg.c
> >  create mode 100644 drivers/vhost/vhost_rpmsg.h
> > 
> > diff --git a/drivers/vhost/Kconfig b/drivers/vhost/Kconfig
> > index 587fbae06182..ee1a19b7ab3d 100644
> > --- a/drivers/vhost/Kconfig
> > +++ b/drivers/vhost/Kconfig
> > @@ -38,6 +38,13 @@ config VHOST_NET
> >  	  To compile this driver as a module, choose M here: the module will
> >  	  be called vhost_net.
> >  
> > +config VHOST_RPMSG
> > +	tristate
> > +	select VHOST
> > +	help
> > +	  Vhost RPMsg API allows vhost drivers to communicate with VirtIO
> > +	  drivers on guest VMs, using the RPMsg over VirtIO protocol.
> > +
> 
> I suppose you intend this to be selectable from another config option?

yes.

> >  config VHOST_SCSI
> >  	tristate "VHOST_SCSI TCM fabric driver"
> >  	depends on TARGET_CORE && EVENTFD
> > diff --git a/drivers/vhost/Makefile b/drivers/vhost/Makefile
> > index f3e1897cce85..9cf459d59f97 100644
> > --- a/drivers/vhost/Makefile
> > +++ b/drivers/vhost/Makefile
> > @@ -2,6 +2,9 @@
> >  obj-$(CONFIG_VHOST_NET) += vhost_net.o
> >  vhost_net-y := net.o
> >  
> > +obj-$(CONFIG_VHOST_RPMSG) += vhost_rpmsg.o
> > +vhost_rpmsg-y := rpmsg.o
> > +
> >  obj-$(CONFIG_VHOST_SCSI) += vhost_scsi.o
> >  vhost_scsi-y := scsi.o
> >  
> > diff --git a/drivers/vhost/rpmsg.c b/drivers/vhost/rpmsg.c
> > new file mode 100644
> > index 000000000000..0ddee5b5f017
> > --- /dev/null
> > +++ b/drivers/vhost/rpmsg.c
> > @@ -0,0 +1,370 @@
> > +// SPDX-License-Identifier: GPL-2.0-only
> > +/*
> > + * Copyright(c) 2020 Intel Corporation. All rights reserved.
> > + *
> > + * Author: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > + *
> > + * Vhost RPMsg VirtIO interface provides a set of functions to be used on the
> > + * host side as a counterpart to the guest side RPMsg VirtIO API, provided by
> > + * drivers/rpmsg/virtio_rpmsg_bus.c. This API can be used by any vhost driver to
> > + * handle RPMsg specific virtqueue processing.
> > + * Vhost drivers, using this API will use their own VirtIO device IDs, that
> > + * should then also be added to the ID table in virtio_rpmsg_bus.c
> > + */
> > +
> > +#include <linux/compat.h>
> > +#include <linux/file.h>
> > +#include <linux/miscdevice.h>
> 
> As far as I can tell the above two are not needed.

Look like left-over, will remove.

> > +#include <linux/module.h>
> > +#include <linux/mutex.h>
> > +#include <linux/rpmsg/virtio.h>
> > +#include <linux/vhost.h>
> > +#include <uapi/linux/rpmsg.h>
> > +
> > +#include "vhost.h"
> > +#include "vhost_rpmsg.h"
> > +
> > +/*
> > + * All virtio-rpmsg virtual queue kicks always come with just one buffer -
> > + * either input or output, but we can also handle split messages
> > + */
> > +static int vhost_rpmsg_get_msg(struct vhost_virtqueue *vq, unsigned int *cnt)
> > +{
> > +	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
> > +	unsigned int out, in;
> > +	int head = vhost_get_vq_desc(vq, vq->iov, ARRAY_SIZE(vq->iov), &out, &in,
> > +				     NULL, NULL);
> > +	if (head < 0) {
> > +		vq_err(vq, "%s(): error %d getting buffer\n",
> > +		       __func__, head);
> > +		return head;
> > +	}
> > +
> > +	/* Nothing new? */
> > +	if (head == vq->num)
> > +		return head;
> > +
> > +	if (vq == &vr->vq[VIRTIO_RPMSG_RESPONSE]) {
> > +		if (out) {
> > +			vq_err(vq, "%s(): invalid %d output in response queue\n",
> > +			       __func__, out);
> > +			goto return_buf;
> > +		}
> > +
> > +		*cnt = in;
> > +	}
> > +
> > +	if (vq == &vr->vq[VIRTIO_RPMSG_REQUEST]) {
> > +		if (in) {
> > +			vq_err(vq, "%s(): invalid %d input in request queue\n",
> > +		       __func__, in);
> > +			goto return_buf;
> > +		}
> > +
> > +		*cnt = out;
> > +	}
> > +
> > +	return head;
> > +
> > +return_buf:
> > +	vhost_add_used(vq, head, 0);
> > +
> > +	return -EINVAL;
> > +}
> > +
> > +static const struct vhost_rpmsg_ept *vhost_rpmsg_ept_find(struct vhost_rpmsg *vr, int addr)
> > +{
> > +	unsigned int i;
> > +
> > +	for (i = 0; i < vr->n_epts; i++)
> > +		if (vr->ept[i].addr == addr)
> > +			return vr->ept + i;
> > +
> > +	return NULL;
> > +}
> > +
> > +/*
> > + * if len < 0, then for reading a request, the complete virtual queue buffer
> > + * size is prepared, for sending a response, the length in the iterator is used
> > + */
> > +int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> > +			   unsigned int qid, ssize_t len)
> > +	__acquires(vq->mutex)
> > +{
> > +	struct vhost_virtqueue *vq = vr->vq + qid;
> > +	unsigned int cnt;
> > +	ssize_t ret;
> > +	size_t tmp;
> > +
> > +	if (qid >= VIRTIO_RPMSG_NUM_OF_VQS)
> > +		return -EINVAL;
> > +
> > +	iter->vq = vq;
> > +
> > +	mutex_lock(&vq->mutex);
> > +	vhost_disable_notify(&vr->dev, vq);
> > +
> > +	iter->head = vhost_rpmsg_get_msg(vq, &cnt);
> > +	if (iter->head == vq->num)
> > +		iter->head = -EAGAIN;
> > +
> > +	if (iter->head < 0) {
> > +		ret = iter->head;
> > +		goto unlock;
> > +	}
> > +
> > +	tmp = iov_length(vq->iov, cnt);
> > +	if (tmp < sizeof(iter->rhdr)) {
> > +		vq_err(vq, "%s(): size %zu too small\n", __func__, tmp);
> > +		ret = -ENOBUFS;
> > +		goto return_buf;
> > +	}
> > +
> > +	switch (qid) {
> > +	case VIRTIO_RPMSG_REQUEST:
> > +		if (len >= 0) {
> > +			if (tmp < sizeof(iter->rhdr) + len) {
> > +				ret = -ENOBUFS;
> > +				goto return_buf;
> > +			}
> > +
> > +			tmp = len + sizeof(iter->rhdr);
> > +		}
> > +
> > +		/* len is now the size of the payload */
> > +		iov_iter_init(&iter->iov_iter, WRITE, vq->iov, cnt, tmp);
> > +
> > +		/* Read the RPMSG header with endpoint addresses */
> > +		tmp = copy_from_iter(&iter->rhdr, sizeof(iter->rhdr), &iter->iov_iter);
> > +		if (tmp != sizeof(iter->rhdr)) {
> > +			vq_err(vq, "%s(): got %zu instead of %zu\n", __func__,
> > +			       tmp, sizeof(iter->rhdr));
> > +			ret = -EIO;
> > +			goto return_buf;
> > +		}
> > +
> > +		iter->ept = vhost_rpmsg_ept_find(vr, vhost32_to_cpu(vq, iter->rhdr.dst));
> > +		if (!iter->ept) {
> > +			vq_err(vq, "%s(): no endpoint with address %d\n",
> > +			       __func__, vhost32_to_cpu(vq, iter->rhdr.dst));
> > +			ret = -ENOENT;
> > +			goto return_buf;
> > +		}
> > +
> > +		/* Let the endpoint read the payload */
> > +		if (iter->ept->read) {
> > +			ret = iter->ept->read(vr, iter);
> > +			if (ret < 0)
> > +				goto return_buf;
> > +
> > +			iter->rhdr.len = cpu_to_vhost16(vq, ret);
> > +		} else {
> > +			iter->rhdr.len = 0;
> > +		}
> > +
> > +		/* Prepare for the response phase */
> > +		iter->rhdr.dst = iter->rhdr.src;
> > +		iter->rhdr.src = cpu_to_vhost32(vq, iter->ept->addr);
> 
> I'm a little puzzled here - what will the response look like?  And why is it
> prepared here?  From what I can see doing so introduces coupling with function
> handle_rpmsg_req_single().  I think confirmation of reception should be handled
> by endpoints rather than in the core. 

RPMsg always contain a header, so we keep the header in the iterator. If the 
caller wants to reply to the request, the easiest way to do that is to reuse the 
iterator. In that case obviously you have to swap source and destination 
addresses. This can be done either in the request handler of the API, or by the 
caller, or in the API response handler. It would be silly to have the user do 
that, that would be repeated code. But I agree, it's a bit unclean to modify the 
header before returning it to the user, without knowing, whether the user will 
use it, in which case it might be surprised to see most fields from the request 
unchanged and only addresses swapped. I'll move this to response with a check 
for a reused iterator.

> > +
> > +		break;
> > +	case VIRTIO_RPMSG_RESPONSE:
> > +		if (!iter->ept && iter->rhdr.dst != cpu_to_vhost32(vq, RPMSG_NS_ADDR)) {
> > +			/*
> > +			 * Usually the iterator is configured when processing a
> > +			 * message on the request queue, but it's also possible
> > +			 * to send a message on the response queue without a
> > +			 * preceding request, in that case the iterator must
> > +			 * contain source and destination addresses.
> > +			 */
> > +			iter->ept = vhost_rpmsg_ept_find(vr, vhost32_to_cpu(vq, iter->rhdr.src));
> > +			if (!iter->ept) {
> > +				ret = -ENOENT;
> > +				goto return_buf;
> > +			}
> > +		}
> > +
> > +		if (len >= 0) {
> > +			if (tmp < sizeof(iter->rhdr) + len) {
> > +				ret = -ENOBUFS;
> > +				goto return_buf;
> > +			}
> > +
> > +			iter->rhdr.len = cpu_to_vhost16(vq, len);
> > +			tmp = len + sizeof(iter->rhdr);
> > +		}
> > +
> > +		/* len is now the size of the payload */
> > +		iov_iter_init(&iter->iov_iter, READ, vq->iov, cnt, tmp);
> > +
> > +		/* Write the RPMSG header with endpoint addresses */
> > +		tmp = copy_to_iter(&iter->rhdr, sizeof(iter->rhdr), &iter->iov_iter);
> > +		if (tmp != sizeof(iter->rhdr)) {
> > +			ret = -EIO;
> > +			goto return_buf;
> > +		}
> > +
> > +		/* Let the endpoint write the payload */
> 
> I would specifically mention that namespace payloads are taken care of by
> vhost_rpmsg_ns_announce().  That makes it easier for people to connect the dots. 

Ok

> > +		if (iter->ept && iter->ept->write) {
> > +			ret = iter->ept->write(vr, iter);
> > +			if (ret < 0)
> > +				goto return_buf;
> > +		}
> > +
> > +		break;
> > +	}
> > +
> > +	return 0;
> > +
> > +return_buf:
> > +	vhost_add_used(vq, iter->head, 0);
> > +unlock:
> > +	vhost_enable_notify(&vr->dev, vq);
> > +	mutex_unlock(&vq->mutex);
> > +
> > +	return ret;
> > +}
> > +EXPORT_SYMBOL_GPL(vhost_rpmsg_start_lock);
> > +
> > +size_t vhost_rpmsg_copy(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> > +			void *data, size_t size)
> > +{
> > +	/*
> > +	 * We could check for excess data, but copy_{to,from}_iter() don't do
> > +	 * that either
> > +	 */
> > +	if (iter->vq == vr->vq + VIRTIO_RPMSG_RESPONSE)
> > +		return copy_to_iter(data, size, &iter->iov_iter);
> > +
> > +	return copy_from_iter(data, size, &iter->iov_iter);
> > +}
> > +EXPORT_SYMBOL_GPL(vhost_rpmsg_copy);
> > +
> > +int vhost_rpmsg_finish_unlock(struct vhost_rpmsg *vr,
> > +			      struct vhost_rpmsg_iter *iter)
> > +	__releases(vq->mutex)
> > +{
> > +	if (iter->head >= 0)
> > +		vhost_add_used_and_signal(iter->vq->dev, iter->vq, iter->head,
> > +					  vhost16_to_cpu(iter->vq, iter->rhdr.len) +
> > +					  sizeof(iter->rhdr));
> > +
> > +	vhost_enable_notify(&vr->dev, iter->vq);
> > +	mutex_unlock(&iter->vq->mutex);
> > +
> > +	return iter->head;
> > +}
> > +EXPORT_SYMBOL_GPL(vhost_rpmsg_finish_unlock);
> > +
> > +/*
> > + * Return false to terminate the external loop only if we fail to obtain either
> > + * a request or a response buffer
> > + */
> > +static bool handle_rpmsg_req_single(struct vhost_rpmsg *vr,
> > +				    struct vhost_virtqueue *vq)
> > +{
> > +	struct vhost_rpmsg_iter iter;
> > +	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_REQUEST, -EINVAL);
> > +	if (!ret)
> > +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > +	if (ret < 0) {
> > +		if (ret != -EAGAIN)
> > +			vq_err(vq, "%s(): RPMSG processing failed %d\n",
> > +			       __func__, ret);
> > +		return false;
> > +	}
> > +
> > +	if (!iter.ept->write)
> > +		return true;
> > +
> > +	ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, -EINVAL);
> > +	if (!ret)
> > +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > +	if (ret < 0) {
> > +		vq_err(vq, "%s(): RPMSG finalising failed %d\n", __func__, ret);
> > +		return false;
> > +	}
> 
> As I said before dealing with the "response" queue here seems to be introducing
> coupling with vhost_rpmsg_start_lock()...  Endpoints should be doing that.

Sorry, could you elaborate a bit, what do you mean by coupling?

> > +
> > +	return true;
> > +}
> > +
> > +static void handle_rpmsg_req_kick(struct vhost_work *work)
> > +{
> > +	struct vhost_virtqueue *vq = container_of(work, struct vhost_virtqueue,
> > +						  poll.work);
> > +	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
> > +
> > +	while (handle_rpmsg_req_single(vr, vq))
> > +		;
> > +}
> > +
> > +/*
> > + * initialise two virtqueues with an array of endpoints,
> > + * request and response callbacks
> > + */
> > +void vhost_rpmsg_init(struct vhost_rpmsg *vr, const struct vhost_rpmsg_ept *ept,
> > +		      unsigned int n_epts)
> > +{
> > +	unsigned int i;
> > +
> > +	for (i = 0; i < ARRAY_SIZE(vr->vq); i++)
> > +		vr->vq_p[i] = &vr->vq[i];
> > +
> > +	/* vq[0]: host -> guest, vq[1]: host <- guest */
> > +	vr->vq[VIRTIO_RPMSG_REQUEST].handle_kick = handle_rpmsg_req_kick;
> > +	vr->vq[VIRTIO_RPMSG_RESPONSE].handle_kick = NULL;
> 
> The comment depicts vq[0] followed by vq[1] but the code initialise vq[1] before
> vq[0], which is wildly confusing.  At the very least this should be: 

Nobody should care which of those is 0 and which is 1 :-) But indeed you have a point, 
that the protocol isn't strictly request-response based, the host can also send 
messages to the guest without preceding requests. So, TX / RX should be a better fit.

> 
> 	vr->vq[VIRTIO_RPMSG_RESPONSE].handle_kick = NULL;
> 	vr->vq[VIRTIO_RPMSG_REQUEST].handle_kick = handle_rpmsg_req_kick;
> 
> And even better:
> 
>         /* See configuration of *vq_cbs[] in rpmsg_probe()  */
> 	vr->vq[VIRTIO_RPMSG_TX].handle_kick = NULL;
> 	vr->vq[VIRTIO_RPMSG_RX].handle_kick = handle_rpmsg_req_kick;
> 
> > +
> > +	vr->ept = ept;
> > +	vr->n_epts = n_epts;
> > +
> > +	vhost_dev_init(&vr->dev, vr->vq_p, VIRTIO_RPMSG_NUM_OF_VQS,
> > +		       UIO_MAXIOV, 0, 0, true, NULL);
> > +}
> > +EXPORT_SYMBOL_GPL(vhost_rpmsg_init);
> > +
> > +void vhost_rpmsg_destroy(struct vhost_rpmsg *vr)
> > +{
> > +	if (vhost_dev_has_owner(&vr->dev))
> > +		vhost_poll_flush(&vr->vq[VIRTIO_RPMSG_REQUEST].poll);
> > +
> > +	vhost_dev_cleanup(&vr->dev);
> > +}
> > +EXPORT_SYMBOL_GPL(vhost_rpmsg_destroy);
> > +
> > +/* send namespace */
> > +int vhost_rpmsg_ns_announce(struct vhost_rpmsg *vr, const char *name, unsigned int src)
> > +{
> > +	struct vhost_virtqueue *vq = &vr->vq[VIRTIO_RPMSG_RESPONSE];
> > +	struct vhost_rpmsg_iter iter = {
> > +		.rhdr = {
> > +			.src = 0,
> > +			.dst = cpu_to_vhost32(vq, RPMSG_NS_ADDR),
> > +		},
> > +	};
> > +	struct rpmsg_ns_msg ns = {
> > +		.addr = cpu_to_vhost32(vq, src),
> > +		.flags = cpu_to_vhost32(vq, RPMSG_NS_CREATE), /* for rpmsg_ns_cb() */
> > +	};
> 
> Here we have to assume the source can be found in the endpoints registered in
> vhost_rpmsg_init().  I would put a check to make sure that is the case and
> return an error otherwise. 

Ok, will do.

> > +	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, sizeof(ns));
> > +
> > +	if (ret < 0)
> > +		return ret;
> > +
> > +	strlcpy(ns.name, name, sizeof(ns.name));
> > +
> > +	ret = vhost_rpmsg_copy(vr, &iter, &ns, sizeof(ns));
> > +	if (ret != sizeof(ns))
> > +		vq_err(iter.vq, "%s(): added %d instead of %zu bytes\n",
> > +		       __func__, ret, sizeof(ns));
> > +
> > +	ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > +	if (ret < 0)
> > +		vq_err(iter.vq, "%s(): namespace announcement failed: %d\n",
> > +		       __func__, ret);
> > +
> > +	return ret;
> > +}
> > +EXPORT_SYMBOL_GPL(vhost_rpmsg_ns_announce);
> > +
> > +MODULE_LICENSE("GPL v2");
> > +MODULE_AUTHOR("Intel, Inc.");
> > +MODULE_DESCRIPTION("Vhost RPMsg API");
> > diff --git a/drivers/vhost/vhost_rpmsg.h b/drivers/vhost/vhost_rpmsg.h
> > new file mode 100644
> > index 000000000000..c020ea14cd16
> > --- /dev/null
> > +++ b/drivers/vhost/vhost_rpmsg.h
> > @@ -0,0 +1,74 @@
> > +/* SPDX-License-Identifier: GPL-2.0 */
> > +/*
> > + * Copyright(c) 2020 Intel Corporation. All rights reserved.
> > + *
> > + * Author: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > + */
> > +
> > +#ifndef VHOST_RPMSG_H
> > +#define VHOST_RPMSG_H
> > +
> > +#include <linux/rpmsg/virtio.h>
> > +#include <linux/uio.h>
> > +
> > +#include "vhost.h"
> > +
> > +/* RPMsg uses two VirtQueues: one for each direction */
> > +enum {
> > +	VIRTIO_RPMSG_RESPONSE,	/* RPMsg response (host->guest) buffers */
> > +	VIRTIO_RPMSG_REQUEST,	/* RPMsg request (guest->host) buffers */
> 
> As I said above things would be much clearer if this was VIRTIO_RPMSG_TX and
> VIRTIO_RPMSG_RX.

Ack.

> I won't be commenting on the mechanic needed to access and send information on
> the virtqueues as it is completely foreign to me.  Other than the above I think
> this is going somewhere. 

I'll wait for your clarifications about "coupling" and send a v8.

Thanks for the comments so far
Guennadi

> Thanks,
> Mathieu
> 
> > +	/* Keep last */
> > +	VIRTIO_RPMSG_NUM_OF_VQS,
> > +};
> > +
> > +struct vhost_rpmsg_ept;
> > +
> > +struct vhost_rpmsg_iter {
> > +	struct iov_iter iov_iter;
> > +	struct rpmsg_hdr rhdr;
> > +	struct vhost_virtqueue *vq;
> > +	const struct vhost_rpmsg_ept *ept;
> > +	int head;
> > +	void *priv;
> > +};
> > +
> > +struct vhost_rpmsg {
> > +	struct vhost_dev dev;
> > +	struct vhost_virtqueue vq[VIRTIO_RPMSG_NUM_OF_VQS];
> > +	struct vhost_virtqueue *vq_p[VIRTIO_RPMSG_NUM_OF_VQS];
> > +	const struct vhost_rpmsg_ept *ept;
> > +	unsigned int n_epts;
> > +};
> > +
> > +struct vhost_rpmsg_ept {
> > +	ssize_t (*read)(struct vhost_rpmsg *, struct vhost_rpmsg_iter *);
> > +	ssize_t (*write)(struct vhost_rpmsg *, struct vhost_rpmsg_iter *);
> > +	int addr;
> > +};
> > +
> > +static inline size_t vhost_rpmsg_iter_len(const struct vhost_rpmsg_iter *iter)
> > +{
> > +	return iter->rhdr.len;
> > +}
> > +
> > +#define VHOST_RPMSG_ITER(_vq, _src, _dst) {			\
> > +	.rhdr = {						\
> > +			.src = cpu_to_vhost32(_vq, _src),	\
> > +			.dst = cpu_to_vhost32(_vq, _dst),	\
> > +		},						\
> > +	}
> > +
> > +void vhost_rpmsg_init(struct vhost_rpmsg *vr, const struct vhost_rpmsg_ept *ept,
> > +		      unsigned int n_epts);
> > +void vhost_rpmsg_destroy(struct vhost_rpmsg *vr);
> > +int vhost_rpmsg_ns_announce(struct vhost_rpmsg *vr, const char *name,
> > +			    unsigned int src);
> > +int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr,
> > +			   struct vhost_rpmsg_iter *iter,
> > +			   unsigned int qid, ssize_t len);
> > +size_t vhost_rpmsg_copy(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> > +			void *data, size_t size);
> > +int vhost_rpmsg_finish_unlock(struct vhost_rpmsg *vr,
> > +			      struct vhost_rpmsg_iter *iter);
> > +
> > +#endif
> > -- 
> > 2.28.0
> > 
_______________________________________________
Virtualization mailing list
Virtualization@lists.linux-foundation.org
https://lists.linuxfoundation.org/mailman/listinfo/virtualization

Re: [PATCH v7 3/3] vhost: add an RPMsg API

[Mathieu Poirier]

Good morning,

On Fri, Sep 18, 2020 at 11:02:29AM +0200, Guennadi Liakhovetski wrote:
> Hi Mathieu,
> 
> On Thu, Sep 17, 2020 at 04:01:38PM -0600, Mathieu Poirier wrote:
> > On Thu, Sep 10, 2020 at 01:13:51PM +0200, Guennadi Liakhovetski wrote:
> > > Linux supports running the RPMsg protocol over the VirtIO transport
> > > protocol, but currently there is only support for VirtIO clients and
> > > no support for VirtIO servers. This patch adds a vhost-based RPMsg
> > > server implementation, which makes it possible to use RPMsg over
> > > VirtIO between guest VMs and the host.
> > 
> > I now get the client/server concept you are describing above but that happened
> > only after a lot of mental gymnastics.  If you drop the whole client/server
> > concept and concentrate on what this patch does, things will go better.  I would
> > personally go with what you have in the Kconfig: 
> > 
> > > +	  Vhost RPMsg API allows vhost drivers to communicate with VirtIO
> > > +	  drivers on guest VMs, using the RPMsg over VirtIO protocol.
> > 
> > It is concise but describes exactly what this patch provide.
> 
> Ok, thanks, will try to improve.
> 
> > > Signed-off-by: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > > ---
> > >  drivers/vhost/Kconfig       |   7 +
> > >  drivers/vhost/Makefile      |   3 +
> > >  drivers/vhost/rpmsg.c       | 370 ++++++++++++++++++++++++++++++++++++
> > >  drivers/vhost/vhost_rpmsg.h |  74 ++++++++
> > >  4 files changed, 454 insertions(+)
> > >  create mode 100644 drivers/vhost/rpmsg.c
> > >  create mode 100644 drivers/vhost/vhost_rpmsg.h
> > > 
> > > diff --git a/drivers/vhost/Kconfig b/drivers/vhost/Kconfig
> > > index 587fbae06182..ee1a19b7ab3d 100644
> > > --- a/drivers/vhost/Kconfig
> > > +++ b/drivers/vhost/Kconfig
> > > @@ -38,6 +38,13 @@ config VHOST_NET
> > >  	  To compile this driver as a module, choose M here: the module will
> > >  	  be called vhost_net.
> > >  
> > > +config VHOST_RPMSG
> > > +	tristate
> > > +	select VHOST
> > > +	help
> > > +	  Vhost RPMsg API allows vhost drivers to communicate with VirtIO
> > > +	  drivers on guest VMs, using the RPMsg over VirtIO protocol.
> > > +
> > 
> > I suppose you intend this to be selectable from another config option?
> 
> yes.
> 
> > >  config VHOST_SCSI
> > >  	tristate "VHOST_SCSI TCM fabric driver"
> > >  	depends on TARGET_CORE && EVENTFD
> > > diff --git a/drivers/vhost/Makefile b/drivers/vhost/Makefile
> > > index f3e1897cce85..9cf459d59f97 100644
> > > --- a/drivers/vhost/Makefile
> > > +++ b/drivers/vhost/Makefile
> > > @@ -2,6 +2,9 @@
> > >  obj-$(CONFIG_VHOST_NET) += vhost_net.o
> > >  vhost_net-y := net.o
> > >  
> > > +obj-$(CONFIG_VHOST_RPMSG) += vhost_rpmsg.o
> > > +vhost_rpmsg-y := rpmsg.o
> > > +
> > >  obj-$(CONFIG_VHOST_SCSI) += vhost_scsi.o
> > >  vhost_scsi-y := scsi.o
> > >  
> > > diff --git a/drivers/vhost/rpmsg.c b/drivers/vhost/rpmsg.c
> > > new file mode 100644
> > > index 000000000000..0ddee5b5f017
> > > --- /dev/null
> > > +++ b/drivers/vhost/rpmsg.c
> > > @@ -0,0 +1,370 @@
> > > +// SPDX-License-Identifier: GPL-2.0-only
> > > +/*
> > > + * Copyright(c) 2020 Intel Corporation. All rights reserved.
> > > + *
> > > + * Author: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > > + *
> > > + * Vhost RPMsg VirtIO interface provides a set of functions to be used on the
> > > + * host side as a counterpart to the guest side RPMsg VirtIO API, provided by
> > > + * drivers/rpmsg/virtio_rpmsg_bus.c. This API can be used by any vhost driver to
> > > + * handle RPMsg specific virtqueue processing.
> > > + * Vhost drivers, using this API will use their own VirtIO device IDs, that
> > > + * should then also be added to the ID table in virtio_rpmsg_bus.c
> > > + */
> > > +
> > > +#include <linux/compat.h>
> > > +#include <linux/file.h>
> > > +#include <linux/miscdevice.h>
> > 
> > As far as I can tell the above two are not needed.
> 
> Look like left-over, will remove.
> 
> > > +#include <linux/module.h>
> > > +#include <linux/mutex.h>
> > > +#include <linux/rpmsg/virtio.h>
> > > +#include <linux/vhost.h>
> > > +#include <uapi/linux/rpmsg.h>
> > > +
> > > +#include "vhost.h"
> > > +#include "vhost_rpmsg.h"
> > > +
> > > +/*
> > > + * All virtio-rpmsg virtual queue kicks always come with just one buffer -
> > > + * either input or output, but we can also handle split messages
> > > + */
> > > +static int vhost_rpmsg_get_msg(struct vhost_virtqueue *vq, unsigned int *cnt)
> > > +{
> > > +	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
> > > +	unsigned int out, in;
> > > +	int head = vhost_get_vq_desc(vq, vq->iov, ARRAY_SIZE(vq->iov), &out, &in,
> > > +				     NULL, NULL);
> > > +	if (head < 0) {
> > > +		vq_err(vq, "%s(): error %d getting buffer\n",
> > > +		       __func__, head);
> > > +		return head;
> > > +	}
> > > +
> > > +	/* Nothing new? */
> > > +	if (head == vq->num)
> > > +		return head;
> > > +
> > > +	if (vq == &vr->vq[VIRTIO_RPMSG_RESPONSE]) {
> > > +		if (out) {
> > > +			vq_err(vq, "%s(): invalid %d output in response queue\n",
> > > +			       __func__, out);
> > > +			goto return_buf;
> > > +		}
> > > +
> > > +		*cnt = in;
> > > +	}
> > > +
> > > +	if (vq == &vr->vq[VIRTIO_RPMSG_REQUEST]) {
> > > +		if (in) {
> > > +			vq_err(vq, "%s(): invalid %d input in request queue\n",
> > > +		       __func__, in);
> > > +			goto return_buf;
> > > +		}
> > > +
> > > +		*cnt = out;
> > > +	}
> > > +
> > > +	return head;
> > > +
> > > +return_buf:
> > > +	vhost_add_used(vq, head, 0);
> > > +
> > > +	return -EINVAL;
> > > +}
> > > +
> > > +static const struct vhost_rpmsg_ept *vhost_rpmsg_ept_find(struct vhost_rpmsg *vr, int addr)
> > > +{
> > > +	unsigned int i;
> > > +
> > > +	for (i = 0; i < vr->n_epts; i++)
> > > +		if (vr->ept[i].addr == addr)
> > > +			return vr->ept + i;
> > > +
> > > +	return NULL;
> > > +}
> > > +
> > > +/*
> > > + * if len < 0, then for reading a request, the complete virtual queue buffer
> > > + * size is prepared, for sending a response, the length in the iterator is used
> > > + */
> > > +int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> > > +			   unsigned int qid, ssize_t len)
> > > +	__acquires(vq->mutex)
> > > +{
> > > +	struct vhost_virtqueue *vq = vr->vq + qid;
> > > +	unsigned int cnt;
> > > +	ssize_t ret;
> > > +	size_t tmp;
> > > +
> > > +	if (qid >= VIRTIO_RPMSG_NUM_OF_VQS)
> > > +		return -EINVAL;
> > > +
> > > +	iter->vq = vq;
> > > +
> > > +	mutex_lock(&vq->mutex);
> > > +	vhost_disable_notify(&vr->dev, vq);
> > > +
> > > +	iter->head = vhost_rpmsg_get_msg(vq, &cnt);
> > > +	if (iter->head == vq->num)
> > > +		iter->head = -EAGAIN;
> > > +
> > > +	if (iter->head < 0) {
> > > +		ret = iter->head;
> > > +		goto unlock;
> > > +	}
> > > +
> > > +	tmp = iov_length(vq->iov, cnt);
> > > +	if (tmp < sizeof(iter->rhdr)) {
> > > +		vq_err(vq, "%s(): size %zu too small\n", __func__, tmp);
> > > +		ret = -ENOBUFS;
> > > +		goto return_buf;
> > > +	}
> > > +
> > > +	switch (qid) {
> > > +	case VIRTIO_RPMSG_REQUEST:
> > > +		if (len >= 0) {
> > > +			if (tmp < sizeof(iter->rhdr) + len) {
> > > +				ret = -ENOBUFS;
> > > +				goto return_buf;
> > > +			}
> > > +
> > > +			tmp = len + sizeof(iter->rhdr);
> > > +		}
> > > +
> > > +		/* len is now the size of the payload */
> > > +		iov_iter_init(&iter->iov_iter, WRITE, vq->iov, cnt, tmp);
> > > +
> > > +		/* Read the RPMSG header with endpoint addresses */
> > > +		tmp = copy_from_iter(&iter->rhdr, sizeof(iter->rhdr), &iter->iov_iter);
> > > +		if (tmp != sizeof(iter->rhdr)) {
> > > +			vq_err(vq, "%s(): got %zu instead of %zu\n", __func__,
> > > +			       tmp, sizeof(iter->rhdr));
> > > +			ret = -EIO;
> > > +			goto return_buf;
> > > +		}
> > > +
> > > +		iter->ept = vhost_rpmsg_ept_find(vr, vhost32_to_cpu(vq, iter->rhdr.dst));
> > > +		if (!iter->ept) {
> > > +			vq_err(vq, "%s(): no endpoint with address %d\n",
> > > +			       __func__, vhost32_to_cpu(vq, iter->rhdr.dst));
> > > +			ret = -ENOENT;
> > > +			goto return_buf;
> > > +		}
> > > +
> > > +		/* Let the endpoint read the payload */
> > > +		if (iter->ept->read) {
> > > +			ret = iter->ept->read(vr, iter);
> > > +			if (ret < 0)
> > > +				goto return_buf;
> > > +
> > > +			iter->rhdr.len = cpu_to_vhost16(vq, ret);
> > > +		} else {
> > > +			iter->rhdr.len = 0;
> > > +		}
> > > +
> > > +		/* Prepare for the response phase */
> > > +		iter->rhdr.dst = iter->rhdr.src;
> > > +		iter->rhdr.src = cpu_to_vhost32(vq, iter->ept->addr);
> > 
> > I'm a little puzzled here - what will the response look like?  And why is it
> > prepared here?  From what I can see doing so introduces coupling with function
> > handle_rpmsg_req_single().  I think confirmation of reception should be handled
> > by endpoints rather than in the core. 
> 
> RPMsg always contain a header, so we keep the header in the iterator. If the 
> caller wants to reply to the request, the easiest way to do that is to reuse the 
> iterator. In that case obviously you have to swap source and destination 
> addresses. This can be done either in the request handler of the API, or by the 
> caller, or in the API response handler. It would be silly to have the user do 
> that, that would be repeated code. But I agree, it's a bit unclean to modify the 
> header before returning it to the user, without knowing, whether the user will 
> use it, in which case it might be surprised to see most fields from the request 
> unchanged and only addresses swapped. I'll move this to response with a check 
> for a reused iterator.
> 
> > > +
> > > +		break;
> > > +	case VIRTIO_RPMSG_RESPONSE:
> > > +		if (!iter->ept && iter->rhdr.dst != cpu_to_vhost32(vq, RPMSG_NS_ADDR)) {
> > > +			/*
> > > +			 * Usually the iterator is configured when processing a
> > > +			 * message on the request queue, but it's also possible
> > > +			 * to send a message on the response queue without a
> > > +			 * preceding request, in that case the iterator must
> > > +			 * contain source and destination addresses.
> > > +			 */
> > > +			iter->ept = vhost_rpmsg_ept_find(vr, vhost32_to_cpu(vq, iter->rhdr.src));
> > > +			if (!iter->ept) {
> > > +				ret = -ENOENT;
> > > +				goto return_buf;
> > > +			}
> > > +		}
> > > +
> > > +		if (len >= 0) {
> > > +			if (tmp < sizeof(iter->rhdr) + len) {
> > > +				ret = -ENOBUFS;
> > > +				goto return_buf;
> > > +			}
> > > +
> > > +			iter->rhdr.len = cpu_to_vhost16(vq, len);
> > > +			tmp = len + sizeof(iter->rhdr);
> > > +		}
> > > +
> > > +		/* len is now the size of the payload */
> > > +		iov_iter_init(&iter->iov_iter, READ, vq->iov, cnt, tmp);
> > > +
> > > +		/* Write the RPMSG header with endpoint addresses */
> > > +		tmp = copy_to_iter(&iter->rhdr, sizeof(iter->rhdr), &iter->iov_iter);
> > > +		if (tmp != sizeof(iter->rhdr)) {
> > > +			ret = -EIO;
> > > +			goto return_buf;
> > > +		}
> > > +
> > > +		/* Let the endpoint write the payload */
> > 
> > I would specifically mention that namespace payloads are taken care of by
> > vhost_rpmsg_ns_announce().  That makes it easier for people to connect the dots. 
> 
> Ok
> 
> > > +		if (iter->ept && iter->ept->write) {
> > > +			ret = iter->ept->write(vr, iter);
> > > +			if (ret < 0)
> > > +				goto return_buf;
> > > +		}
> > > +
> > > +		break;
> > > +	}
> > > +
> > > +	return 0;
> > > +
> > > +return_buf:
> > > +	vhost_add_used(vq, iter->head, 0);
> > > +unlock:
> > > +	vhost_enable_notify(&vr->dev, vq);
> > > +	mutex_unlock(&vq->mutex);
> > > +
> > > +	return ret;
> > > +}
> > > +EXPORT_SYMBOL_GPL(vhost_rpmsg_start_lock);
> > > +
> > > +size_t vhost_rpmsg_copy(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> > > +			void *data, size_t size)
> > > +{
> > > +	/*
> > > +	 * We could check for excess data, but copy_{to,from}_iter() don't do
> > > +	 * that either
> > > +	 */
> > > +	if (iter->vq == vr->vq + VIRTIO_RPMSG_RESPONSE)
> > > +		return copy_to_iter(data, size, &iter->iov_iter);
> > > +
> > > +	return copy_from_iter(data, size, &iter->iov_iter);
> > > +}
> > > +EXPORT_SYMBOL_GPL(vhost_rpmsg_copy);
> > > +
> > > +int vhost_rpmsg_finish_unlock(struct vhost_rpmsg *vr,
> > > +			      struct vhost_rpmsg_iter *iter)
> > > +	__releases(vq->mutex)
> > > +{
> > > +	if (iter->head >= 0)
> > > +		vhost_add_used_and_signal(iter->vq->dev, iter->vq, iter->head,
> > > +					  vhost16_to_cpu(iter->vq, iter->rhdr.len) +
> > > +					  sizeof(iter->rhdr));
> > > +
> > > +	vhost_enable_notify(&vr->dev, iter->vq);
> > > +	mutex_unlock(&iter->vq->mutex);
> > > +
> > > +	return iter->head;
> > > +}
> > > +EXPORT_SYMBOL_GPL(vhost_rpmsg_finish_unlock);
> > > +
> > > +/*
> > > + * Return false to terminate the external loop only if we fail to obtain either
> > > + * a request or a response buffer
> > > + */
> > > +static bool handle_rpmsg_req_single(struct vhost_rpmsg *vr,
> > > +				    struct vhost_virtqueue *vq)
> > > +{
> > > +	struct vhost_rpmsg_iter iter;
> > > +	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_REQUEST, -EINVAL);
> > > +	if (!ret)
> > > +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > > +	if (ret < 0) {
> > > +		if (ret != -EAGAIN)
> > > +			vq_err(vq, "%s(): RPMSG processing failed %d\n",
> > > +			       __func__, ret);
> > > +		return false;
> > > +	}
> > > +
> > > +	if (!iter.ept->write)
> > > +		return true;
> > > +
> > > +	ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, -EINVAL);
> > > +	if (!ret)
> > > +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > > +	if (ret < 0) {
> > > +		vq_err(vq, "%s(): RPMSG finalising failed %d\n", __func__, ret);
> > > +		return false;
> > > +	}
> > 
> > As I said before dealing with the "response" queue here seems to be introducing
> > coupling with vhost_rpmsg_start_lock()...  Endpoints should be doing that.
> 
> Sorry, could you elaborate a bit, what do you mean by coupling?

In function vhost_rpmsg_start_lock() the rpmsg header is prepared for a response
at the end of the processing associated with the reception of a
VIRTIO_RPMSG_REQUEST.  I assumed (perhaps wrongly) that such as response was
sent here.  In that case preparing the response and sending the response should
be done at the same place.

But my assumption may be completely wrong... A better question should probably
be why is the VIRTIO_RPMSG_RESPONSE probed in handle_rpmsg_req_single()?
Shouldn't this be solely concerned with handling requests from the guest?  If
I'm wondering what is going on I expect other people will also do the same,
something that could be alleviated with more comments.

Thanks,
Mathieu

> 
> > > +
> > > +	return true;
> > > +}
> > > +
> > > +static void handle_rpmsg_req_kick(struct vhost_work *work)
> > > +{
> > > +	struct vhost_virtqueue *vq = container_of(work, struct vhost_virtqueue,
> > > +						  poll.work);
> > > +	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
> > > +
> > > +	while (handle_rpmsg_req_single(vr, vq))
> > > +		;
> > > +}
> > > +
> > > +/*
> > > + * initialise two virtqueues with an array of endpoints,
> > > + * request and response callbacks
> > > + */
> > > +void vhost_rpmsg_init(struct vhost_rpmsg *vr, const struct vhost_rpmsg_ept *ept,
> > > +		      unsigned int n_epts)
> > > +{
> > > +	unsigned int i;
> > > +
> > > +	for (i = 0; i < ARRAY_SIZE(vr->vq); i++)
> > > +		vr->vq_p[i] = &vr->vq[i];
> > > +
> > > +	/* vq[0]: host -> guest, vq[1]: host <- guest */
> > > +	vr->vq[VIRTIO_RPMSG_REQUEST].handle_kick = handle_rpmsg_req_kick;
> > > +	vr->vq[VIRTIO_RPMSG_RESPONSE].handle_kick = NULL;
> > 
> > The comment depicts vq[0] followed by vq[1] but the code initialise vq[1] before
> > vq[0], which is wildly confusing.  At the very least this should be: 
> 
> Nobody should care which of those is 0 and which is 1 :-) But indeed you have a point, 
> that the protocol isn't strictly request-response based, the host can also send 
> messages to the guest without preceding requests. So, TX / RX should be a better fit.
> 
> > 
> > 	vr->vq[VIRTIO_RPMSG_RESPONSE].handle_kick = NULL;
> > 	vr->vq[VIRTIO_RPMSG_REQUEST].handle_kick = handle_rpmsg_req_kick;
> > 
> > And even better:
> > 
> >         /* See configuration of *vq_cbs[] in rpmsg_probe()  */
> > 	vr->vq[VIRTIO_RPMSG_TX].handle_kick = NULL;
> > 	vr->vq[VIRTIO_RPMSG_RX].handle_kick = handle_rpmsg_req_kick;
> > 
> > > +
> > > +	vr->ept = ept;
> > > +	vr->n_epts = n_epts;
> > > +
> > > +	vhost_dev_init(&vr->dev, vr->vq_p, VIRTIO_RPMSG_NUM_OF_VQS,
> > > +		       UIO_MAXIOV, 0, 0, true, NULL);
> > > +}
> > > +EXPORT_SYMBOL_GPL(vhost_rpmsg_init);
> > > +
> > > +void vhost_rpmsg_destroy(struct vhost_rpmsg *vr)
> > > +{
> > > +	if (vhost_dev_has_owner(&vr->dev))
> > > +		vhost_poll_flush(&vr->vq[VIRTIO_RPMSG_REQUEST].poll);
> > > +
> > > +	vhost_dev_cleanup(&vr->dev);
> > > +}
> > > +EXPORT_SYMBOL_GPL(vhost_rpmsg_destroy);
> > > +
> > > +/* send namespace */
> > > +int vhost_rpmsg_ns_announce(struct vhost_rpmsg *vr, const char *name, unsigned int src)
> > > +{
> > > +	struct vhost_virtqueue *vq = &vr->vq[VIRTIO_RPMSG_RESPONSE];
> > > +	struct vhost_rpmsg_iter iter = {
> > > +		.rhdr = {
> > > +			.src = 0,
> > > +			.dst = cpu_to_vhost32(vq, RPMSG_NS_ADDR),
> > > +		},
> > > +	};
> > > +	struct rpmsg_ns_msg ns = {
> > > +		.addr = cpu_to_vhost32(vq, src),
> > > +		.flags = cpu_to_vhost32(vq, RPMSG_NS_CREATE), /* for rpmsg_ns_cb() */
> > > +	};
> > 
> > Here we have to assume the source can be found in the endpoints registered in
> > vhost_rpmsg_init().  I would put a check to make sure that is the case and
> > return an error otherwise. 
> 
> Ok, will do.
> 
> > > +	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, sizeof(ns));
> > > +
> > > +	if (ret < 0)
> > > +		return ret;
> > > +
> > > +	strlcpy(ns.name, name, sizeof(ns.name));
> > > +
> > > +	ret = vhost_rpmsg_copy(vr, &iter, &ns, sizeof(ns));
> > > +	if (ret != sizeof(ns))
> > > +		vq_err(iter.vq, "%s(): added %d instead of %zu bytes\n",
> > > +		       __func__, ret, sizeof(ns));
> > > +
> > > +	ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > > +	if (ret < 0)
> > > +		vq_err(iter.vq, "%s(): namespace announcement failed: %d\n",
> > > +		       __func__, ret);
> > > +
> > > +	return ret;
> > > +}
> > > +EXPORT_SYMBOL_GPL(vhost_rpmsg_ns_announce);
> > > +
> > > +MODULE_LICENSE("GPL v2");
> > > +MODULE_AUTHOR("Intel, Inc.");
> > > +MODULE_DESCRIPTION("Vhost RPMsg API");
> > > diff --git a/drivers/vhost/vhost_rpmsg.h b/drivers/vhost/vhost_rpmsg.h
> > > new file mode 100644
> > > index 000000000000..c020ea14cd16
> > > --- /dev/null
> > > +++ b/drivers/vhost/vhost_rpmsg.h
> > > @@ -0,0 +1,74 @@
> > > +/* SPDX-License-Identifier: GPL-2.0 */
> > > +/*
> > > + * Copyright(c) 2020 Intel Corporation. All rights reserved.
> > > + *
> > > + * Author: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > > + */
> > > +
> > > +#ifndef VHOST_RPMSG_H
> > > +#define VHOST_RPMSG_H
> > > +
> > > +#include <linux/rpmsg/virtio.h>
> > > +#include <linux/uio.h>
> > > +
> > > +#include "vhost.h"
> > > +
> > > +/* RPMsg uses two VirtQueues: one for each direction */
> > > +enum {
> > > +	VIRTIO_RPMSG_RESPONSE,	/* RPMsg response (host->guest) buffers */
> > > +	VIRTIO_RPMSG_REQUEST,	/* RPMsg request (guest->host) buffers */
> > 
> > As I said above things would be much clearer if this was VIRTIO_RPMSG_TX and
> > VIRTIO_RPMSG_RX.
> 
> Ack.
> 
> > I won't be commenting on the mechanic needed to access and send information on
> > the virtqueues as it is completely foreign to me.  Other than the above I think
> > this is going somewhere. 
> 
> I'll wait for your clarifications about "coupling" and send a v8.
> 
> Thanks for the comments so far
> Guennadi
> 
> > Thanks,
> > Mathieu
> > 
> > > +	/* Keep last */
> > > +	VIRTIO_RPMSG_NUM_OF_VQS,
> > > +};
> > > +
> > > +struct vhost_rpmsg_ept;
> > > +
> > > +struct vhost_rpmsg_iter {
> > > +	struct iov_iter iov_iter;
> > > +	struct rpmsg_hdr rhdr;
> > > +	struct vhost_virtqueue *vq;
> > > +	const struct vhost_rpmsg_ept *ept;
> > > +	int head;
> > > +	void *priv;
> > > +};
> > > +
> > > +struct vhost_rpmsg {
> > > +	struct vhost_dev dev;
> > > +	struct vhost_virtqueue vq[VIRTIO_RPMSG_NUM_OF_VQS];
> > > +	struct vhost_virtqueue *vq_p[VIRTIO_RPMSG_NUM_OF_VQS];
> > > +	const struct vhost_rpmsg_ept *ept;
> > > +	unsigned int n_epts;
> > > +};
> > > +
> > > +struct vhost_rpmsg_ept {
> > > +	ssize_t (*read)(struct vhost_rpmsg *, struct vhost_rpmsg_iter *);
> > > +	ssize_t (*write)(struct vhost_rpmsg *, struct vhost_rpmsg_iter *);
> > > +	int addr;
> > > +};
> > > +
> > > +static inline size_t vhost_rpmsg_iter_len(const struct vhost_rpmsg_iter *iter)
> > > +{
> > > +	return iter->rhdr.len;
> > > +}
> > > +
> > > +#define VHOST_RPMSG_ITER(_vq, _src, _dst) {			\
> > > +	.rhdr = {						\
> > > +			.src = cpu_to_vhost32(_vq, _src),	\
> > > +			.dst = cpu_to_vhost32(_vq, _dst),	\
> > > +		},						\
> > > +	}
> > > +
> > > +void vhost_rpmsg_init(struct vhost_rpmsg *vr, const struct vhost_rpmsg_ept *ept,
> > > +		      unsigned int n_epts);
> > > +void vhost_rpmsg_destroy(struct vhost_rpmsg *vr);
> > > +int vhost_rpmsg_ns_announce(struct vhost_rpmsg *vr, const char *name,
> > > +			    unsigned int src);
> > > +int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr,
> > > +			   struct vhost_rpmsg_iter *iter,
> > > +			   unsigned int qid, ssize_t len);
> > > +size_t vhost_rpmsg_copy(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> > > +			void *data, size_t size);
> > > +int vhost_rpmsg_finish_unlock(struct vhost_rpmsg *vr,
> > > +			      struct vhost_rpmsg_iter *iter);
> > > +
> > > +#endif
> > > -- 
> > > 2.28.0
> > > 

Re: [PATCH v7 3/3] vhost: add an RPMsg API

[Guennadi Liakhovetski]

Hi Mathieu,

On Fri, Sep 18, 2020 at 09:52:49AM -0600, Mathieu Poirier wrote:
> Good morning,
> 
> On Fri, Sep 18, 2020 at 11:02:29AM +0200, Guennadi Liakhovetski wrote:
> > Hi Mathieu,
> > 
> > On Thu, Sep 17, 2020 at 04:01:38PM -0600, Mathieu Poirier wrote:
> > > On Thu, Sep 10, 2020 at 01:13:51PM +0200, Guennadi Liakhovetski wrote:
> > > > Linux supports running the RPMsg protocol over the VirtIO transport
> > > > protocol, but currently there is only support for VirtIO clients and
> > > > no support for VirtIO servers. This patch adds a vhost-based RPMsg
> > > > server implementation, which makes it possible to use RPMsg over
> > > > VirtIO between guest VMs and the host.
> > > 
> > > I now get the client/server concept you are describing above but that happened
> > > only after a lot of mental gymnastics.  If you drop the whole client/server
> > > concept and concentrate on what this patch does, things will go better.  I would
> > > personally go with what you have in the Kconfig: 
> > > 
> > > > +	  Vhost RPMsg API allows vhost drivers to communicate with VirtIO
> > > > +	  drivers on guest VMs, using the RPMsg over VirtIO protocol.
> > > 
> > > It is concise but describes exactly what this patch provide.
> > 
> > Ok, thanks, will try to improve.
> > 
> > > > Signed-off-by: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > > > ---
> > > >  drivers/vhost/Kconfig       |   7 +
> > > >  drivers/vhost/Makefile      |   3 +
> > > >  drivers/vhost/rpmsg.c       | 370 ++++++++++++++++++++++++++++++++++++
> > > >  drivers/vhost/vhost_rpmsg.h |  74 ++++++++
> > > >  4 files changed, 454 insertions(+)
> > > >  create mode 100644 drivers/vhost/rpmsg.c
> > > >  create mode 100644 drivers/vhost/vhost_rpmsg.h

[snip]

> > > > diff --git a/drivers/vhost/rpmsg.c b/drivers/vhost/rpmsg.c
> > > > new file mode 100644
> > > > index 000000000000..0ddee5b5f017
> > > > --- /dev/null
> > > > +++ b/drivers/vhost/rpmsg.c
> > > > @@ -0,0 +1,370 @@

[snip]

> > > > +/*
> > > > + * Return false to terminate the external loop only if we fail to obtain either
> > > > + * a request or a response buffer
> > > > + */
> > > > +static bool handle_rpmsg_req_single(struct vhost_rpmsg *vr,
> > > > +				    struct vhost_virtqueue *vq)
> > > > +{
> > > > +	struct vhost_rpmsg_iter iter;
> > > > +	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_REQUEST, -EINVAL);
> > > > +	if (!ret)
> > > > +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > > > +	if (ret < 0) {
> > > > +		if (ret != -EAGAIN)
> > > > +			vq_err(vq, "%s(): RPMSG processing failed %d\n",
> > > > +			       __func__, ret);
> > > > +		return false;
> > > > +	}
> > > > +
> > > > +	if (!iter.ept->write)
> > > > +		return true;
> > > > +
> > > > +	ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, -EINVAL);
> > > > +	if (!ret)
> > > > +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > > > +	if (ret < 0) {
> > > > +		vq_err(vq, "%s(): RPMSG finalising failed %d\n", __func__, ret);
> > > > +		return false;
> > > > +	}
> > > 
> > > As I said before dealing with the "response" queue here seems to be introducing
> > > coupling with vhost_rpmsg_start_lock()...  Endpoints should be doing that.
> > 
> > Sorry, could you elaborate a bit, what do you mean by coupling?
> 
> In function vhost_rpmsg_start_lock() the rpmsg header is prepared for a response
> at the end of the processing associated with the reception of a
> VIRTIO_RPMSG_REQUEST.  I assumed (perhaps wrongly) that such as response was
> sent here.  In that case preparing the response and sending the response should
> be done at the same place.

This will change in the next version, in it I'll remove response preparation from 
request handling.

> But my assumption may be completely wrong... A better question should probably
> be why is the VIRTIO_RPMSG_RESPONSE probed in handle_rpmsg_req_single()?
> Shouldn't this be solely concerned with handling requests from the guest?  If
> I'm wondering what is going on I expect other people will also do the same,
> something that could be alleviated with more comments.

My RPMsg implementation supports two modes for sending data from the host (in 
VM terms) to guests: as responses to their requests and as asynchronous 
messages. If there isn't a strict request-response pattern on a certain endpont, 
you leave the .write callback NULL and then you send your messages as you please 
independent of requests. But you can also specify a .write pointer in which case 
after each request to generate a response.

In principle this response handling could be removed, but then drivers, that do 
need to respond to requests would have to schedule an asynchronous action in 
their .read callbacks to be triggered after request processing has completed.

Thanks
Guennadi

Re: [PATCH v7 3/3] vhost: add an RPMsg API

[Guennadi Liakhovetski]

Hi Mathieu,

On Fri, Sep 18, 2020 at 09:52:49AM -0600, Mathieu Poirier wrote:
> Good morning,
> 
> On Fri, Sep 18, 2020 at 11:02:29AM +0200, Guennadi Liakhovetski wrote:
> > Hi Mathieu,
> > 
> > On Thu, Sep 17, 2020 at 04:01:38PM -0600, Mathieu Poirier wrote:
> > > On Thu, Sep 10, 2020 at 01:13:51PM +0200, Guennadi Liakhovetski wrote:
> > > > Linux supports running the RPMsg protocol over the VirtIO transport
> > > > protocol, but currently there is only support for VirtIO clients and
> > > > no support for VirtIO servers. This patch adds a vhost-based RPMsg
> > > > server implementation, which makes it possible to use RPMsg over
> > > > VirtIO between guest VMs and the host.
> > > 
> > > I now get the client/server concept you are describing above but that happened
> > > only after a lot of mental gymnastics.  If you drop the whole client/server
> > > concept and concentrate on what this patch does, things will go better.  I would
> > > personally go with what you have in the Kconfig: 
> > > 
> > > > +	  Vhost RPMsg API allows vhost drivers to communicate with VirtIO
> > > > +	  drivers on guest VMs, using the RPMsg over VirtIO protocol.
> > > 
> > > It is concise but describes exactly what this patch provide.
> > 
> > Ok, thanks, will try to improve.
> > 
> > > > Signed-off-by: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > > > ---
> > > >  drivers/vhost/Kconfig       |   7 +
> > > >  drivers/vhost/Makefile      |   3 +
> > > >  drivers/vhost/rpmsg.c       | 370 ++++++++++++++++++++++++++++++++++++
> > > >  drivers/vhost/vhost_rpmsg.h |  74 ++++++++
> > > >  4 files changed, 454 insertions(+)
> > > >  create mode 100644 drivers/vhost/rpmsg.c
> > > >  create mode 100644 drivers/vhost/vhost_rpmsg.h

[snip]

> > > > diff --git a/drivers/vhost/rpmsg.c b/drivers/vhost/rpmsg.c
> > > > new file mode 100644
> > > > index 000000000000..0ddee5b5f017
> > > > --- /dev/null
> > > > +++ b/drivers/vhost/rpmsg.c
> > > > @@ -0,0 +1,370 @@

[snip]

> > > > +/*
> > > > + * Return false to terminate the external loop only if we fail to obtain either
> > > > + * a request or a response buffer
> > > > + */
> > > > +static bool handle_rpmsg_req_single(struct vhost_rpmsg *vr,
> > > > +				    struct vhost_virtqueue *vq)
> > > > +{
> > > > +	struct vhost_rpmsg_iter iter;
> > > > +	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_REQUEST, -EINVAL);
> > > > +	if (!ret)
> > > > +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > > > +	if (ret < 0) {
> > > > +		if (ret != -EAGAIN)
> > > > +			vq_err(vq, "%s(): RPMSG processing failed %d\n",
> > > > +			       __func__, ret);
> > > > +		return false;
> > > > +	}
> > > > +
> > > > +	if (!iter.ept->write)
> > > > +		return true;
> > > > +
> > > > +	ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, -EINVAL);
> > > > +	if (!ret)
> > > > +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > > > +	if (ret < 0) {
> > > > +		vq_err(vq, "%s(): RPMSG finalising failed %d\n", __func__, ret);
> > > > +		return false;
> > > > +	}
> > > 
> > > As I said before dealing with the "response" queue here seems to be introducing
> > > coupling with vhost_rpmsg_start_lock()...  Endpoints should be doing that.
> > 
> > Sorry, could you elaborate a bit, what do you mean by coupling?
> 
> In function vhost_rpmsg_start_lock() the rpmsg header is prepared for a response
> at the end of the processing associated with the reception of a
> VIRTIO_RPMSG_REQUEST.  I assumed (perhaps wrongly) that such as response was
> sent here.  In that case preparing the response and sending the response should
> be done at the same place.

This will change in the next version, in it I'll remove response preparation from 
request handling.

> But my assumption may be completely wrong... A better question should probably
> be why is the VIRTIO_RPMSG_RESPONSE probed in handle_rpmsg_req_single()?
> Shouldn't this be solely concerned with handling requests from the guest?  If
> I'm wondering what is going on I expect other people will also do the same,
> something that could be alleviated with more comments.

My RPMsg implementation supports two modes for sending data from the host (in 
VM terms) to guests: as responses to their requests and as asynchronous 
messages. If there isn't a strict request-response pattern on a certain endpont, 
you leave the .write callback NULL and then you send your messages as you please 
independent of requests. But you can also specify a .write pointer in which case 
after each request to generate a response.

In principle this response handling could be removed, but then drivers, that do 
need to respond to requests would have to schedule an asynchronous action in 
their .read callbacks to be triggered after request processing has completed.

Thanks
Guennadi
_______________________________________________
Virtualization mailing list
Virtualization@lists.linux-foundation.org
https://lists.linuxfoundation.org/mailman/listinfo/virtualization

Re: [PATCH v7 3/3] vhost: add an RPMsg API

[Mathieu Poirier]

On Fri, Sep 18, 2020 at 11:02:29AM +0200, Guennadi Liakhovetski wrote:
> Hi Mathieu,
> 
> On Thu, Sep 17, 2020 at 04:01:38PM -0600, Mathieu Poirier wrote:
> > On Thu, Sep 10, 2020 at 01:13:51PM +0200, Guennadi Liakhovetski wrote:
> > > Linux supports running the RPMsg protocol over the VirtIO transport
> > > protocol, but currently there is only support for VirtIO clients and
> > > no support for VirtIO servers. This patch adds a vhost-based RPMsg
> > > server implementation, which makes it possible to use RPMsg over
> > > VirtIO between guest VMs and the host.
> > 
> > I now get the client/server concept you are describing above but that happened
> > only after a lot of mental gymnastics.  If you drop the whole client/server
> > concept and concentrate on what this patch does, things will go better.  I would
> > personally go with what you have in the Kconfig: 
> > 
> > > +	  Vhost RPMsg API allows vhost drivers to communicate with VirtIO
> > > +	  drivers on guest VMs, using the RPMsg over VirtIO protocol.
> > 
> > It is concise but describes exactly what this patch provide.
> 
> Ok, thanks, will try to improve.
> 
> > > Signed-off-by: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > > ---
> > >  drivers/vhost/Kconfig       |   7 +
> > >  drivers/vhost/Makefile      |   3 +
> > >  drivers/vhost/rpmsg.c       | 370 ++++++++++++++++++++++++++++++++++++
> > >  drivers/vhost/vhost_rpmsg.h |  74 ++++++++
> > >  4 files changed, 454 insertions(+)
> > >  create mode 100644 drivers/vhost/rpmsg.c
> > >  create mode 100644 drivers/vhost/vhost_rpmsg.h
> > > 
> > > diff --git a/drivers/vhost/Kconfig b/drivers/vhost/Kconfig
> > > index 587fbae06182..ee1a19b7ab3d 100644
> > > --- a/drivers/vhost/Kconfig
> > > +++ b/drivers/vhost/Kconfig
> > > @@ -38,6 +38,13 @@ config VHOST_NET
> > >  	  To compile this driver as a module, choose M here: the module will
> > >  	  be called vhost_net.
> > >  
> > > +config VHOST_RPMSG
> > > +	tristate
> > > +	select VHOST
> > > +	help
> > > +	  Vhost RPMsg API allows vhost drivers to communicate with VirtIO
> > > +	  drivers on guest VMs, using the RPMsg over VirtIO protocol.
> > > +
> > 
> > I suppose you intend this to be selectable from another config option?
> 
> yes.
> 
> > >  config VHOST_SCSI
> > >  	tristate "VHOST_SCSI TCM fabric driver"
> > >  	depends on TARGET_CORE && EVENTFD
> > > diff --git a/drivers/vhost/Makefile b/drivers/vhost/Makefile
> > > index f3e1897cce85..9cf459d59f97 100644
> > > --- a/drivers/vhost/Makefile
> > > +++ b/drivers/vhost/Makefile
> > > @@ -2,6 +2,9 @@
> > >  obj-$(CONFIG_VHOST_NET) += vhost_net.o
> > >  vhost_net-y := net.o
> > >  
> > > +obj-$(CONFIG_VHOST_RPMSG) += vhost_rpmsg.o
> > > +vhost_rpmsg-y := rpmsg.o
> > > +
> > >  obj-$(CONFIG_VHOST_SCSI) += vhost_scsi.o
> > >  vhost_scsi-y := scsi.o
> > >  
> > > diff --git a/drivers/vhost/rpmsg.c b/drivers/vhost/rpmsg.c
> > > new file mode 100644
> > > index 000000000000..0ddee5b5f017
> > > --- /dev/null
> > > +++ b/drivers/vhost/rpmsg.c
> > > @@ -0,0 +1,370 @@
> > > +// SPDX-License-Identifier: GPL-2.0-only
> > > +/*
> > > + * Copyright(c) 2020 Intel Corporation. All rights reserved.
> > > + *
> > > + * Author: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > > + *
> > > + * Vhost RPMsg VirtIO interface provides a set of functions to be used on the
> > > + * host side as a counterpart to the guest side RPMsg VirtIO API, provided by
> > > + * drivers/rpmsg/virtio_rpmsg_bus.c. This API can be used by any vhost driver to
> > > + * handle RPMsg specific virtqueue processing.
> > > + * Vhost drivers, using this API will use their own VirtIO device IDs, that
> > > + * should then also be added to the ID table in virtio_rpmsg_bus.c
> > > + */
> > > +
> > > +#include <linux/compat.h>
> > > +#include <linux/file.h>
> > > +#include <linux/miscdevice.h>
> > 
> > As far as I can tell the above two are not needed.
> 
> Look like left-over, will remove.
> 
> > > +#include <linux/module.h>
> > > +#include <linux/mutex.h>
> > > +#include <linux/rpmsg/virtio.h>
> > > +#include <linux/vhost.h>
> > > +#include <uapi/linux/rpmsg.h>
> > > +
> > > +#include "vhost.h"
> > > +#include "vhost_rpmsg.h"
> > > +
> > > +/*
> > > + * All virtio-rpmsg virtual queue kicks always come with just one buffer -
> > > + * either input or output, but we can also handle split messages
> > > + */
> > > +static int vhost_rpmsg_get_msg(struct vhost_virtqueue *vq, unsigned int *cnt)
> > > +{
> > > +	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
> > > +	unsigned int out, in;
> > > +	int head = vhost_get_vq_desc(vq, vq->iov, ARRAY_SIZE(vq->iov), &out, &in,
> > > +				     NULL, NULL);
> > > +	if (head < 0) {
> > > +		vq_err(vq, "%s(): error %d getting buffer\n",
> > > +		       __func__, head);
> > > +		return head;
> > > +	}
> > > +
> > > +	/* Nothing new? */
> > > +	if (head == vq->num)
> > > +		return head;
> > > +
> > > +	if (vq == &vr->vq[VIRTIO_RPMSG_RESPONSE]) {
> > > +		if (out) {
> > > +			vq_err(vq, "%s(): invalid %d output in response queue\n",
> > > +			       __func__, out);
> > > +			goto return_buf;
> > > +		}
> > > +
> > > +		*cnt = in;
> > > +	}
> > > +
> > > +	if (vq == &vr->vq[VIRTIO_RPMSG_REQUEST]) {
> > > +		if (in) {
> > > +			vq_err(vq, "%s(): invalid %d input in request queue\n",
> > > +		       __func__, in);
> > > +			goto return_buf;
> > > +		}
> > > +
> > > +		*cnt = out;
> > > +	}
> > > +
> > > +	return head;
> > > +
> > > +return_buf:
> > > +	vhost_add_used(vq, head, 0);
> > > +
> > > +	return -EINVAL;
> > > +}
> > > +
> > > +static const struct vhost_rpmsg_ept *vhost_rpmsg_ept_find(struct vhost_rpmsg *vr, int addr)
> > > +{
> > > +	unsigned int i;
> > > +
> > > +	for (i = 0; i < vr->n_epts; i++)
> > > +		if (vr->ept[i].addr == addr)
> > > +			return vr->ept + i;
> > > +
> > > +	return NULL;
> > > +}
> > > +
> > > +/*
> > > + * if len < 0, then for reading a request, the complete virtual queue buffer
> > > + * size is prepared, for sending a response, the length in the iterator is used
> > > + */
> > > +int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> > > +			   unsigned int qid, ssize_t len)
> > > +	__acquires(vq->mutex)
> > > +{
> > > +	struct vhost_virtqueue *vq = vr->vq + qid;
> > > +	unsigned int cnt;
> > > +	ssize_t ret;
> > > +	size_t tmp;
> > > +
> > > +	if (qid >= VIRTIO_RPMSG_NUM_OF_VQS)
> > > +		return -EINVAL;
> > > +
> > > +	iter->vq = vq;
> > > +
> > > +	mutex_lock(&vq->mutex);
> > > +	vhost_disable_notify(&vr->dev, vq);
> > > +
> > > +	iter->head = vhost_rpmsg_get_msg(vq, &cnt);
> > > +	if (iter->head == vq->num)
> > > +		iter->head = -EAGAIN;
> > > +
> > > +	if (iter->head < 0) {
> > > +		ret = iter->head;
> > > +		goto unlock;
> > > +	}
> > > +
> > > +	tmp = iov_length(vq->iov, cnt);
> > > +	if (tmp < sizeof(iter->rhdr)) {
> > > +		vq_err(vq, "%s(): size %zu too small\n", __func__, tmp);
> > > +		ret = -ENOBUFS;
> > > +		goto return_buf;
> > > +	}
> > > +
> > > +	switch (qid) {
> > > +	case VIRTIO_RPMSG_REQUEST:
> > > +		if (len >= 0) {
> > > +			if (tmp < sizeof(iter->rhdr) + len) {
> > > +				ret = -ENOBUFS;
> > > +				goto return_buf;
> > > +			}
> > > +
> > > +			tmp = len + sizeof(iter->rhdr);
> > > +		}
> > > +
> > > +		/* len is now the size of the payload */
> > > +		iov_iter_init(&iter->iov_iter, WRITE, vq->iov, cnt, tmp);
> > > +
> > > +		/* Read the RPMSG header with endpoint addresses */
> > > +		tmp = copy_from_iter(&iter->rhdr, sizeof(iter->rhdr), &iter->iov_iter);
> > > +		if (tmp != sizeof(iter->rhdr)) {
> > > +			vq_err(vq, "%s(): got %zu instead of %zu\n", __func__,
> > > +			       tmp, sizeof(iter->rhdr));
> > > +			ret = -EIO;
> > > +			goto return_buf;
> > > +		}
> > > +
> > > +		iter->ept = vhost_rpmsg_ept_find(vr, vhost32_to_cpu(vq, iter->rhdr.dst));
> > > +		if (!iter->ept) {
> > > +			vq_err(vq, "%s(): no endpoint with address %d\n",
> > > +			       __func__, vhost32_to_cpu(vq, iter->rhdr.dst));
> > > +			ret = -ENOENT;
> > > +			goto return_buf;
> > > +		}
> > > +
> > > +		/* Let the endpoint read the payload */
> > > +		if (iter->ept->read) {
> > > +			ret = iter->ept->read(vr, iter);
> > > +			if (ret < 0)
> > > +				goto return_buf;
> > > +
> > > +			iter->rhdr.len = cpu_to_vhost16(vq, ret);
> > > +		} else {
> > > +			iter->rhdr.len = 0;
> > > +		}
> > > +
> > > +		/* Prepare for the response phase */
> > > +		iter->rhdr.dst = iter->rhdr.src;
> > > +		iter->rhdr.src = cpu_to_vhost32(vq, iter->ept->addr);
> > 
> > I'm a little puzzled here - what will the response look like?  And why is it
> > prepared here?  From what I can see doing so introduces coupling with function
> > handle_rpmsg_req_single().  I think confirmation of reception should be handled
> > by endpoints rather than in the core. 
> 
> RPMsg always contain a header, so we keep the header in the iterator. If the 
> caller wants to reply to the request, the easiest way to do that is to reuse the 
> iterator. In that case obviously you have to swap source and destination 
> addresses. This can be done either in the request handler of the API, or by the 
> caller, or in the API response handler. It would be silly to have the user do 
> that, that would be repeated code. But I agree, it's a bit unclean to modify the 
> header before returning it to the user, without knowing, whether the user will 
> use it, in which case it might be surprised to see most fields from the request 
> unchanged and only addresses swapped. I'll move this to response with a check 
> for a reused iterator.
> 
> > > +
> > > +		break;
> > > +	case VIRTIO_RPMSG_RESPONSE:
> > > +		if (!iter->ept && iter->rhdr.dst != cpu_to_vhost32(vq, RPMSG_NS_ADDR)) {
> > > +			/*
> > > +			 * Usually the iterator is configured when processing a
> > > +			 * message on the request queue, but it's also possible
> > > +			 * to send a message on the response queue without a
> > > +			 * preceding request, in that case the iterator must
> > > +			 * contain source and destination addresses.
> > > +			 */
> > > +			iter->ept = vhost_rpmsg_ept_find(vr, vhost32_to_cpu(vq, iter->rhdr.src));
> > > +			if (!iter->ept) {
> > > +				ret = -ENOENT;
> > > +				goto return_buf;
> > > +			}
> > > +		}
> > > +
> > > +		if (len >= 0) {
> > > +			if (tmp < sizeof(iter->rhdr) + len) {
> > > +				ret = -ENOBUFS;
> > > +				goto return_buf;
> > > +			}
> > > +
> > > +			iter->rhdr.len = cpu_to_vhost16(vq, len);
> > > +			tmp = len + sizeof(iter->rhdr);
> > > +		}
> > > +
> > > +		/* len is now the size of the payload */
> > > +		iov_iter_init(&iter->iov_iter, READ, vq->iov, cnt, tmp);
> > > +
> > > +		/* Write the RPMSG header with endpoint addresses */
> > > +		tmp = copy_to_iter(&iter->rhdr, sizeof(iter->rhdr), &iter->iov_iter);
> > > +		if (tmp != sizeof(iter->rhdr)) {
> > > +			ret = -EIO;
> > > +			goto return_buf;
> > > +		}
> > > +
> > > +		/* Let the endpoint write the payload */
> > 
> > I would specifically mention that namespace payloads are taken care of by
> > vhost_rpmsg_ns_announce().  That makes it easier for people to connect the dots. 
> 
> Ok
> 
> > > +		if (iter->ept && iter->ept->write) {
> > > +			ret = iter->ept->write(vr, iter);
> > > +			if (ret < 0)
> > > +				goto return_buf;
> > > +		}
> > > +
> > > +		break;
> > > +	}
> > > +
> > > +	return 0;
> > > +
> > > +return_buf:
> > > +	vhost_add_used(vq, iter->head, 0);
> > > +unlock:
> > > +	vhost_enable_notify(&vr->dev, vq);
> > > +	mutex_unlock(&vq->mutex);
> > > +
> > > +	return ret;
> > > +}
> > > +EXPORT_SYMBOL_GPL(vhost_rpmsg_start_lock);
> > > +
> > > +size_t vhost_rpmsg_copy(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> > > +			void *data, size_t size)
> > > +{
> > > +	/*
> > > +	 * We could check for excess data, but copy_{to,from}_iter() don't do
> > > +	 * that either
> > > +	 */
> > > +	if (iter->vq == vr->vq + VIRTIO_RPMSG_RESPONSE)
> > > +		return copy_to_iter(data, size, &iter->iov_iter);
> > > +
> > > +	return copy_from_iter(data, size, &iter->iov_iter);
> > > +}
> > > +EXPORT_SYMBOL_GPL(vhost_rpmsg_copy);
> > > +
> > > +int vhost_rpmsg_finish_unlock(struct vhost_rpmsg *vr,
> > > +			      struct vhost_rpmsg_iter *iter)
> > > +	__releases(vq->mutex)
> > > +{
> > > +	if (iter->head >= 0)
> > > +		vhost_add_used_and_signal(iter->vq->dev, iter->vq, iter->head,
> > > +					  vhost16_to_cpu(iter->vq, iter->rhdr.len) +
> > > +					  sizeof(iter->rhdr));
> > > +
> > > +	vhost_enable_notify(&vr->dev, iter->vq);
> > > +	mutex_unlock(&iter->vq->mutex);
> > > +
> > > +	return iter->head;
> > > +}
> > > +EXPORT_SYMBOL_GPL(vhost_rpmsg_finish_unlock);
> > > +
> > > +/*
> > > + * Return false to terminate the external loop only if we fail to obtain either
> > > + * a request or a response buffer
> > > + */
> > > +static bool handle_rpmsg_req_single(struct vhost_rpmsg *vr,
> > > +				    struct vhost_virtqueue *vq)
> > > +{
> > > +	struct vhost_rpmsg_iter iter;
> > > +	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_REQUEST, -EINVAL);
> > > +	if (!ret)
> > > +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > > +	if (ret < 0) {
> > > +		if (ret != -EAGAIN)
> > > +			vq_err(vq, "%s(): RPMSG processing failed %d\n",
> > > +			       __func__, ret);
> > > +		return false;
> > > +	}
> > > +
> > > +	if (!iter.ept->write)
> > > +		return true;
> > > +
> > > +	ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, -EINVAL);
> > > +	if (!ret)
> > > +		ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > > +	if (ret < 0) {
> > > +		vq_err(vq, "%s(): RPMSG finalising failed %d\n", __func__, ret);
> > > +		return false;
> > > +	}
> > 
> > As I said before dealing with the "response" queue here seems to be introducing
> > coupling with vhost_rpmsg_start_lock()...  Endpoints should be doing that.
> 
> Sorry, could you elaborate a bit, what do you mean by coupling?
> 
> > > +
> > > +	return true;
> > > +}
> > > +
> > > +static void handle_rpmsg_req_kick(struct vhost_work *work)
> > > +{
> > > +	struct vhost_virtqueue *vq = container_of(work, struct vhost_virtqueue,
> > > +						  poll.work);
> > > +	struct vhost_rpmsg *vr = container_of(vq->dev, struct vhost_rpmsg, dev);
> > > +
> > > +	while (handle_rpmsg_req_single(vr, vq))
> > > +		;
> > > +}
> > > +
> > > +/*
> > > + * initialise two virtqueues with an array of endpoints,
> > > + * request and response callbacks
> > > + */
> > > +void vhost_rpmsg_init(struct vhost_rpmsg *vr, const struct vhost_rpmsg_ept *ept,
> > > +		      unsigned int n_epts)
> > > +{
> > > +	unsigned int i;
> > > +
> > > +	for (i = 0; i < ARRAY_SIZE(vr->vq); i++)
> > > +		vr->vq_p[i] = &vr->vq[i];
> > > +
> > > +	/* vq[0]: host -> guest, vq[1]: host <- guest */
> > > +	vr->vq[VIRTIO_RPMSG_REQUEST].handle_kick = handle_rpmsg_req_kick;
> > > +	vr->vq[VIRTIO_RPMSG_RESPONSE].handle_kick = NULL;
> > 
> > The comment depicts vq[0] followed by vq[1] but the code initialise vq[1] before
> > vq[0], which is wildly confusing.  At the very least this should be: 
> 
> Nobody should care which of those is 0 and which is 1 :-) But indeed you have a point, 
> that the protocol isn't strictly request-response based, the host can also send 
> messages to the guest without preceding requests. So, TX / RX should be a better fit.
> 
> > 
> > 	vr->vq[VIRTIO_RPMSG_RESPONSE].handle_kick = NULL;
> > 	vr->vq[VIRTIO_RPMSG_REQUEST].handle_kick = handle_rpmsg_req_kick;
> > 
> > And even better:
> > 
> >         /* See configuration of *vq_cbs[] in rpmsg_probe()  */
> > 	vr->vq[VIRTIO_RPMSG_TX].handle_kick = NULL;
> > 	vr->vq[VIRTIO_RPMSG_RX].handle_kick = handle_rpmsg_req_kick;
> > 
> > > +
> > > +	vr->ept = ept;
> > > +	vr->n_epts = n_epts;
> > > +
> > > +	vhost_dev_init(&vr->dev, vr->vq_p, VIRTIO_RPMSG_NUM_OF_VQS,
> > > +		       UIO_MAXIOV, 0, 0, true, NULL);
> > > +}
> > > +EXPORT_SYMBOL_GPL(vhost_rpmsg_init);
> > > +
> > > +void vhost_rpmsg_destroy(struct vhost_rpmsg *vr)
> > > +{
> > > +	if (vhost_dev_has_owner(&vr->dev))
> > > +		vhost_poll_flush(&vr->vq[VIRTIO_RPMSG_REQUEST].poll);
> > > +
> > > +	vhost_dev_cleanup(&vr->dev);
> > > +}
> > > +EXPORT_SYMBOL_GPL(vhost_rpmsg_destroy);
> > > +
> > > +/* send namespace */
> > > +int vhost_rpmsg_ns_announce(struct vhost_rpmsg *vr, const char *name, unsigned int src)
> > > +{
> > > +	struct vhost_virtqueue *vq = &vr->vq[VIRTIO_RPMSG_RESPONSE];
> > > +	struct vhost_rpmsg_iter iter = {
> > > +		.rhdr = {
> > > +			.src = 0,
> > > +			.dst = cpu_to_vhost32(vq, RPMSG_NS_ADDR),
> > > +		},
> > > +	};
> > > +	struct rpmsg_ns_msg ns = {
> > > +		.addr = cpu_to_vhost32(vq, src),
> > > +		.flags = cpu_to_vhost32(vq, RPMSG_NS_CREATE), /* for rpmsg_ns_cb() */
> > > +	};
> > 
> > Here we have to assume the source can be found in the endpoints registered in
> > vhost_rpmsg_init().  I would put a check to make sure that is the case and
> > return an error otherwise. 
> 
> Ok, will do.
> 
> > > +	int ret = vhost_rpmsg_start_lock(vr, &iter, VIRTIO_RPMSG_RESPONSE, sizeof(ns));
> > > +
> > > +	if (ret < 0)
> > > +		return ret;
> > > +
> > > +	strlcpy(ns.name, name, sizeof(ns.name));
> > > +
> > > +	ret = vhost_rpmsg_copy(vr, &iter, &ns, sizeof(ns));
> > > +	if (ret != sizeof(ns))
> > > +		vq_err(iter.vq, "%s(): added %d instead of %zu bytes\n",
> > > +		       __func__, ret, sizeof(ns));
> > > +
> > > +	ret = vhost_rpmsg_finish_unlock(vr, &iter);
> > > +	if (ret < 0)
> > > +		vq_err(iter.vq, "%s(): namespace announcement failed: %d\n",
> > > +		       __func__, ret);
> > > +
> > > +	return ret;
> > > +}
> > > +EXPORT_SYMBOL_GPL(vhost_rpmsg_ns_announce);
> > > +
> > > +MODULE_LICENSE("GPL v2");
> > > +MODULE_AUTHOR("Intel, Inc.");
> > > +MODULE_DESCRIPTION("Vhost RPMsg API");
> > > diff --git a/drivers/vhost/vhost_rpmsg.h b/drivers/vhost/vhost_rpmsg.h
> > > new file mode 100644
> > > index 000000000000..c020ea14cd16
> > > --- /dev/null
> > > +++ b/drivers/vhost/vhost_rpmsg.h
> > > @@ -0,0 +1,74 @@
> > > +/* SPDX-License-Identifier: GPL-2.0 */
> > > +/*
> > > + * Copyright(c) 2020 Intel Corporation. All rights reserved.
> > > + *
> > > + * Author: Guennadi Liakhovetski <guennadi.liakhovetski@linux.intel.com>
> > > + */
> > > +
> > > +#ifndef VHOST_RPMSG_H
> > > +#define VHOST_RPMSG_H
> > > +
> > > +#include <linux/rpmsg/virtio.h>
> > > +#include <linux/uio.h>
> > > +
> > > +#include "vhost.h"
> > > +
> > > +/* RPMsg uses two VirtQueues: one for each direction */
> > > +enum {
> > > +	VIRTIO_RPMSG_RESPONSE,	/* RPMsg response (host->guest) buffers */
> > > +	VIRTIO_RPMSG_REQUEST,	/* RPMsg request (guest->host) buffers */
> > 
> > As I said above things would be much clearer if this was VIRTIO_RPMSG_TX and
> > VIRTIO_RPMSG_RX.
> 
> Ack.
> 
> > I won't be commenting on the mechanic needed to access and send information on
> > the virtqueues as it is completely foreign to me.  Other than the above I think
> > this is going somewhere. 
> 
> I'll wait for your clarifications about "coupling" and send a v8.
>

Please hold off on sending another revision.  We need to find a solution to keep
rpmsg_hdr and rpmsg_ns_msg generic (see my reply to Arnaud).

Mathieu
 
> Thanks for the comments so far
> Guennadi
> 
> > Thanks,
> > Mathieu
> > 
> > > +	/* Keep last */
> > > +	VIRTIO_RPMSG_NUM_OF_VQS,
> > > +};
> > > +
> > > +struct vhost_rpmsg_ept;
> > > +
> > > +struct vhost_rpmsg_iter {
> > > +	struct iov_iter iov_iter;
> > > +	struct rpmsg_hdr rhdr;
> > > +	struct vhost_virtqueue *vq;
> > > +	const struct vhost_rpmsg_ept *ept;
> > > +	int head;
> > > +	void *priv;
> > > +};
> > > +
> > > +struct vhost_rpmsg {
> > > +	struct vhost_dev dev;
> > > +	struct vhost_virtqueue vq[VIRTIO_RPMSG_NUM_OF_VQS];
> > > +	struct vhost_virtqueue *vq_p[VIRTIO_RPMSG_NUM_OF_VQS];
> > > +	const struct vhost_rpmsg_ept *ept;
> > > +	unsigned int n_epts;
> > > +};
> > > +
> > > +struct vhost_rpmsg_ept {
> > > +	ssize_t (*read)(struct vhost_rpmsg *, struct vhost_rpmsg_iter *);
> > > +	ssize_t (*write)(struct vhost_rpmsg *, struct vhost_rpmsg_iter *);
> > > +	int addr;
> > > +};
> > > +
> > > +static inline size_t vhost_rpmsg_iter_len(const struct vhost_rpmsg_iter *iter)
> > > +{
> > > +	return iter->rhdr.len;
> > > +}
> > > +
> > > +#define VHOST_RPMSG_ITER(_vq, _src, _dst) {			\
> > > +	.rhdr = {						\
> > > +			.src = cpu_to_vhost32(_vq, _src),	\
> > > +			.dst = cpu_to_vhost32(_vq, _dst),	\
> > > +		},						\
> > > +	}
> > > +
> > > +void vhost_rpmsg_init(struct vhost_rpmsg *vr, const struct vhost_rpmsg_ept *ept,
> > > +		      unsigned int n_epts);
> > > +void vhost_rpmsg_destroy(struct vhost_rpmsg *vr);
> > > +int vhost_rpmsg_ns_announce(struct vhost_rpmsg *vr, const char *name,
> > > +			    unsigned int src);
> > > +int vhost_rpmsg_start_lock(struct vhost_rpmsg *vr,
> > > +			   struct vhost_rpmsg_iter *iter,
> > > +			   unsigned int qid, ssize_t len);
> > > +size_t vhost_rpmsg_copy(struct vhost_rpmsg *vr, struct vhost_rpmsg_iter *iter,
> > > +			void *data, size_t size);
> > > +int vhost_rpmsg_finish_unlock(struct vhost_rpmsg *vr,
> > > +			      struct vhost_rpmsg_iter *iter);
> > > +
> > > +#endif
> > > -- 
> > > 2.28.0
> > >