From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: (majordomo@vger.kernel.org) by vger.kernel.org via listexpand id S1755221AbdDRDHw (ORCPT ); Mon, 17 Apr 2017 23:07:52 -0400 Received: from mx1.redhat.com ([209.132.183.28]:39448 "EHLO mx1.redhat.com" rhost-flags-OK-OK-OK-OK) by vger.kernel.org with ESMTP id S1752136AbdDRDHu (ORCPT ); Mon, 17 Apr 2017 23:07:50 -0400 DMARC-Filter: OpenDMARC Filter v1.3.2 mx1.redhat.com 565E861D38 Authentication-Results: ext-mx10.extmail.prod.ext.phx2.redhat.com; dmarc=none (p=none dis=none) header.from=redhat.com Authentication-Results: ext-mx10.extmail.prod.ext.phx2.redhat.com; spf=pass smtp.mailfrom=jasowang@redhat.com DKIM-Filter: OpenDKIM Filter v2.11.0 mx1.redhat.com 565E861D38 Subject: Re: [PATCH RFC] ptr_ring: add ptr_ring_unconsume To: "Michael S. Tsirkin" , linux-kernel@vger.kernel.org References: <1492384496-22644-1-git-send-email-mst@redhat.com> Cc: netdev@vger.kernel.org From: Jason Wang Message-ID: <0defb746-3d4b-14b3-1ad7-82842048ebba@redhat.com> Date: Tue, 18 Apr 2017 11:07:42 +0800 User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:45.0) Gecko/20100101 Thunderbird/45.8.0 MIME-Version: 1.0 In-Reply-To: <1492384496-22644-1-git-send-email-mst@redhat.com> Content-Type: text/plain; charset=UTF-8; format=flowed Content-Transfer-Encoding: 8bit X-Greylist: Sender IP whitelisted, not delayed by milter-greylist-4.5.16 (mx1.redhat.com [10.5.110.39]); Tue, 18 Apr 2017 03:07:49 +0000 (UTC) Sender: linux-kernel-owner@vger.kernel.org List-ID: X-Mailing-List: linux-kernel@vger.kernel.org On 2017年04月17日 07:19, Michael S. Tsirkin wrote: > Applications that consume a batch of entries in one go > can benefit from ability to return some of them back > into the ring. > > Add an API for that - assuming there's space. If there's no space > naturally we can't do this and have to drop entries, but this implies > ring is full so we'd likely drop some anyway. > > Signed-off-by: Michael S. Tsirkin > --- > > Jason, in my mind the biggest issue with your batching patchset is the > backet drops on disconnect. This API will help avoid that in the common > case. Ok, I will rebase the series on top of this. (Though I don't think we care the packet loss). > > I would still prefer that we understand what's going on, I try to reply in another thread, does it make sense? > and I would > like to know what's the smallest batch size that's still helpful, Yes, I've replied in another thread, the result is: no batching 1.88Mpps RX_BATCH=1 1.93Mpps RX_BATCH=4 2.11Mpps RX_BATCH=16 2.14Mpps RX_BATCH=64 2.25Mpps RX_BATCH=256 2.18Mpps > but > I'm not going to block the patch on these grounds assuming packet drops > are fixed. Thanks a lot. > > Lightly tested - this is on top of consumer batching patches. > > Thanks! > > include/linux/ptr_ring.h | 57 ++++++++++++++++++++++++++++++++++++++++++++++++ > 1 file changed, 57 insertions(+) > > diff --git a/include/linux/ptr_ring.h b/include/linux/ptr_ring.h > index 783e7f5..5fbeab4 100644 > --- a/include/linux/ptr_ring.h > +++ b/include/linux/ptr_ring.h > @@ -457,6 +457,63 @@ static inline int ptr_ring_init(struct ptr_ring *r, int size, gfp_t gfp) > return 0; > } > > +/* > + * Return entries into ring. Destroy entries that don't fit. > + * > + * Note: this is expected to be a rare slow path operation. > + * > + * Note: producer lock is nested within consumer lock, so if you > + * resize you must make sure all uses nest correctly. > + * In particular if you consume ring in interrupt or BH context, you must > + * disable interrupts/BH when doing so. > + */ > +static inline void ptr_ring_unconsume(struct ptr_ring *r, void **batch, int n, > + void (*destroy)(void *)) > +{ > + unsigned long flags; > + int head; > + > + spin_lock_irqsave(&(r)->consumer_lock, flags); > + spin_lock(&(r)->producer_lock); > + > + if (!r->size) > + goto done; > + > + /* > + * Clean out buffered entries (for simplicity). This way following code > + * can test entries for NULL and if not assume they are valid. > + */ > + head = r->consumer_head - 1; > + while (likely(head >= r->consumer_tail)) > + r->queue[head--] = NULL; > + r->consumer_tail = r->consumer_head; > + > + /* > + * Go over entries in batch, start moving head back and copy entries. > + * Stop when we run into previously unconsumed entries. > + */ > + while (n--) { > + head = r->consumer_head - 1; > + if (head < 0) > + head = r->size - 1; > + if (r->queue[head]) { > + /* This batch entry will have to be destroyed. */ > + ++n; > + goto done; > + } > + r->queue[head] = batch[n]; > + r->consumer_tail = r->consumer_head = head; > + } > + > +done: > + /* Destroy all entries left in the batch. */ > + while (n--) { > + destroy(batch[n]); > + } > + spin_unlock(&(r)->producer_lock); > + spin_unlock_irqrestore(&(r)->consumer_lock, flags); > +} > + > static inline void **__ptr_ring_swap_queue(struct ptr_ring *r, void **queue, > int size, gfp_t gfp, > void (*destroy)(void *))