Re: [Xen-devel] [PATCH 15/25] argo: implement the sendv op

[Christopher Clark <christopher.w.clark@xxxxxxxxx>]

On Wed, Dec 12, 2018 at 3:53 AM Jan Beulich <JBeulich@xxxxxxxx> wrote:
>
> >>> On 01.12.18 at 02:32, <christopher.w.clark@xxxxxxxxx> wrote:
> > +static void
> > +argo_signal_domain(struct domain *d)
> > +{
> > +    argo_dprintk("signalling domid:%d\n", d->domain_id);
> > +
> > +    if ( !d->argo ) /* This can happen if the domain is being destroyed */
> > +        return;
>
> If such a precaution is necessary, how is it guaranteed that
> the pointer doesn't change to NULL between the check above
> and ...
>
> > +    evtchn_send(d, d->argo->evtchn_port);
>
> ... the use here?

ack, this code is gone in v2.
d->argo is safe to access when holding either read or write
of L1, the global argo lock, so won't switch to NULL as a surprise.

> > +static int
> > +argo_iov_count(XEN_GUEST_HANDLE_PARAM(argo_iov_t) iovs, uint8_t niov,
> > +               uint32_t *count)
> > +{
> > +    argo_iov_t iov;
> > +    uint32_t sum_iov_lens = 0;
> > +    int ret;
> > +
> > +    if ( niov > ARGO_MAXIOV )
> > +        return -EINVAL;
> > +
> > +    while ( niov-- )
> > +    {
> > +        ret = copy_from_guest_errno(&iov, iovs, 1);
> > +        if ( ret )
> > +            return ret;
> > +
> > +        /* check each to protect sum against integer overflow */
> > +        if ( iov.iov_len > ARGO_MAX_RING_SIZE )
> > +            return -EINVAL;
> > +
> > +        sum_iov_lens += iov.iov_len;
> > +
> > +        /*
> > +         * Again protect sum from integer overflow
> > +         * and ensure total msg size will be within bounds.
> > +         */
> > +        if ( sum_iov_lens > ARGO_MAX_MSG_SIZE )
> > +            return -EINVAL;
>
> So you do overflow checks here. But how does this help when ...
>
> > +        guest_handle_add_offset(iovs, 1);
> > +    }
> > +
> > +    *count = sum_iov_lens;
> > +    return 0;
> > +}
> > +
> > +static int
> > +argo_ringbuf_insert(struct domain *d,
> > +                    struct argo_ring_info *ring_info,
> > +                    const struct argo_ring_id *src_id,
> > +                    XEN_GUEST_HANDLE_PARAM(argo_iov_t) iovs, uint8_t niov,
> > +                    uint32_t message_type, unsigned long *out_len)
> > +{
> > +    argo_ring_t ring;
> > +    struct argo_ring_message_header mh = { 0 };
> > +    int32_t sp;
> > +    int32_t ret = 0;
> > +    uint32_t len;
> > +    uint32_t iov_len;
> > +    uint32_t sum_iov_len = 0;
> > +
> > +    ASSERT(spin_is_locked(&ring_info->lock));
> > +
> > +    if ( (ret = argo_iov_count(iovs, niov, &len)) )
> > +        return ret;
> > +
> > +    if ( ((ARGO_ROUNDUP(len) + sizeof (struct argo_ring_message_header) ) 
> > >=
> > +          ring_info->len)
> > +         || (len > ARGO_MAX_MSG_SIZE) )
> > +        return -EMSGSIZE;
> > +
> > +    do {
> > +        ret =  argo_ringbuf_get_rx_ptr(ring_info, &ring.rx_ptr);
> > +        if ( ret )
> > +            break;
> > +
> > +        argo_sanitize_ring(&ring, ring_info);
> > +
> > +        argo_dprintk("ring.tx_ptr=%d ring.rx_ptr=%d ring.len=%d"
> > +                     " ring_info->tx_ptr=%d\n",
> > +                     ring.tx_ptr, ring.rx_ptr, ring.len, 
> > ring_info->tx_ptr);
> > +
> > +        if ( ring.rx_ptr == ring.tx_ptr )
> > +            sp = ring_info->len;
> > +        else
> > +        {
> > +            sp = ring.rx_ptr - ring.tx_ptr;
> > +            if ( sp < 0 )
> > +                sp += ring.len;
> > +        }
> > +
> > +        if ( (ARGO_ROUNDUP(len) + sizeof(struct argo_ring_message_header)) 
> > >= sp )
> > +        {
> > +            argo_dprintk("EAGAIN\n");
> > +            ret = -EAGAIN;
> > +            break;
> > +        }
> > +
> > +        mh.len = len + sizeof(struct argo_ring_message_header);
> > +        mh.source.port = src_id->addr.port;
> > +        mh.source.domain_id = src_id->addr.domain_id;
> > +        mh.message_type = message_type;
> > +
> > +        /*
> > +         * For this copy to the guest ring, tx_ptr is always 16-byte 
> > aligned
> > +         * and the message header is 16 bytes long.
> > +         */
> > +        BUILD_BUG_ON(sizeof(struct argo_ring_message_header) != 
> > ARGO_ROUNDUP(1));
> > +
> > +        if ( (ret = argo_memcpy_to_guest_ring(ring_info,
> > +                                              ring.tx_ptr + 
> > sizeof(argo_ring_t),
> > +                                              &mh,
> > +                                              
> > XEN_GUEST_HANDLE_NULL(uint8_t),
> > +                                              sizeof(mh))) )
> > +            break;
> > +
> > +        ring.tx_ptr += sizeof(mh);
> > +        if ( ring.tx_ptr == ring_info->len )
> > +            ring.tx_ptr = 0;
> > +
> > +        while ( niov-- )
> > +        {
> > +            XEN_GUEST_HANDLE_PARAM(uint8_t) bufp_hnd;
> > +            XEN_GUEST_HANDLE(uint8_t) buf_hnd;
> > +            argo_iov_t iov;
> > +
> > +            ret = copy_from_guest_errno(&iov, iovs, 1);
>
> ... here you copy the structure again from guest memory, at
> which point it may have changed? I see you do some checks
> further down, but the question then is - is the checking in
> argo_iov_count() redundant and hence unnecessary? Are
> you really safe here against inconsistencies between the
> first and second reads? If so, a thorough explanation in a
> comment is needed here.

Fair point and comments have been added to v2.

>
> > +            if ( ret )
> > +                break;
> > +
> > +            bufp_hnd = guest_handle_from_ptr((uintptr_t)iov.iov_base, 
> > uint8_t);
>
> Please use a handle in the public interface instead of such a
> cast.

ack.

> > +            sp = ring.len - ring.tx_ptr;
> > +
> > +            if ( iov_len > sp )
> > +            {
> > +                ret = argo_memcpy_to_guest_ring(ring_info,
> > +                        ring.tx_ptr + sizeof(argo_ring_t),
> > +                        NULL, buf_hnd, sp);
> > +                if ( ret )
> > +                    break;
> > +
> > +                ring.tx_ptr = 0;
> > +                iov_len -= sp;
> > +                guest_handle_add_offset(buf_hnd, sp);
> > +            }
> > +
> > +            ret = argo_memcpy_to_guest_ring(ring_info,
> > +                        ring.tx_ptr + sizeof(argo_ring_t),
> > +                        NULL, buf_hnd, iov_len);
>
> Extending the remark on double guest memory read above, is
> it certain you won't overrun the ring here?

Yes, certain it's ok. Comments added to explain.

>
> > +            if ( ret )
> > +                break;
> > +
> > +            ring.tx_ptr += iov_len;
> > +
> > +            if ( ring.tx_ptr == ring_info->len )
> > +                ring.tx_ptr = 0;
> > +
> > +            guest_handle_add_offset(iovs, 1);
> > +        }
> > +
> > +        if ( ret )
> > +            break;
> > +
> > +        ring.tx_ptr = ARGO_ROUNDUP(ring.tx_ptr);
> > +
> > +        if ( ring.tx_ptr >= ring_info->len )
> > +            ring.tx_ptr -= ring_info->len;
> > +
> > +        mb();
> > +        ring_info->tx_ptr = ring.tx_ptr;
>
> What does the above barrier guard against? It's all hypervisor
> local memory which gets altered afaict.

ack, dropped.

>
> > +static int
> > +argo_pending_requeue(struct argo_ring_info *ring_info, domid_t src_id, int 
> > len)
> > +{
> > +    struct hlist_node *node;
> > +    struct argo_pending_ent *ent;
> > +
> > +    ASSERT(spin_is_locked(&ring_info->lock));
> > +
> > +    hlist_for_each_entry(ent, node, &ring_info->pending, node)
> > +    {
> > +        if ( ent->id == src_id )
> > +        {
> > +            if ( ent->len < len )
> > +                ent->len = len;
>
> What does this achieve? I.e. why is this not either a plain
> assignment or a check that the length is the same?

New comment added:

/*
 * Reuse an existing queue entry for a notification rather than add
 * another. If the existing entry is waiting for a smaller size than
 * the current message then adjust the record to wait for the
 * current (larger) size to be available before triggering a
 * notification.
 * This assists the waiting sender by ensuring that whenever a
 * notification is triggered, there is sufficient space available
 * for (at least) any one of the messages awaiting transmission.
 */

> > +static struct argo_ring_info *
> > +argo_ring_find_info_by_match(const struct domain *d, uint32_t port,
> > +                             domid_t partner_id, uint64_t partner_cookie)
> > +{
> > +    argo_ring_id_t id;
> > +    struct argo_ring_info *ring_info;
> > +
> > +    ASSERT(rw_is_locked(&d->argo->lock));
> > +
> > +    id.addr.port = port;
> > +    id.addr.domain_id = d->domain_id;
> > +    id.partner = partner_id;
> > +
> > +    ring_info = argo_ring_find_info(d, &id);
> > +    if ( ring_info && (partner_cookie == ring_info->partner_cookie) )
> > +        return ring_info;
>
> Such a cookie makes mismatches unlikely, but it doesn't exclude
> them. If there are other checks, is the cookie useful at all?

Yes, I think so and it's proved useful elsewhere in the second
version of the series: it helps avoid sending signals to incorrect
domains that may not be argo-enabled.

> > @@ -813,6 +1318,29 @@ do_argo_message_op(int cmd, 
> > XEN_GUEST_HANDLE_PARAM(void) arg1,
> >          rc = argo_unregister_ring(d, ring_hnd);
> >          break;
> >      }
> > +    case ARGO_MESSAGE_OP_sendv:
> > +    {
> > +        argo_send_addr_t send_addr;
> > +        uint32_t niov = arg3;
> > +        uint32_t message_type = arg4;
>
> At the example of these (perhaps I've again overlooked earlier
> instances), what about the upper halves on 64-bit? Given the
> rather generic interface of the actual hypercall, I don't think it
> is a good idea to ignore the bits. The situation is different for
> the "cmd" parameter, which is uniformly 32-bit for all sub-ops.

ack.

> Talking of "cmd" and its type: In case it wasn't said by anyone
> else yet, please use unsigned types wherever negative values
> are impossible.
>
> > +        XEN_GUEST_HANDLE_PARAM(argo_send_addr_t) send_addr_hnd =
> > +            guest_handle_cast(arg1, argo_send_addr_t);
> > +        XEN_GUEST_HANDLE_PARAM(argo_iov_t) iovs =
> > +            guest_handle_cast(arg2, argo_iov_t);
> > +
> > +        if ( unlikely(!guest_handle_okay(send_addr_hnd, 1)) )
> > +            break;
> > +        rc = copy_from_guest_errno(&send_addr, send_addr_hnd, 1);
> > +        if ( rc )
> > +            break;
> > +
> > +        send_addr.src.domain_id = d->domain_id;
>
> What use is the field if you override it like this?

ack, have switched to a correct match check in v2.

> I don't think I've found any checking of this field to be zero, to
> allow for future re-use.

ack.

Christopher

_______________________________________________
Xen-devel mailing list
Xen-devel@xxxxxxxxxxxxxxxxxxxx
https://lists.xenproject.org/mailman/listinfo/xen-devel