Implement chunk queing for the udp sender.

diff --git a/udp_send.c b/udp_send.c
index 7704b0a1a8f01b1580bae58c59a2817ca324371e..1408ae7f251bc4fe0c4e79e67e00dc7914d0b91b 100644 (file)
--- a/udp_send.c
+++ b/udp_send.c
@@ -76,13 +76,32 @@ static void udp_send_buf(char *buf, size_t len)
        list_for_each_entry_safe(ut, tmp, &targets, node) {
                if (ut->fd < 0)
                        continue;
        list_for_each_entry_safe(ut, tmp, &targets, node) {
                if (ut->fd < 0)
                        continue;

-               ret = write_nonblock(ut->fd, buf, len, len);
+               ret = send_queued_chunks(ut->fd, ut->cq, 0);

                if (ret < 0) {
                        udp_delete_target(ut, para_strerror(-ret));
                        continue;
                }
                if (ret < 0) {
                        udp_delete_target(ut, para_strerror(-ret));
                        continue;
                }

-               if (ret != len)
-                       PARA_WARNING_LOG("short write %zu/%zu\n", ret, len);
+               if (!len)
+                       continue;
+               if (!ret) { /* still data left in the queue */
+                       ret = cq_enqueue(ut->cq, buf, len);
+                       if (ret < 0) {
+                               udp_delete_target(ut, para_strerror(-ret));
+                               continue;
+                       }
+               }
+               ret = write_nonblock(ut->fd, buf, len, 0);
+               if (ret < 0) {
+                       udp_delete_target(ut, para_strerror(-ret));
+                       continue;
+               }
+               if (ret != len) {
+                       ret = cq_enqueue(ut->cq, buf + ret, len - ret);
+                       if (ret < 0) {
+                               udp_delete_target(ut, para_strerror(-ret));
+                               continue;
+                       }
+               }

        }
 }
 
        }
 }