Signed-off-by: Yang Hongyang <yan...@cn.fujitsu.com>
---
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
