4 files changed, 758 insertions, 1 deletions

diff --git a/include/linux/ptr_ring.h b/include/linux/ptr_ring.h
new file mode 100644
index 000000000000..562a65e8bcc0
--- /dev/null
+++ b/include/linux/ptr_ring.h
@@ -0,0 +1,393 @@
+/*
+ *      Definitions for the 'struct ptr_ring' datastructure.
+ *
+ *      Author:
+ *              Michael S. Tsirkin <mst@redhat.com>
+ *
+ *      Copyright (C) 2016 Red Hat, Inc.
+ *
+ *      This program is free software; you can redistribute it and/or modify it
+ *      under the terms of the GNU General Public License as published by the
+ *      Free Software Foundation; either version 2 of the License, or (at your
+ *      option) any later version.
+ *
+ *      This is a limited-size FIFO maintaining pointers in FIFO order, with
+ *      one CPU producing entries and another consuming entries from a FIFO.
+ *
+ *      This implementation tries to minimize cache-contention when there is a
+ *      single producer and a single consumer CPU.
+ */
+
+#ifndef _LINUX_PTR_RING_H
+#define _LINUX_PTR_RING_H 1
+
+#ifdef __KERNEL__
+#include <linux/spinlock.h>
+#include <linux/cache.h>
+#include <linux/types.h>
+#include <linux/compiler.h>
+#include <linux/cache.h>
+#include <linux/slab.h>
+#include <asm/errno.h>
+#endif
+
+struct ptr_ring {
+        int producer ____cacheline_aligned_in_smp;
+        spinlock_t producer_lock;
+        int consumer ____cacheline_aligned_in_smp;
+        spinlock_t consumer_lock;
+        /* Shared consumer/producer data */
+        /* Read-only by both the producer and the consumer */
+        int size ____cacheline_aligned_in_smp; /* max entries in queue */
+        void **queue;
+};
+
+/* Note: callers invoking this in a loop must use a compiler barrier,
+ * for example cpu_relax().  If ring is ever resized, callers must hold
+ * producer_lock - see e.g. ptr_ring_full.  Otherwise, if callers don't hold
+ * producer_lock, the next call to __ptr_ring_produce may fail.
+ */
+static inline bool __ptr_ring_full(struct ptr_ring *r)
+{
+        return r->queue[r->producer];
+}
+
+static inline bool ptr_ring_full(struct ptr_ring *r)
+{
+        bool ret;
+
+        spin_lock(&r->producer_lock);
+        ret = __ptr_ring_full(r);
+        spin_unlock(&r->producer_lock);
+
+        return ret;
+}
+
+static inline bool ptr_ring_full_irq(struct ptr_ring *r)
+{
+        bool ret;
+
+        spin_lock_irq(&r->producer_lock);
+        ret = __ptr_ring_full(r);
+        spin_unlock_irq(&r->producer_lock);
+
+        return ret;
+}
+
+static inline bool ptr_ring_full_any(struct ptr_ring *r)
+{
+        unsigned long flags;
+        bool ret;
+
+        spin_lock_irqsave(&r->producer_lock, flags);
+        ret = __ptr_ring_full(r);
+        spin_unlock_irqrestore(&r->producer_lock, flags);
+
+        return ret;
+}
+
+static inline bool ptr_ring_full_bh(struct ptr_ring *r)
+{
+        bool ret;
+
+        spin_lock_bh(&r->producer_lock);
+        ret = __ptr_ring_full(r);
+        spin_unlock_bh(&r->producer_lock);
+
+        return ret;
+}
+
+/* Note: callers invoking this in a loop must use a compiler barrier,
+ * for example cpu_relax(). Callers must hold producer_lock.
+ */
+static inline int __ptr_ring_produce(struct ptr_ring *r, void *ptr)
+{
+        if (r->queue[r->producer])
+                return -ENOSPC;
+
+        r->queue[r->producer++] = ptr;
+        if (unlikely(r->producer >= r->size))
+                r->producer = 0;
+        return 0;
+}
+
+static inline int ptr_ring_produce(struct ptr_ring *r, void *ptr)
+{
+        int ret;
+
+        spin_lock(&r->producer_lock);
+        ret = __ptr_ring_produce(r, ptr);
+        spin_unlock(&r->producer_lock);
+
+        return ret;
+}
+
+static inline int ptr_ring_produce_irq(struct ptr_ring *r, void *ptr)
+{
+        int ret;
+
+        spin_lock_irq(&r->producer_lock);
+        ret = __ptr_ring_produce(r, ptr);
+        spin_unlock_irq(&r->producer_lock);
+
+        return ret;
+}
+
+static inline int ptr_ring_produce_any(struct ptr_ring *r, void *ptr)
+{
+        unsigned long flags;
+        int ret;
+
+        spin_lock_irqsave(&r->producer_lock, flags);
+        ret = __ptr_ring_produce(r, ptr);
+        spin_unlock_irqrestore(&r->producer_lock, flags);
+
+        return ret;
+}
+
+static inline int ptr_ring_produce_bh(struct ptr_ring *r, void *ptr)
+{
+        int ret;
+
+        spin_lock_bh(&r->producer_lock);
+        ret = __ptr_ring_produce(r, ptr);
+        spin_unlock_bh(&r->producer_lock);
+
+        return ret;
+}
+
+/* Note: callers invoking this in a loop must use a compiler barrier,
+ * for example cpu_relax(). Callers must take consumer_lock
+ * if they dereference the pointer - see e.g. PTR_RING_PEEK_CALL.
+ * If ring is never resized, and if the pointer is merely
+ * tested, there's no need to take the lock - see e.g.  __ptr_ring_empty.
+ */
+static inline void *__ptr_ring_peek(struct ptr_ring *r)
+{
+        return r->queue[r->consumer];
+}
+
+/* Note: callers invoking this in a loop must use a compiler barrier,
+ * for example cpu_relax(). Callers must take consumer_lock
+ * if the ring is ever resized - see e.g. ptr_ring_empty.
+ */
+static inline bool __ptr_ring_empty(struct ptr_ring *r)
+{
+        return !__ptr_ring_peek(r);
+}
+
+static inline bool ptr_ring_empty(struct ptr_ring *r)
+{
+        bool ret;
+
+        spin_lock(&r->consumer_lock);
+        ret = __ptr_ring_empty(r);
+        spin_unlock(&r->consumer_lock);
+
+        return ret;
+}
+
+static inline bool ptr_ring_empty_irq(struct ptr_ring *r)
+{
+        bool ret;
+
+        spin_lock_irq(&r->consumer_lock);
+        ret = __ptr_ring_empty(r);
+        spin_unlock_irq(&r->consumer_lock);
+
+        return ret;
+}
+
+static inline bool ptr_ring_empty_any(struct ptr_ring *r)
+{
+        unsigned long flags;
+        bool ret;
+
+        spin_lock_irqsave(&r->consumer_lock, flags);
+        ret = __ptr_ring_empty(r);
+        spin_unlock_irqrestore(&r->consumer_lock, flags);
+
+        return ret;
+}
+
+static inline bool ptr_ring_empty_bh(struct ptr_ring *r)
+{
+        bool ret;
+
+        spin_lock_bh(&r->consumer_lock);
+        ret = __ptr_ring_empty(r);
+        spin_unlock_bh(&r->consumer_lock);
+
+        return ret;
+}
+
+/* Must only be called after __ptr_ring_peek returned !NULL */
+static inline void __ptr_ring_discard_one(struct ptr_ring *r)
+{
+        r->queue[r->consumer++] = NULL;
+        if (unlikely(r->consumer >= r->size))
+                r->consumer = 0;
+}
+
+static inline void *__ptr_ring_consume(struct ptr_ring *r)
+{
+        void *ptr;
+
+        ptr = __ptr_ring_peek(r);
+        if (ptr)
+                __ptr_ring_discard_one(r);
+
+        return ptr;
+}
+
+static inline void *ptr_ring_consume(struct ptr_ring *r)
+{
+        void *ptr;
+
+        spin_lock(&r->consumer_lock);
+        ptr = __ptr_ring_consume(r);
+        spin_unlock(&r->consumer_lock);
+
+        return ptr;
+}
+
+static inline void *ptr_ring_consume_irq(struct ptr_ring *r)
+{
+        void *ptr;
+
+        spin_lock_irq(&r->consumer_lock);
+        ptr = __ptr_ring_consume(r);
+        spin_unlock_irq(&r->consumer_lock);
+
+        return ptr;
+}
+
+static inline void *ptr_ring_consume_any(struct ptr_ring *r)
+{
+        unsigned long flags;
+        void *ptr;
+
+        spin_lock_irqsave(&r->consumer_lock, flags);
+        ptr = __ptr_ring_consume(r);
+        spin_unlock_irqrestore(&r->consumer_lock, flags);
+
+        return ptr;
+}
+
+static inline void *ptr_ring_consume_bh(struct ptr_ring *r)
+{
+        void *ptr;
+
+        spin_lock_bh(&r->consumer_lock);
+        ptr = __ptr_ring_consume(r);
+        spin_unlock_bh(&r->consumer_lock);
+
+        return ptr;
+}
+
+/* Cast to structure type and call a function without discarding from FIFO.
+ * Function must return a value.
+ * Callers must take consumer_lock.
+ */
+#define __PTR_RING_PEEK_CALL(r, f) ((f)(__ptr_ring_peek(r)))
+
+#define PTR_RING_PEEK_CALL(r, f) ({ \
+        typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
+        \
+        spin_lock(&(r)->consumer_lock); \
+        __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
+        spin_unlock(&(r)->consumer_lock); \
+        __PTR_RING_PEEK_CALL_v; \
+})
+
+#define PTR_RING_PEEK_CALL_IRQ(r, f) ({ \
+        typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
+        \
+        spin_lock_irq(&(r)->consumer_lock); \
+        __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
+        spin_unlock_irq(&(r)->consumer_lock); \
+        __PTR_RING_PEEK_CALL_v; \
+})
+
+#define PTR_RING_PEEK_CALL_BH(r, f) ({ \
+        typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
+        \
+        spin_lock_bh(&(r)->consumer_lock); \
+        __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
+        spin_unlock_bh(&(r)->consumer_lock); \
+        __PTR_RING_PEEK_CALL_v; \
+})
+
+#define PTR_RING_PEEK_CALL_ANY(r, f) ({ \
+        typeof((f)(NULL)) __PTR_RING_PEEK_CALL_v; \
+        unsigned long __PTR_RING_PEEK_CALL_f;\
+        \
+        spin_lock_irqsave(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
+        __PTR_RING_PEEK_CALL_v = __PTR_RING_PEEK_CALL(r, f); \
+        spin_unlock_irqrestore(&(r)->consumer_lock, __PTR_RING_PEEK_CALL_f); \
+        __PTR_RING_PEEK_CALL_v; \
+})
+
+static inline void **__ptr_ring_init_queue_alloc(int size, gfp_t gfp)
+{
+        return kzalloc(ALIGN(size * sizeof(void *), SMP_CACHE_BYTES), gfp);
+}
+
+static inline int ptr_ring_init(struct ptr_ring *r, int size, gfp_t gfp)
+{
+        r->queue = __ptr_ring_init_queue_alloc(size, gfp);
+        if (!r->queue)
+                return -ENOMEM;
+
+        r->size = size;
+        r->producer = r->consumer = 0;
+        spin_lock_init(&r->producer_lock);
+        spin_lock_init(&r->consumer_lock);
+
+        return 0;
+}
+
+static inline int ptr_ring_resize(struct ptr_ring *r, int size, gfp_t gfp,
+                                  void (*destroy)(void *))
+{
+        unsigned long flags;
+        int producer = 0;
+        void **queue = __ptr_ring_init_queue_alloc(size, gfp);
+        void **old;
+        void *ptr;
+
+        if (!queue)
+                return -ENOMEM;
+
+        spin_lock_irqsave(&(r)->producer_lock, flags);
+
+        while ((ptr = ptr_ring_consume(r)))
+                if (producer < size)
+                        queue[producer++] = ptr;
+                else if (destroy)
+                        destroy(ptr);
+
+        r->size = size;
+        r->producer = producer;
+        r->consumer = 0;
+        old = r->queue;
+        r->queue = queue;
+
+        spin_unlock_irqrestore(&(r)->producer_lock, flags);
+
+        kfree(old);
+
+        return 0;
+}
+
+static inline void ptr_ring_cleanup(struct ptr_ring *r, void (*destroy)(void *))
+{
+        void *ptr;
+
+        if (destroy)
+                while ((ptr = ptr_ring_consume(r)))
+                        destroy(ptr);
+        kfree(r->queue);
+}
+
+#endif /* _LINUX_PTR_RING_H  */
diff --git a/include/linux/skb_array.h b/include/linux/skb_array.h
new file mode 100644
index 000000000000..678bfbf78ac4
--- /dev/null
+++ b/include/linux/skb_array.h
@@ -0,0 +1,169 @@
+/*
+ *      Definitions for the 'struct skb_array' datastructure.
+ *
+ *      Author:
+ *              Michael S. Tsirkin <mst@redhat.com>
+ *
+ *      Copyright (C) 2016 Red Hat, Inc.
+ *
+ *      This program is free software; you can redistribute it and/or modify it
+ *      under the terms of the GNU General Public License as published by the
+ *      Free Software Foundation; either version 2 of the License, or (at your
+ *      option) any later version.
+ *
+ *      Limited-size FIFO of skbs. Can be used more or less whenever
+ *      sk_buff_head can be used, except you need to know the queue size in
+ *      advance.
+ *      Implemented as a type-safe wrapper around ptr_ring.
+ */
+
+#ifndef _LINUX_SKB_ARRAY_H
+#define _LINUX_SKB_ARRAY_H 1
+
+#ifdef __KERNEL__
+#include <linux/ptr_ring.h>
+#include <linux/skbuff.h>
+#include <linux/if_vlan.h>
+#endif
+
+struct skb_array {
+        struct ptr_ring ring;
+};
+
+/* Might be slightly faster than skb_array_full below, but callers invoking
+ * this in a loop must use a compiler barrier, for example cpu_relax().
+ */
+static inline bool __skb_array_full(struct skb_array *a)
+{
+        return __ptr_ring_full(&a->ring);
+}
+
+static inline bool skb_array_full(struct skb_array *a)
+{
+        return ptr_ring_full(&a->ring);
+}
+
+static inline int skb_array_produce(struct skb_array *a, struct sk_buff *skb)
+{
+        return ptr_ring_produce(&a->ring, skb);
+}
+
+static inline int skb_array_produce_irq(struct skb_array *a, struct sk_buff *skb)
+{
+        return ptr_ring_produce_irq(&a->ring, skb);
+}
+
+static inline int skb_array_produce_bh(struct skb_array *a, struct sk_buff *skb)
+{
+        return ptr_ring_produce_bh(&a->ring, skb);
+}
+
+static inline int skb_array_produce_any(struct skb_array *a, struct sk_buff *skb)
+{
+        return ptr_ring_produce_any(&a->ring, skb);
+}
+
+/* Might be slightly faster than skb_array_empty below, but only safe if the
+ * array is never resized. Also, callers invoking this in a loop must take care
+ * to use a compiler barrier, for example cpu_relax().
+ */
+static inline bool __skb_array_empty(struct skb_array *a)
+{
+        return !__ptr_ring_peek(&a->ring);
+}
+
+static inline bool skb_array_empty(struct skb_array *a)
+{
+        return ptr_ring_empty(&a->ring);
+}
+
+static inline bool skb_array_empty_bh(struct skb_array *a)
+{
+        return ptr_ring_empty_bh(&a->ring);
+}
+
+static inline bool skb_array_empty_irq(struct skb_array *a)
+{
+        return ptr_ring_empty_irq(&a->ring);
+}
+
+static inline bool skb_array_empty_any(struct skb_array *a)
+{
+        return ptr_ring_empty_any(&a->ring);
+}
+
+static inline struct sk_buff *skb_array_consume(struct skb_array *a)
+{
+        return ptr_ring_consume(&a->ring);
+}
+
+static inline struct sk_buff *skb_array_consume_irq(struct skb_array *a)
+{
+        return ptr_ring_consume_irq(&a->ring);
+}
+
+static inline struct sk_buff *skb_array_consume_any(struct skb_array *a)
+{
+        return ptr_ring_consume_any(&a->ring);
+}
+
+static inline struct sk_buff *skb_array_consume_bh(struct skb_array *a)
+{
+        return ptr_ring_consume_bh(&a->ring);
+}
+
+static inline int __skb_array_len_with_tag(struct sk_buff *skb)
+{
+        if (likely(skb)) {
+                int len = skb->len;
+
+                if (skb_vlan_tag_present(skb))
+                        len += VLAN_HLEN;
+
+                return len;
+        } else {
+                return 0;
+        }
+}
+
+static inline int skb_array_peek_len(struct skb_array *a)
+{
+        return PTR_RING_PEEK_CALL(&a->ring, __skb_array_len_with_tag);
+}
+
+static inline int skb_array_peek_len_irq(struct skb_array *a)
+{
+        return PTR_RING_PEEK_CALL_IRQ(&a->ring, __skb_array_len_with_tag);
+}
+
+static inline int skb_array_peek_len_bh(struct skb_array *a)
+{
+        return PTR_RING_PEEK_CALL_BH(&a->ring, __skb_array_len_with_tag);
+}
+
+static inline int skb_array_peek_len_any(struct skb_array *a)
+{
+        return PTR_RING_PEEK_CALL_ANY(&a->ring, __skb_array_len_with_tag);
+}
+
+static inline int skb_array_init(struct skb_array *a, int size, gfp_t gfp)
+{
+        return ptr_ring_init(&a->ring, size, gfp);
+}
+
+void __skb_array_destroy_skb(void *ptr)
+{
+        kfree_skb(ptr);
+}
+
+int skb_array_resize(struct skb_array *a, int size, gfp_t gfp)
+{
+        return ptr_ring_resize(&a->ring, size, gfp, __skb_array_destroy_skb);
+}
+
+static inline void skb_array_cleanup(struct skb_array *a)
+{
+        ptr_ring_cleanup(&a->ring, __skb_array_destroy_skb);
+}
+
+#endif /* _LINUX_SKB_ARRAY_H  */
diff --git a/tools/virtio/ringtest/Makefile b/tools/virtio/ringtest/Makefile
index 6ba745529833..50e086c6a7b6 100644
--- a/tools/virtio/ringtest/Makefile
+++ b/tools/virtio/ringtest/Makefile
@@ -1,6 +1,6 @@
 all:
 
