server/red_channel: move out_bytes_counter from Outgoing to RedChannel

3 files changed, 17 insertions, 7 deletions

diff --git a/server/red_channel.c b/server/red_channel.c
index 783fa487..25854981 100644
--- a/server/red_channel.c
+++ b/server/red_channel.c
@@ -186,7 +186,7 @@ static void red_peer_handle_outgoing(RedsStream *stream, OutgoingHandler *handle
             }
         } else {
             handler->pos += n;
-            stat_inc_counter(handler->out_bytes_counter, n);
+            handler->cb->on_output(handler->opaque, n);
             if (handler->pos == handler->size) { // finished writing data
                 handler->cb->on_msg_done(handler->opaque);
                 handler->vec = handler->vec_buf;
@@ -198,6 +198,13 @@ static void red_peer_handle_outgoing(RedsStream *stream, OutgoingHandler *handle
     }
 }
 
+void red_channel_on_output(void *opaque, int n)
+{
+    RedChannel *channel = opaque;
+
+    stat_inc_counter(channel->out_bytes_counter, n);
+}
+
 void red_channel_default_peer_on_error(RedChannel *channel)
 {
     channel->disconnect(channel);
@@ -359,18 +366,19 @@ RedChannel *red_channel_create(int size, RedsStream *stream,
     channel->outgoing.opaque = channel;
     channel->outgoing.pos = 0;
     channel->outgoing.size = 0;
-    channel->outgoing.out_bytes_counter = NULL;
 
     channel->outgoing_cb.get_msg_size = red_channel_peer_get_out_msg_size;
     channel->outgoing_cb.prepare = red_channel_peer_prepare_out_msg;
     channel->outgoing_cb.on_block = red_channel_peer_on_out_block;
     channel->outgoing_cb.on_error = (on_outgoing_error_proc)red_channel_default_peer_on_error;
     channel->outgoing_cb.on_msg_done = red_channel_peer_on_out_msg_done;
+    channel->outgoing_cb.on_output = red_channel_on_output;
 
     channel->incoming.cb = &channel->incoming_cb;
     channel->outgoing.cb = &channel->outgoing_cb;
 
     channel->shut = 0; // came here from inputs, perhaps can be removed? XXX
+    channel->out_bytes_counter = 0;
 
     if (!config_socket(channel)) {
         goto error;
diff --git a/server/red_channel.h b/server/red_channel.h
index 7ec07347..97da5a82 100644
--- a/server/red_channel.h
+++ b/server/red_channel.h
@@ -69,6 +69,7 @@ typedef void (*prepare_outgoing_proc)(void *opaque, struct iovec *vec, int *vec_
 typedef void (*on_outgoing_error_proc)(void *opaque);
 typedef void (*on_outgoing_block_proc)(void *opaque);
 typedef void (*on_outgoing_msg_done_proc)(void *opaque);
+typedef void (*on_output_proc)(void *opaque, int n);
 
 typedef struct OutgoingHandlerInterface {
     get_outgoing_msg_size_proc get_msg_size;
@@ -76,6 +77,7 @@ typedef struct OutgoingHandlerInterface {
     on_outgoing_error_proc on_error;
     on_outgoing_block_proc on_block;
     on_outgoing_msg_done_proc on_msg_done;
+    on_output_proc on_output;
 } OutgoingHandlerInterface;
 
 typedef struct OutgoingHandler {
@@ -86,9 +88,6 @@ typedef struct OutgoingHandler {
     struct iovec *vec;
     int pos;
     int size;
-#ifdef RED_STATISTICS
-    uint64_t *out_bytes_counter;
-#endif
 } OutgoingHandler;
 
 /* Red Channel interface */
@@ -184,6 +183,9 @@ struct RedChannel {
     channel_handle_migrate_flush_mark handle_migrate_flush_mark;
     channel_handle_migrate_data handle_migrate_data;
     channel_handle_migrate_data_get_serial handle_migrate_data_get_serial;
+#ifdef RED_STATISTICS
+    uint64_t *out_bytes_counter;
+#endif
 };
 
 /* if one of the callbacks should cause disconnect, use red_channel_shutdown and don't
diff --git a/server/red_worker.c b/server/red_worker.c
index 446c72e0..c6a49ce5 100644
--- a/server/red_worker.c
+++ b/server/red_worker.c
@@ -9163,7 +9163,7 @@ static void handle_new_display_channel(RedWorker *worker, RedsStream *stream, in
     }
 #ifdef RED_STATISTICS
     display_channel->stat = stat_add_node(worker->stat, "display_channel", TRUE);
-    display_channel->common.base.outgoing.out_bytes_counter = stat_add_counter(display_channel->stat,
+    display_channel->common.base.out_bytes_counter = stat_add_counter(display_channel->stat,
                                                                "out_bytes", TRUE);
     display_channel->cache_hits_counter = stat_add_counter(display_channel->stat,
                                                            "cache_hits", TRUE);
@@ -9298,7 +9298,7 @@ static void red_connect_cursor(RedWorker *worker, RedsStream *stream, int migrat
     }
 #ifdef RED_STATISTICS
     channel->stat = stat_add_node(worker->stat, "cursor_channel", TRUE);
-    channel->common.base.outgoing.out_bytes_counter = stat_add_counter(channel->stat, "out_bytes", TRUE);
+    channel->common.base.out_bytes_counter = stat_add_counter(channel->stat, "out_bytes", TRUE);
 #endif
     ring_init(&channel->cursor_cache_lru);
     channel->cursor_cache_available = CLIENT_CURSOR_CACHE_SIZE;