Re: [Xen-devel] [PATCH v4 16/18] xen/pvcalls: implement read

2017-06-21 Thread Stefano Stabellini 
On Wed, 21 Jun 2017, Boris Ostrovsky wrote:
> On 06/15/2017 03:09 PM, Stefano Stabellini wrote:
> > When an active socket has data available, increment the io and read
> > counters, and schedule the ioworker.
> >
> > Implement the read function by reading from the socket, writing the data
> > to the data ring.
> >
> > Set in_error on error.
> >
> > Signed-off-by: Stefano Stabellini 
> > CC: boris.ostrov...@oracle.com
> > CC: jgr...@suse.com
> > ---
> >  drivers/xen/pvcalls-back.c | 85 
> > ++
> >  1 file changed, 85 insertions(+)
> >
> > diff --git a/drivers/xen/pvcalls-back.c b/drivers/xen/pvcalls-back.c
> > index b9a10b9..65d9eba 100644
> > --- a/drivers/xen/pvcalls-back.c
> > +++ b/drivers/xen/pvcalls-back.c
> > @@ -100,6 +100,81 @@ static int pvcalls_back_release_active(struct 
> > xenbus_device *dev,
> >  
> >  static void pvcalls_conn_back_read(void *opaque)
> >  {
> > +   struct sock_mapping *map = (struct sock_mapping *)opaque;
> > +   struct msghdr msg;
> > +   struct kvec vec[2];
> > +   RING_IDX cons, prod, size, wanted, array_size, masked_prod, masked_cons;
> > +   int32_t error;
> > +   struct pvcalls_data_intf *intf = map->ring;
> > +   struct pvcalls_data *data = >data;
> > +   unsigned long flags;
> > +   int ret;
> > +
> > +   array_size = XEN_FLEX_RING_SIZE(map->ring_order);
> 
> I noticed that in the next patch you call this 'ring_size. Can you make
> those things consistent? (There may be more than just this variable and,
> in fact, perhaps some things can be factored out? There are code
> fragments that look similar)

Yes, I'll make them more consistent. I don't think we can actually share
code between the two functions are they do different things.


> > +   cons = intf->in_cons;
> > +   prod = intf->in_prod;
> > +   error = intf->in_error;
> > +   /* read the indexes first, then deal with the data */
> > +   virt_mb();
> > +
> > +   if (error)
> > +   return;
> > +
> > +   size = pvcalls_queued(prod, cons, array_size);
> > +   if (size >= array_size)
> > +   return;
> > +   spin_lock_irqsave(>sock->sk->sk_receive_queue.lock, flags);
> > +   if (skb_queue_empty(>sock->sk->sk_receive_queue)) {
> > +   atomic_set(>read, 0);
> > +   spin_unlock_irqrestore(>sock->sk->sk_receive_queue.lock,
> > +   flags);
> > +   return;
> > +   }
> > +   spin_unlock_irqrestore(>sock->sk->sk_receive_queue.lock, flags);
> > +   wanted = array_size - size;
> > +   masked_prod = pvcalls_mask(prod, array_size);
> > +   masked_cons = pvcalls_mask(cons, array_size);
> > +
> > +   memset(, 0, sizeof(msg));
> > +   msg.msg_iter.type = ITER_KVEC|WRITE;
> > +   msg.msg_iter.count = wanted;
> > +   if (masked_prod < masked_cons) {
> > +   vec[0].iov_base = data->in + masked_prod;
> > +   vec[0].iov_len = wanted;
> > +   msg.msg_iter.kvec = vec;
> > +   msg.msg_iter.nr_segs = 1;
> > +   } else {
> > +   vec[0].iov_base = data->in + masked_prod;
> > +   vec[0].iov_len = array_size - masked_prod;
> > +   vec[1].iov_base = data->in;
> > +   vec[1].iov_len = wanted - vec[0].iov_len;
> > +   msg.msg_iter.kvec = vec;
> > +   msg.msg_iter.nr_segs = 2;
> > +   }
> 
> 
> This is probably obvious to everyone but me but can you explain what is
> going on here? ;-)

We are setting up iovecs based on the "in" array (similarly the write
function does the same for the "out" array). Then we are passing the
iovecs to inet_recvmsg to do IO. Depending on the indexes on the array
we need one iovec entry or two, in case we need to wrap around the
circular buffer.

 
> > +
> > +   atomic_set(>read, 0);
> 
> Is this not atomic_dec() by any chance?

It is meant to be atomic_set: the idea is that we are going to drain all
the data. If there is any remaming data after inet_recvmsg, we'll
increase map->read again.


> > +   ret = inet_recvmsg(map->sock, , wanted, MSG_DONTWAIT);
> > +   WARN_ON(ret > wanted);
> > +   if (ret == -EAGAIN) /* shouldn't happen */
> > +   return;
> > +   if (!ret)
> > +   ret = -ENOTCONN;
> > +   spin_lock_irqsave(>sock->sk->sk_receive_queue.lock, flags);
> > +   if (ret > 0 && !skb_queue_empty(>sock->sk->sk_receive_queue))
> > +   atomic_inc(>read);
> > +   spin_unlock_irqrestore(>sock->sk->sk_receive_queue.lock, flags);
> > +
> > +   /* write the data, then modify the indexes */
> > +   virt_wmb();
> > +   if (ret < 0)
> > +   intf->in_error = ret;
> > +   else
> > +   intf->in_prod = prod + ret;
> > +   /* update the indexes, then notify the other end */
> > +   virt_wmb();
> > +   notify_remote_via_irq(map->irq);
> > +
> > +   return;
> >  }
> >  
> >  static int pvcalls_conn_back_write(struct sock_mapping *map)
> > @@ -172,6 +247,16 @@ static void pvcalls_sk_state_change(struct sock *sock)
> >  
> >  static void pvcalls_sk_data_ready(struct sock *sock)
> >  {
> > +

Re: [Xen-devel] [PATCH v4 16/18] xen/pvcalls: implement read

2017-06-21 Thread Boris Ostrovsky 
On 06/15/2017 03:09 PM, Stefano Stabellini wrote:
> When an active socket has data available, increment the io and read
> counters, and schedule the ioworker.
>
> Implement the read function by reading from the socket, writing the data
> to the data ring.
>
> Set in_error on error.
>
> Signed-off-by: Stefano Stabellini 
> CC: boris.ostrov...@oracle.com
> CC: jgr...@suse.com
> ---
>  drivers/xen/pvcalls-back.c | 85 
> ++
>  1 file changed, 85 insertions(+)
>
> diff --git a/drivers/xen/pvcalls-back.c b/drivers/xen/pvcalls-back.c
> index b9a10b9..65d9eba 100644
> --- a/drivers/xen/pvcalls-back.c
> +++ b/drivers/xen/pvcalls-back.c
> @@ -100,6 +100,81 @@ static int pvcalls_back_release_active(struct 
> xenbus_device *dev,
>  
>  static void pvcalls_conn_back_read(void *opaque)
>  {
> + struct sock_mapping *map = (struct sock_mapping *)opaque;
> + struct msghdr msg;
> + struct kvec vec[2];
> + RING_IDX cons, prod, size, wanted, array_size, masked_prod, masked_cons;
> + int32_t error;
> + struct pvcalls_data_intf *intf = map->ring;
> + struct pvcalls_data *data = >data;
> + unsigned long flags;
> + int ret;
> +
> + array_size = XEN_FLEX_RING_SIZE(map->ring_order);

I noticed that in the next patch you call this 'ring_size. Can you make
those things consistent? (There may be more than just this variable and,
in fact, perhaps some things can be factored out? There are code
fragments that look similar)

> + cons = intf->in_cons;
> + prod = intf->in_prod;
> + error = intf->in_error;
> + /* read the indexes first, then deal with the data */
> + virt_mb();
> +
> + if (error)
> + return;
> +
> + size = pvcalls_queued(prod, cons, array_size);
> + if (size >= array_size)
> + return;
> + spin_lock_irqsave(>sock->sk->sk_receive_queue.lock, flags);
> + if (skb_queue_empty(>sock->sk->sk_receive_queue)) {
> + atomic_set(>read, 0);
> + spin_unlock_irqrestore(>sock->sk->sk_receive_queue.lock,
> + flags);
> + return;
> + }
> + spin_unlock_irqrestore(>sock->sk->sk_receive_queue.lock, flags);
> + wanted = array_size - size;
> + masked_prod = pvcalls_mask(prod, array_size);
> + masked_cons = pvcalls_mask(cons, array_size);
> +
> + memset(, 0, sizeof(msg));
> + msg.msg_iter.type = ITER_KVEC|WRITE;
> + msg.msg_iter.count = wanted;
> + if (masked_prod < masked_cons) {
> + vec[0].iov_base = data->in + masked_prod;
> + vec[0].iov_len = wanted;
> + msg.msg_iter.kvec = vec;
> + msg.msg_iter.nr_segs = 1;
> + } else {
> + vec[0].iov_base = data->in + masked_prod;
> + vec[0].iov_len = array_size - masked_prod;
> + vec[1].iov_base = data->in;
> + vec[1].iov_len = wanted - vec[0].iov_len;
> + msg.msg_iter.kvec = vec;
> + msg.msg_iter.nr_segs = 2;
> + }


This is probably obvious to everyone but me but can you explain what is
going on here? ;-)


> +
> + atomic_set(>read, 0);

Is this not atomic_dec() by any chance?


-boris

> + ret = inet_recvmsg(map->sock, , wanted, MSG_DONTWAIT);
> + WARN_ON(ret > wanted);
> + if (ret == -EAGAIN) /* shouldn't happen */
> + return;
> + if (!ret)
> + ret = -ENOTCONN;
> + spin_lock_irqsave(>sock->sk->sk_receive_queue.lock, flags);
> + if (ret > 0 && !skb_queue_empty(>sock->sk->sk_receive_queue))
> + atomic_inc(>read);
> + spin_unlock_irqrestore(>sock->sk->sk_receive_queue.lock, flags);
> +
> + /* write the data, then modify the indexes */
> + virt_wmb();
> + if (ret < 0)
> + intf->in_error = ret;
> + else
> + intf->in_prod = prod + ret;
> + /* update the indexes, then notify the other end */
> + virt_wmb();
> + notify_remote_via_irq(map->irq);
> +
> + return;
>  }
>  
>  static int pvcalls_conn_back_write(struct sock_mapping *map)
> @@ -172,6 +247,16 @@ static void pvcalls_sk_state_change(struct sock *sock)
>  
>  static void pvcalls_sk_data_ready(struct sock *sock)
>  {
> + struct sock_mapping *map = sock->sk_user_data;
> + struct pvcalls_ioworker *iow;
> +
> + if (map == NULL)
> + return;
> +
> + iow = >ioworker;
> + atomic_inc(>read);
> + atomic_inc(>io);
> + queue_work(iow->wq, >register_work);
>  }
>  
>  static struct sock_mapping *pvcalls_new_active_socket(


___
Xen-devel mailing list
Xen-devel@lists.xen.org
https://lists.xen.org/xen-devel

[Xen-devel] [PATCH v4 16/18] xen/pvcalls: implement read

2017-06-15 Thread Stefano Stabellini 
When an active socket has data available, increment the io and read
counters, and schedule the ioworker.

Implement the read function by reading from the socket, writing the data
to the data ring.

Set in_error on error.

Signed-off-by: Stefano Stabellini 
CC: boris.ostrov...@oracle.com
CC: jgr...@suse.com
---
 drivers/xen/pvcalls-back.c | 85 ++
 1 file changed, 85 insertions(+)

diff --git a/drivers/xen/pvcalls-back.c b/drivers/xen/pvcalls-back.c
index b9a10b9..65d9eba 100644
--- a/drivers/xen/pvcalls-back.c
+++ b/drivers/xen/pvcalls-back.c
@@ -100,6 +100,81 @@ static int pvcalls_back_release_active(struct 
xenbus_device *dev,
 
 static void pvcalls_conn_back_read(void *opaque)
 {
+   struct sock_mapping *map = (struct sock_mapping *)opaque;
+   struct msghdr msg;
+   struct kvec vec[2];
+   RING_IDX cons, prod, size, wanted, array_size, masked_prod, masked_cons;
+   int32_t error;
+   struct pvcalls_data_intf *intf = map->ring;
+   struct pvcalls_data *data = >data;
+   unsigned long flags;
+   int ret;
+
+   array_size = XEN_FLEX_RING_SIZE(map->ring_order);
+   cons = intf->in_cons;
+   prod = intf->in_prod;
+   error = intf->in_error;
+   /* read the indexes first, then deal with the data */
+   virt_mb();
+
+   if (error)
+   return;
+
+   size = pvcalls_queued(prod, cons, array_size);
+   if (size >= array_size)
+   return;
+   spin_lock_irqsave(>sock->sk->sk_receive_queue.lock, flags);
+   if (skb_queue_empty(>sock->sk->sk_receive_queue)) {
+   atomic_set(>read, 0);
+   spin_unlock_irqrestore(>sock->sk->sk_receive_queue.lock,
+   flags);
+   return;
+   }
+   spin_unlock_irqrestore(>sock->sk->sk_receive_queue.lock, flags);
+   wanted = array_size - size;
+   masked_prod = pvcalls_mask(prod, array_size);
+   masked_cons = pvcalls_mask(cons, array_size);
+
+   memset(, 0, sizeof(msg));
+   msg.msg_iter.type = ITER_KVEC|WRITE;
+   msg.msg_iter.count = wanted;
+   if (masked_prod < masked_cons) {
+   vec[0].iov_base = data->in + masked_prod;
+   vec[0].iov_len = wanted;
+   msg.msg_iter.kvec = vec;
+   msg.msg_iter.nr_segs = 1;
+   } else {
+   vec[0].iov_base = data->in + masked_prod;
+   vec[0].iov_len = array_size - masked_prod;
+   vec[1].iov_base = data->in;
+   vec[1].iov_len = wanted - vec[0].iov_len;
+   msg.msg_iter.kvec = vec;
+   msg.msg_iter.nr_segs = 2;
+   }
+
+   atomic_set(>read, 0);
+   ret = inet_recvmsg(map->sock, , wanted, MSG_DONTWAIT);
+   WARN_ON(ret > wanted);
+   if (ret == -EAGAIN) /* shouldn't happen */
+   return;
+   if (!ret)
+   ret = -ENOTCONN;
+   spin_lock_irqsave(>sock->sk->sk_receive_queue.lock, flags);
+   if (ret > 0 && !skb_queue_empty(>sock->sk->sk_receive_queue))
+   atomic_inc(>read);
+   spin_unlock_irqrestore(>sock->sk->sk_receive_queue.lock, flags);
+
+   /* write the data, then modify the indexes */
+   virt_wmb();
+   if (ret < 0)
+   intf->in_error = ret;
+   else
+   intf->in_prod = prod + ret;
+   /* update the indexes, then notify the other end */
+   virt_wmb();
+   notify_remote_via_irq(map->irq);
+
+   return;
 }
 
 static int pvcalls_conn_back_write(struct sock_mapping *map)
@@ -172,6 +247,16 @@ static void pvcalls_sk_state_change(struct sock *sock)
 
 static void pvcalls_sk_data_ready(struct sock *sock)
 {
+   struct sock_mapping *map = sock->sk_user_data;
+   struct pvcalls_ioworker *iow;
+
+   if (map == NULL)
+   return;
+
+   iow = >ioworker;
+   atomic_inc(>read);
+   atomic_inc(>io);
+   queue_work(iow->wq, >register_work);
 }
 
 static struct sock_mapping *pvcalls_new_active_socket(
-- 
1.9.1


___
Xen-devel mailing list
Xen-devel@lists.xen.org
https://lists.xen.org/xen-devel