[RFC,1/1] ring: add infrastructure to allow callbacks within the ring library

diff --git a/lib/ring/rte_ring.h b/lib/ring/rte_ring.h
index 7e4cd60650..9cda49ee51 100644
--- a/lib/ring/rte_ring.h
+++ b/lib/ring/rte_ring.h
@@ -789,6 +789,126 @@  rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table,
 			n, available);
 }
 
+#ifdef RTE_RING_DEQUEUE_CALLBACKS
+/**
+ * Function type used for ring dequeue callbacks.
+ *
+ * The callback function is called on dequeue with a burst of objects that have
+ * been dequeued on the given ring.
+ *
+ * @param r
+ *   A pointer to the ring structure.
+ * @param obj_table
+ *   A pointer to a table of void * pointers (objects) filled by the dequeue.
+ * @param n
+ *   The number of objects to dequeued from the ring to the obj_table.
+ * @param available
+ *   If non-NULL, returns the number of remaining ring entries after the
+ *   dequeue has finished.
+ * @param user_param
+ *   The arbitrary user parameter passed in by the application when the callback
+ *   was originally configured.
+ * @return
+ *   The number of packets returned to the user.
+ */
+typedef uint16_t (*rte_ring_dequeue_callback_fn)(struct rte_ring *r,
+		void **obj_table, unsigned int n, unsigned int *available,
+		void *user_param);
+
+/**
+ * @internal
+ * Structure used to hold information about the callbacks to be called for a
+ * ring on dequeue.
+ */
+struct rte_ring_dequeue_callback {
+	struct rte_ring_dequeue_callback *next;
+	rte_ring_dequeue_callback_fn fn;
+	void *param;
+};
+
+/**
+ * Add a callback to be called on dequeue of a given ring.
+ *
+ * This API configures a function to be called for each burst of
+ * objects dequeued from a given RING. The return value is a pointer
+ * that can be used to later remove the callback using
+ * rte_ring_remove_dequeue_callback().
+ *
+ * Multiple functions are called in the order that they are added.
+ *
+ * @param r
+ *   A pointer to the ring structure.
+ * @param fn
+ *   The callback function.
+ * @param user_param
+ *   A generic pointer parameter which will be passed to each invocation of the
+ *   callback function on this ring.
+ *
+ * @return
+ *   NULL on error.
+ *   On success, a pointer value which can later be used to remove the callback.
+ */
+const struct rte_ring_dequeue_callback *
+rte_ring_add_dequeue_callback(struct rte_ring *r,
+		rte_ring_dequeue_callback_fn fn, void *user_param);
+
+/**
+ * Remove a callback from a given ring.
+ *
+ * This function is used to removed callbacks that were added to a ring
+ * dequeue using rte_ring_add_dequeue_callback().
+ *
+ * Note: the callback is removed from the callback list but it isn't freed
+ * since the it may still be in use. The memory for the callback can be
+ * subsequently freed back by the application by calling rte_free():
+ *
+ * - Immediately - if the user knows that no callbacks are in flight
+ *   e.g. if called from the thread doing the dequeue on that ring.
+ *
+ * - After a short delay - where the delay is sufficient to allow any
+ *   in-flight callbacks to complete. Alternately, the RCU mechanism can be
+ *   used to detect when data plane threads have ceased referencing the
+ *   callback memory.
+ *
+ * @param r
+ *   A pointer to the ring structure.
+ * @param user_cb
+ *   User supplied callback created via rte_ring_add_dequeue_callback().
+ *
+ * @return
+ *   - 0: Success. Callback was removed.
+ *   - -ENODEV:  If *ring* is invalid.
+ *   - -ENOTSUP: Callback support is not available.
+ *   - -EINVAL:  The callback is NULL or not found for the ring.
+ */
+int rte_ring_remove_dequeue_callback(struct rte_ring *r,
+		const struct rte_ring_dequeue_callback *user_cb);
+
+/**
+ * @internal
+ * Helper routine for rte_ring_dequeue_burst().
+ * Invokes dequeue callbacks if any, etc.
+ *
+ * @param r
+ *   A pointer to the ring structure.
+ * @param obj_table
+ *   A pointer to a table of void * pointers (objects) filled by the dequeue.
+ * @param n
+ *   The number of objects to dequeued from the ring to the obj_table.
+ * @param available
+ *   If non-NULL, returns the number of remaining ring entries after the
+ *   dequeue has finished.
+ * @param opaque
+ *   Opaque pointer of ring dequeue callback related data.
+ *
+ * @return
+ *  The number of objects effectively supplied to the @p obj_table array.
+ */
+uint16_t rte_ring_call_dequeue_callbacks(struct rte_ring *r,
+		void **obj_table, unsigned int n, unsigned int *available,
+		void *opaque);
+#endif
+
 /**
  * Dequeue multiple objects from a ring up to a maximum number.
  *
@@ -812,8 +932,19 @@  static __rte_always_inline unsigned int
 rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table,
 		unsigned int n, unsigned int *available)
 {
-	return rte_ring_dequeue_burst_elem(r, obj_table, sizeof(void *),
+	unsigned int nb_rx;
+	nb_rx = rte_ring_dequeue_burst_elem(r, obj_table, sizeof(void *),
 			n, available);
+
+#ifdef RTE_RING_DEQUEUE_CALLBACKS
+	{
+		if (unlikely(r->cb != NULL))
+			nb_rx = rte_ring_call_dequeue_callbacks(r, obj_table,
+					nb_rx, available, r->cb);
+	}
+#endif
+
+	return nb_rx;
 }
 
 #ifdef __cplusplus
diff --git a/lib/ring/rte_ring_core.h b/lib/ring/rte_ring_core.h
index 82b237091b..603e1b45f0 100644
--- a/lib/ring/rte_ring_core.h
+++ b/lib/ring/rte_ring_core.h
@@ -147,6 +147,9 @@  struct rte_ring {
 		struct rte_ring_rts_headtail rts_cons;
 	}  __rte_cache_aligned;
 
+	/** points to array of ring callback data pointers */
+	void **cb;
+
 	char pad2 __rte_cache_aligned; /**< empty cache line */
 };