add read blocked / resume callbacks

diff --git a/src/lib/bio/fd.c b/src/lib/bio/fd.c
index 57adc6ec4bd4de3a4a6544199f4b7a38f05bc047..ced826da51a87bff97d995358f44d49217f3f4a7 100644 (file)
--- a/src/lib/bio/fd.c
+++ b/src/lib/bio/fd.c
@@ -112,12 +112,8 @@ static ssize_t fr_bio_fd_read_stream(fr_bio_t *bio, UNUSED void *packet_ctx, voi
        ssize_t rcode;
        fr_bio_fd_t *my = talloc_get_type_abort(bio, fr_bio_fd_t);
 
-       my->info.read_blocked = false;
-
 retry:
        rcode = read(my->info.socket.fd, buffer, size);
-       if (rcode > 0) return rcode;
-
        if (rcode == 0) {
                /*
                 *      Stream sockets return 0 at EOF.  However, we want to distinguish that from the case of datagram
@@ -131,9 +127,7 @@ retry:
                return fr_bio_error(EOF);
        }
 
-#undef flag_blocked
-#define flag_blocked read_blocked
-#include "fd_errno.h"
+#include "fd_read.h"
 
        return fr_bio_error(IO);
 }
@@ -151,15 +145,10 @@ static ssize_t fr_bio_fd_read_connected_datagram(fr_bio_t *bio, UNUSED void *pac
        ssize_t rcode;
        fr_bio_fd_t *my = talloc_get_type_abort(bio, fr_bio_fd_t);
 
-       my->info.read_blocked = false;
-
 retry:
        rcode = read(my->info.socket.fd, buffer, size);
-       if (rcode >= 0) return rcode;
 
-#undef flag_blocked
-#define flag_blocked read_blocked
-#include "fd_errno.h"
+#include "fd_read.h"
 
        return fr_bio_error(IO);
 }
@@ -174,8 +163,6 @@ static ssize_t fr_bio_fd_recvfrom(fr_bio_t *bio, void *packet_ctx, void *buffer,
        socklen_t salen;
        struct sockaddr_storage sockaddr;
 
-       my->info.read_blocked = false;
-
 retry:
        salen = sizeof(sockaddr);
 
@@ -190,14 +177,9 @@ retry:
 
                (void) fr_ipaddr_from_sockaddr(&addr->socket.inet.src_ipaddr, &addr->socket.inet.src_port,
                                               &sockaddr, salen);
-               return rcode;
        }
 
-       if (rcode == 0) return rcode;
-
-#undef flag_blocked
-#define flag_blocked read_blocked
-#include "fd_errno.h"
+#include "fd_read.h"
 
        return fr_bio_error(IO);
 }
@@ -280,8 +262,6 @@ static ssize_t fd_fd_recvfromto_common(fr_bio_fd_t *my, void *packet_ctx, void *
        from.ss_family = AF_UNSPEC;
 #endif
 
-       my->info.read_blocked = false;
-
        memset(&my->cbuf, 0, sizeof(my->cbuf));
        memset(&my->msgh, 0, sizeof(struct msghdr));
 
@@ -307,15 +287,9 @@ retry:
 
                (void) fr_ipaddr_from_sockaddr(&addr->socket.inet.src_ipaddr, &addr->socket.inet.src_port,
                                               &from, my->msgh.msg_namelen);
-
-               return rcode;
        }
 
-       if (rcode == 0) return rcode;
-
-#undef flag_blocked
-#define flag_blocked read_blocked
-#include "fd_errno.h"
+#include "fd_read.h"
 
        return fr_bio_error(IO);
 }
@@ -1163,15 +1137,11 @@ static ssize_t fr_bio_fd_read_discard(fr_bio_t *bio, UNUSED void *packet_ctx, vo
        ssize_t rcode;
        fr_bio_fd_t *my = talloc_get_type_abort(bio, fr_bio_fd_t);
 
-       my->info.read_blocked = false;
-
 retry:
        rcode = read(my->info.socket.fd, buffer, size);
-       if (rcode >= 0) return 0;
+       if (rcode > 0) rcode = 0; /* always return that we read no data */
 
-#undef flag_blocked
-#define flag_blocked read_blocked
-#include "fd_errno.h"
+#include "fd_read.h"
 
        return fr_bio_error(IO);
 }

diff --git a/src/lib/bio/fd_read.h b/src/lib/bio/fd_read.h
new file mode 100644 (file)
index 0000000..bf2e589
--- /dev/null
+++ b/src/lib/bio/fd_read.h
@@ -0,0 +1,34 @@
+/*
+ *     Common finalization code for the read functions.
+ *
+ *     This is in a header file because of "goto retry" in fd_errno.h.
+ *
+ *     @todo - do we want the callbacks to notify the _previous_ BIO in the chain?  That way the top-level
+ *     BIO can notify the application.
+ */
+if (rcode > 0) {
+       /*
+        *      We weren't blocked, so we're still not blocked.
+        */
+       if (!my->info.read_blocked) {
+               return rcode;   
+       }
+
+       /*
+        *      We were blocked.  Since we just read data, we're now unblocked.
+        */
+       my->info.read_blocked = false;
+
+       /*
+        *      Call the "resume" function when we transition to being unblocked.
+        */
+       if (my->cb.read_resume) my->cb.read_resume((fr_bio_t *) my);
+
+       return rcode;
+}
+
+if (rcode == 0) return rcode;
+
+#undef flag_blocked
+#define flag_blocked read_blocked
+#include "fd_errno.h"