-all: ring virtio_ring_0_9 virtio_ring_poll virtio_ring_inorder
+all: ring virtio_ring_0_9 virtio_ring_poll virtio_ring_inorder ptr_ring
 
 CFLAGS += -Wall
 CFLAGS += -pthread -O2 -ggdb
@@ -8,6 +8,7 @@ LDFLAGS += -pthread -O2 -ggdb
 
 main.o: main.c main.h
 ring.o: ring.c main.h
+ptr_ring.o: ptr_ring.c main.h ../../../include/linux/ptr_ring.h
 virtio_ring_0_9.o: virtio_ring_0_9.c main.h
 virtio_ring_poll.o: virtio_ring_poll.c virtio_ring_0_9.c main.h
 virtio_ring_inorder.o: virtio_ring_inorder.c virtio_ring_0_9.c main.h
@@ -15,11 +16,13 @@ ring: ring.o main.o
 virtio_ring_0_9: virtio_ring_0_9.o main.o
 virtio_ring_poll: virtio_ring_poll.o main.o
 virtio_ring_inorder: virtio_ring_inorder.o main.o
+ptr_ring: ptr_ring.o main.o
 clean:
         -rm main.o
         -rm ring.o ring
         -rm virtio_ring_0_9.o virtio_ring_0_9
         -rm virtio_ring_poll.o virtio_ring_poll
         -rm virtio_ring_inorder.o virtio_ring_inorder
