[PATCH v3 16/16] hub: add the support for hub own flow control

[zwu.kernel@gmail.com]

From: Zhi Yong Wu <wuzhy@linux.vnet.ibm.com>

Signed-off-by: Zhi Yong Wu <wuzhy@linux.vnet.ibm.com>
---
 net/hub.c   |   35 ++++++++++++++++++++++++++++++++---
 net/hub.h   |    2 ++
 net/queue.c |    5 +++++
 3 files changed, 39 insertions(+), 3 deletions(-)

diff --git a/net/hub.c b/net/hub.c
index 8a583ab..d27c52a 100644
--- a/net/hub.c
+++ b/net/hub.c
@@ -28,6 +28,7 @@ typedef struct NetHubPort {
     QLIST_ENTRY(NetHubPort) next;
     NetHub *hub;
     unsigned int id;
+    uint64_t nr_packets;
 } NetHubPort;
 
 struct NetHub {
@@ -39,19 +40,37 @@ struct NetHub {
 
 static QLIST_HEAD(, NetHub) hubs = QLIST_HEAD_INITIALIZER(&hubs);
 
+static void net_hub_receive_completed(NetClientState *nc, ssize_t len)
+{
+    NetHubPort *port = DO_UPCAST(NetHubPort, nc, nc);
+    port->nr_packets--;
+    if (!port->nr_packets) {
+        qemu_net_queue_flush(nc->peer->send_queue);
+    }
+}
+
+void net_hub_port_packet_stats(NetClientState *nc)
+{
+    NetHubPort *port = DO_UPCAST(NetHubPort, nc, nc);
+
+    port->nr_packets++;
+}
+
 static ssize_t net_hub_receive(NetHub *hub, NetHubPort *source_port,
                                const uint8_t *buf, size_t len)
 {
     NetHubPort *port;
+    ssize_t ret = 0;
 
     QLIST_FOREACH(port, &hub->ports, next) {
         if (port == source_port) {
             continue;
         }
 
-        qemu_send_packet(&port->nc, buf, len);
+       ret = qemu_send_packet_async(&port->nc, buf, len,
+                                    net_hub_receive_completed);
     }
-    return len;
+    return ret;
 }
 
 static ssize_t net_hub_receive_iov(NetHub *hub, NetHubPort *source_port,
@@ -65,7 +84,8 @@ static ssize_t net_hub_receive_iov(NetHub *hub, NetHubPort *source_port,
             continue;
         }
 
-        ret = qemu_sendv_packet(&port->nc, iov, iovcnt);
+        ret = qemu_sendv_packet_async(&port->nc, iov, iovcnt,
+                                      net_hub_receive_completed);
     }
     return ret;
 }
@@ -84,6 +104,13 @@ static NetHub *net_hub_new(unsigned int id)
     return hub;
 }
 
+static int net_hub_port_can_receive(NetClientState *nc)
+{
+    NetHubPort *port = DO_UPCAST(NetHubPort, nc, nc);
+
+    return port->nr_packets ? 0 : 1;
+}
+
 static ssize_t net_hub_port_receive(NetClientState *nc,
                                     const uint8_t *buf, size_t len)
 {
@@ -110,6 +137,7 @@ static void net_hub_port_cleanup(NetClientState *nc)
 static NetClientInfo net_hub_port_info = {
     .type = NET_CLIENT_TYPE_HUB,
     .size = sizeof(NetHubPort),
+    .can_receive = net_hub_port_can_receive,
     .receive = net_hub_port_receive,
     .receive_iov = net_hub_port_receive_iov,
     .cleanup = net_hub_port_cleanup,
@@ -128,6 +156,7 @@ static NetHubPort *net_hub_port_new(NetHub *hub)
     port = DO_UPCAST(NetHubPort, nc, nc);
     port->id = id;
     port->hub = hub;
+    port->nr_packets = 0;
 
     QLIST_INSERT_HEAD(&hub->ports, port, next);
 
diff --git a/net/hub.h b/net/hub.h
index d04f1b1..542e657 100644
--- a/net/hub.h
+++ b/net/hub.h
@@ -23,4 +23,6 @@ void net_hub_info(Monitor *mon);
 int net_hub_id_for_client(NetClientState *nc, unsigned int *id);
 void net_hub_check_clients(void);
 
+void net_hub_port_packet_stats(NetClientState *nc);
+
 #endif /* NET_HUB_H */
diff --git a/net/queue.c b/net/queue.c
index 7484d2a..ebf18aa 100644
--- a/net/queue.c
+++ b/net/queue.c
@@ -22,6 +22,7 @@
  */
 
 #include "net/queue.h"
+#include "net/hub.h"
 #include "qemu-queue.h"
 #include "net.h"
 
@@ -101,6 +102,8 @@ static ssize_t qemu_net_queue_append(NetQueue *queue,
 
     QTAILQ_INSERT_TAIL(&queue->packets, packet, entry);
 
+    net_hub_port_packet_stats(sender);
+
     return size;
 }
 
@@ -134,6 +137,8 @@ static ssize_t qemu_net_queue_append_iov(NetQueue *queue,
 
     QTAILQ_INSERT_TAIL(&queue->packets, packet, entry);
 
+    net_hub_port_packet_stats(sender);
+    
     return packet->size;
 }
 
-- 
1.7.6