[Qemu-devel] [PATCH 9/9] add a public api to release buffer

[Yang Hongyang]

Signed-off-by: Yang Hongyang <address@hidden>
---
 include/net/filter.h |   3 ++
 net/filter-buffer.c  | 107 ++++++++++++++++++++++++++++++++++++++++++++++++++-
 2 files changed, 109 insertions(+), 1 deletion(-)

diff --git a/include/net/filter.h b/include/net/filter.h
index 44ba10f..f640cf7 100644
--- a/include/net/filter.h
+++ b/include/net/filter.h
@@ -15,4 +15,7 @@ NetClientState *filter_backend(NetClientState *nc);
 int filter_add_plugin(NetClientState *nc, NetClientState *plugin);
 int filter_del_plugin(NetClientState *nc, NetClientState *plugin);
 
+/* filter buffer plugin */
+void filter_buffer_release_all(void);
+
 #endif /* QEMU_NET_FILTER_H */
diff --git a/net/filter-buffer.c b/net/filter-buffer.c
index 15ac903..a32785d 100644
--- a/net/filter-buffer.c
+++ b/net/filter-buffer.c
@@ -12,14 +12,19 @@
 #include "qemu-common.h"
 #include "qemu/error-report.h"
 #include "net/filter.h"
+#include "qemu/main-loop.h"
 
 typedef struct FILTERBUFFERState {
     NetClientState nc;
     NetClientState *filter;
     int interval;
     NetQueue *inflight_queue;
+    QEMUBH *flush_bh;
 } FILTERBUFFERState;
 
+static void packet_send_completed(NetClientState *nc, ssize_t len);
+static void filter_buffer_flush(NetClientState *nc);
+
 static ssize_t filter_buffer_receive(NetClientState *nc, NetClientState 
*sender,
                               unsigned flags, const uint8_t *data, size_t size)
 {
@@ -27,7 +32,9 @@ static ssize_t filter_buffer_receive(NetClientState *nc, 
NetClientState *sender,
 
     if (sender->info->type == NET_CLIENT_OPTIONS_KIND_NIC) {
         /* we only buffer guest output packets */
-        qemu_net_queue_append(queue, sender, flags, data, size, NULL);
+        qemu_net_queue_append(queue, sender, flags, data, size,
+                              packet_send_completed);
+
         /* Now that we have buffered the packet, return sucess */
         return size;
     }
@@ -38,7 +45,23 @@ static ssize_t filter_buffer_receive(NetClientState *nc, 
NetClientState *sender,
 static void filter_buffer_cleanup(NetClientState *nc)
 {
     FILTERBUFFERState *s = DO_UPCAST(FILTERBUFFERState, nc, nc);
+
+    /* flush inflight packets */
+    if (s->inflight_queue) {
+        filter_buffer_flush(nc);
+    }
+
+    /* flush incoming packets */
+    s->inflight_queue = nc->incoming_queue;
+    nc->incoming_queue = NULL;
+    filter_buffer_flush(nc);
+
+    if (s->flush_bh) {
+        qemu_bh_delete(s->flush_bh);
+        s->flush_bh = NULL;
+    }
     filter_del_plugin(s->filter, nc);
+    nc->peer = NULL;
     return;
 }
 
@@ -50,6 +73,54 @@ static NetClientInfo net_filter_buffer_info = {
     .cleanup = filter_buffer_cleanup,
 };
 
+static void packet_send_completed(NetClientState *nc, ssize_t len)
+{
+    FILTERBUFFERState *s = DO_UPCAST(FILTERBUFFERState, nc, nc);
+    qemu_bh_schedule(s->flush_bh);
+}
+
+static void filter_buffer_flush(NetClientState *nc)
+{
+    FILTERBUFFERState *s = DO_UPCAST(FILTERBUFFERState, nc, nc);
+    NetQueue *queue = s->inflight_queue;
+    NetPacket *packet;
+    int ret;
+
+    while (queue && !QTAILQ_EMPTY(&queue->packets)) {
+        packet = QTAILQ_FIRST(&queue->packets);
+        QTAILQ_REMOVE(&queue->packets, packet, entry);
+        queue->nq_count--;
+
+        if (packet->flags & QEMU_NET_PACKET_FLAG_RAW) {
+            ret = qemu_send_packet_raw(nc, packet->data, packet->size);
+        } else {
+            ret = qemu_send_packet_async(nc, packet->data, packet->size,
+                                         packet->sent_cb);
+        }
+
+        if (ret == 0) {
+            queue->nq_count++;
+            QTAILQ_INSERT_HEAD(&queue->packets, packet, entry);
+            /* shedule out */
+            return;
+        }
+
+        g_free(packet);
+    }
+
+    if (QTAILQ_EMPTY(&queue->packets)) {
+        g_free(queue);
+        s->inflight_queue = NULL;
+    }
+}
+
+static void filter_buffer_flush_bh(void *opaque)
+{
+    FILTERBUFFERState *s = opaque;
+    NetClientState *nc = &s->nc;
+    filter_buffer_flush(nc);
+}
+
 int net_init_filter_buffer(const NetClientOptions *opts, const char *name,
                            NetClientState *peer, Error **errp)
 {
@@ -71,10 +142,44 @@ int net_init_filter_buffer(const NetClientOptions *opts, 
const char *name,
     }
 
     nc = qemu_new_net_client(&net_filter_buffer_info, peer, "filter_buffer", 
name);
+    /*
+     * we are buffering guest output packets, our buffered packets should be
+     * sent to real network backend, so our peer should be that backend
+     */
+    nc->peer = filter_backend(filter);
     s = DO_UPCAST(FILTERBUFFERState, nc, nc);
     s->filter = filter;
     s->interval = bufferopt->has_interval ? bufferopt->interval : 0;
+    s->flush_bh = qemu_bh_new(filter_buffer_flush_bh, s);
     filter_add_plugin(filter, nc);
 
     return 0;
 }
+
+static void filter_buffer_release_one(NetClientState *nc)
+{
+    FILTERBUFFERState *s = DO_UPCAST(FILTERBUFFERState, nc, nc);
+
+    /* flush inflight packets */
+    if (s->inflight_queue) {
+        filter_buffer_flush(nc);
+    }
+
+    s->inflight_queue = nc->incoming_queue;
+    nc->incoming_queue = qemu_new_net_queue(nc);
+    qemu_bh_schedule(s->flush_bh);
+}
+
+/* public APIs */
+void filter_buffer_release_all(void)
+{
+    NetClientState *ncs[MAX_QUEUE_NUM];
+    int queues, i;
+
+    queues = qemu_find_net_clients_by_model("filter_buffer", ncs,
+                                            MAX_QUEUE_NUM);
+
+    for (i = 0; i < queues; i++) {
+        filter_buffer_release_one(ncs[i]);
+    }
+}
-- 
1.9.1