+        -rm ptr_ring.o ptr_ring
 
 .PHONY: all clean
diff --git a/tools/virtio/ringtest/ptr_ring.c b/tools/virtio/ringtest/ptr_ring.c
new file mode 100644
index 000000000000..74abd746ae91
--- /dev/null
+++ b/tools/virtio/ringtest/ptr_ring.c
@@ -0,0 +1,192 @@
+#define _GNU_SOURCE
+#include "main.h"
+#include <stdlib.h>
+#include <stdio.h>
+#include <string.h>
+#include <pthread.h>
+#include <malloc.h>
+#include <assert.h>
+#include <errno.h>
+#include <limits.h>
+
+#define SMP_CACHE_BYTES 64
+#define cache_line_size() SMP_CACHE_BYTES
+#define ____cacheline_aligned_in_smp __attribute__ ((aligned (SMP_CACHE_BYTES)))
+#define unlikely(x)    (__builtin_expect(!!(x), 0))
+#define ALIGN(x, a) (((x) + (a) - 1) / (a) * (a))
+typedef pthread_spinlock_t  spinlock_t;
+
+typedef int gfp_t;
+static void *kzalloc(unsigned size, gfp_t gfp)
+{
+        void *p = memalign(64, size);
+        if (!p)
+                return p;
+        memset(p, 0, size);
+
+        return p;
+}
+
+static void kfree(void *p)
+{
+        if (p)
+                free(p);
+}
+
+static void spin_lock_init(spinlock_t *lock)
+{
+        int r = pthread_spin_init(lock, 0);
+        assert(!r);
+}
+
+static void spin_lock(spinlock_t *lock)
+{
+        int ret = pthread_spin_lock(lock);
+        assert(!ret);
+}
+
+static void spin_unlock(spinlock_t *lock)
+{
+        int ret = pthread_spin_unlock(lock);
+        assert(!ret);
+}
+
+static void spin_lock_bh(spinlock_t *lock)
+{
+        spin_lock(lock);
+}
+
+static void spin_unlock_bh(spinlock_t *lock)
+{
+        spin_unlock(lock);
+}
+
+static void spin_lock_irq(spinlock_t *lock)
+{
+        spin_lock(lock);
+}
+
+static void spin_unlock_irq(spinlock_t *lock)
+{
+        spin_unlock(lock);
+}
+
+static void spin_lock_irqsave(spinlock_t *lock, unsigned long f)
+{
+        spin_lock(lock);
+}
+
+static void spin_unlock_irqrestore(spinlock_t *lock, unsigned long f)
+{
+        spin_unlock(lock);
+}
+
+#include "../../../include/linux/ptr_ring.h"
+
+static unsigned long long headcnt, tailcnt;
+static struct ptr_ring array ____cacheline_aligned_in_smp;
+
+/* implemented by ring */
+void alloc_ring(void)
+{
+        int ret = ptr_ring_init(&array, ring_size, 0);
+        assert(!ret);
+}
+
+/* guest side */
+int add_inbuf(unsigned len, void *buf, void *datap)
+{
+        int ret;
+
+        ret = __ptr_ring_produce(&array, buf);
+        if (ret >= 0) {
+                ret = 0;
+                headcnt++;
+        }
+
+        return ret;
+}
+
+/*
+ * ptr_ring API provides no way for producer to find out whether a given
+ * buffer was consumed.  Our tests merely require that a successful get_buf
+ * implies that add_inbuf succeed in the past, and that add_inbuf will succeed,
+ * fake it accordingly.
+ */
+void *get_buf(unsigned *lenp, void **bufp)
+{
+        void *datap;
+
+        if (tailcnt == headcnt || __ptr_ring_full(&array))
+                datap = NULL;
+        else {
+                datap = "Buffer\n";
+                ++tailcnt;
+        }
+
+        return datap;
+}
+
+void poll_used(void)
+{
+        void *b;
+
+        do {
+                if (tailcnt == headcnt || __ptr_ring_full(&array)) {
+                        b = NULL;
+                        barrier();
+                } else {
+                        b = "Buffer\n";
+                }
+        } while (!b);
+}
+
+void disable_call()
+{
+        assert(0);
+}
+
+bool enable_call()
+{
+        assert(0);
+}
+
+void kick_available(void)
+{
+        assert(0);
+}
+
+/* host side */
+void disable_kick()
+{
+        assert(0);
+}
+
+bool enable_kick()
+{
+        assert(0);
+}
+
+void poll_avail(void)
+{
+        void *b;
+
+        do {
+                barrier();
+                b = __ptr_ring_peek(&array);
+        } while (!b);
+}
+
+bool use_buf(unsigned *lenp, void **bufp)
+{
+        void *ptr;
+
+        ptr = __ptr_ring_consume(&array);
+
+        return ptr;
+}
+
+void call_used(void)
+{
+        assert(0);
+}