MINOR: ring: count processed messages in ring_dispatch_messages()

ring_dispatch_messages() now takes an optional argument <processed> which
must point to a size_t counter when provided.

When provided, the value is updated to the number of messages processed
by the function.

diff --git a/include/haproxy/ring.h b/include/haproxy/ring.h
index 201ede465872e00e1f483299c3da5fbad0a22d1f..3eb965a283ed9cb7956b4b7fc3f78c4758dab8cd 100644 (file)
--- a/include/haproxy/ring.h
+++ b/include/haproxy/ring.h
@@ -43,7 +43,8 @@ void cli_io_release_show_ring(struct appctx *appctx);
 
 size_t ring_max_payload(const struct ring *ring);
 int ring_dispatch_messages(struct ring *ring, void *ctx, size_t *ofs_ptr, size_t *last_ofs_ptr, uint flags,
-                          ssize_t (*msg_handler)(void *ctx, struct ist v1, struct ist v2, size_t ofs, size_t len));
+                           ssize_t (*msg_handler)(void *ctx, struct ist v1, struct ist v2, size_t ofs, size_t len),
+                           size_t *processed);
 
 /* returns the ring storage's usable area */
 static inline void *ring_area(const struct ring *ring)

diff --git a/src/ring.c b/src/ring.c
index 1c32bd8eb7d4070d0e02609c3c39832177533b00..5b5833365b3e4e6d1a6e72f09f1d5a3e807ba7bd 100644 (file)
--- a/src/ring.c
+++ b/src/ring.c
@@ -545,9 +545,13 @@ int ring_attach_cli(struct ring *ring, struct appctx *appctx, uint flags)
  * the last known tail pointer will be copied there so that the caller may use
  * this to detect new data have arrived since we left the function. Returns 0
  * if it needs to pause, 1 once finished.
+ *
+ * If <processed> is not NULL, it will be set to the number of messages
+ * processed by the function (even when the function returns 0)
  */
 int ring_dispatch_messages(struct ring *ring, void *ctx, size_t *ofs_ptr, size_t *last_ofs_ptr, uint flags,
-                          ssize_t (*msg_handler)(void *ctx, struct ist v1, struct ist v2, size_t ofs, size_t len))
+                           ssize_t (*msg_handler)(void *ctx, struct ist v1, struct ist v2, size_t ofs, size_t len),
+                           size_t *processed)
 {
        size_t head_ofs, tail_ofs, prev_ofs;
        size_t ring_size;
@@ -555,6 +559,7 @@ int ring_dispatch_messages(struct ring *ring, void *ctx, size_t *ofs_ptr, size_t
        struct ist v1, v2;
        uint64_t msg_len;
        size_t len, cnt;
+       size_t msg_count = 0;
        ssize_t copied;
        uint8_t readers;
        int ret;
@@ -650,6 +655,7 @@ int ring_dispatch_messages(struct ring *ring, void *ctx, size_t *ofs_ptr, size_t
                        break;
                }
        skip:
+               msg_count += 1;
                vp_skip(&v1, &v2, cnt + msg_len);
        }
 
@@ -672,6 +678,8 @@ int ring_dispatch_messages(struct ring *ring, void *ctx, size_t *ofs_ptr, size_t
        if (last_ofs_ptr)
                *last_ofs_ptr = tail_ofs;
        *ofs_ptr = head_ofs;
+       if (processed)
+               *processed = msg_count;
        return ret;
 }
 
@@ -695,7 +703,7 @@ int cli_io_handler_show_ring(struct appctx *appctx)
 
        MT_LIST_DELETE(&appctx->wait_entry);
 
-       ret = ring_dispatch_messages(ring, appctx, &ctx->ofs, &last_ofs, ctx->flags, applet_append_line);
+       ret = ring_dispatch_messages(ring, appctx, &ctx->ofs, &last_ofs, ctx->flags, applet_append_line, NULL);
 
        if (ret && (ctx->flags & RING_WF_WAIT_MODE)) {
                /* we've drained everything and are configured to wait for more

diff --git a/src/sink.c b/src/sink.c
index 8cc8acf242ddae115c08cf7cd50c40dbfc904ccf..29b68a0189244d11681e53556986bb76b13a4998 100644 (file)
--- a/src/sink.c
+++ b/src/sink.c
@@ -382,7 +382,8 @@ static void _sink_forward_io_handler(struct appctx *appctx,
 
        MT_LIST_DELETE(&appctx->wait_entry);
 
-       ret = ring_dispatch_messages(ring, appctx, &sft->ofs, &last_ofs, 0, msg_handler);
+       ret = ring_dispatch_messages(ring, appctx, &sft->ofs, &last_ofs, 0,
+                                    msg_handler, NULL);
 
        if (ret) {
                /* let's be woken up once new data arrive */