Index: ossp-pkg/sio/sio_bio.c
RCS File: /v/ossp/cvs/ossp-pkg/sio/sio_bio.c,v
rcsdiff -q -kk '-r1.2' '-r1.3' -u '/v/ossp/cvs/ossp-pkg/sio/sio_bio.c,v' 2>/dev/null
--- sio_bio.c	2002/11/19 22:29:20	1.2
+++ sio_bio.c	2002/11/24 19:39:17	1.3
@@ -1,23 +1,188 @@
 #include <stddef.h>
 #include <stdlib.h>
 #include <string.h>
+#include <assert.h>
 
 #include "al.h"
 #include "sio.h"
 
 #include <openssl/bio.h>
+#include <openssl/err.h>
+
+typedef enum {
+    INIT,
+    LOWER,
+    UPPER
+} state_t;
 
 typedef struct {
-    BIO *bio;
-    size_t inputsize;
-    al_label_t data_label;
-    al_label_t eof_label;
-    al_label_t error_label;
-    char       eof;
-    char       error;
-    size_t     total;
+    state_t      state;
+    int          isoutput;
+    al_t        *in_buf;
+    al_t        *out_buf;
+    BIO         *bio;
+    BIO         *nbio;
+    al_t        *al_in, *al_out;   /* upstream buffers */
+    int          reader_writes;
+    int          writer_reads;
+    int          issink;
+    int          freebio;
+    size_t       inputsize;
+    al_label_t   data_label;
+    al_label_t   eof_label;
+    al_label_t   error_label;
+    char         eof;
+    char         error;
+    size_t       total;
+    int          should_retry;
+    int          flush_upper;
+    int          flush_lower;
+    int          eof_reached;
 } private_t;
 
+/********************************************************************/
+
+static
+int b_new(BIO *bi)
+{
+    bi->init  = 0;
+    bi->num   = -1;
+    bi->flags = 0;
+    bi->ptr   = NULL;
+    return 1;
+}
+
+static
+int b_free(BIO *a)
+{
+    if (a == NULL)
+        return 0;
+    return 1;
+}
+
+static
+long b_pending(private_t *my)
+{
+    return my ? al_bytes(my->al_in) : 0;
+}
+static
+long b_wpending(private_t *my)
+{
+    return my ? al_bytes(my->al_out) : 0;
+}
+static
+long b_setflush(private_t *my)
+{
+    if (my) my->flush_upper = 1;
+    return 1;
+}
+static
+long b_geteof(private_t *my)
+{
+    return my->eof_reached;
+}
+static
+long b_ctrl(BIO *b, int cmd, long num, void *ptr)
+{
+    long ret = 1;
+
+    switch (cmd) {
+    case BIO_CTRL_PENDING:
+        ret = b_pending((private_t *)b->ptr);
+        break;
+    case BIO_CTRL_WPENDING:
+        ret = b_wpending((private_t *)b->ptr);
+        break;
+    case BIO_CTRL_FLUSH:
+        ret = b_setflush((private_t *)b->ptr);
+        break;
+    case BIO_CTRL_EOF:
+        ret = b_geteof((private_t *)b->ptr);
+        break;
+    case BIO_C_SET_FD:
+        b->ptr  = ptr;
+        b->init = 1;
+        ret = 1;
+        break;
+    case BIO_CTRL_RESET:
+    case BIO_CTRL_SET:
+    case BIO_CTRL_SET_CLOSE:
+    case BIO_CTRL_DUP:
+        ret = 1;
+        break;
+    case BIO_CTRL_GET_CLOSE:
+    case BIO_CTRL_INFO:
+    case BIO_CTRL_GET:
+    default:
+        ret = 0;
+        break;
+    }
+
+    return ret;
+}
+
+static
+int b_read(BIO *b, char *out, int outl)
+{
+    private_t *my = (private_t *)b->ptr;
+    size_t n, m;
+    al_label_t label;
+
+    m = 0;
+    if (al_bytes(my->al_in) > 0) {
+        al_firstlabel(my->al_in, 0, 1, AL_FORWARD, NULL, &label);
+        if (label == my->data_label) {
+            al_flatten(my->al_in, 0, outl, AL_FORWARD_SPAN, label, out, &n);
+            m = n;
+        } else {
+            al_flatten(my->al_in, 0, outl, AL_FORWARD_SPAN, label, NULL, &n);
+            m = -1;
+        }
+        al_splice(my->al_in, 0, n, NULL, NULL);
+    }
+
+    if (m < 0)
+        my->eof_reached = 1;
+
+    BIO_clear_retry_flags(b);
+    if (m == 0)
+        BIO_set_retry_read(b);
+
+    return m;
+}
+
+static
+int b_write(BIO *b, const char *in, int inl)
+{
+    private_t *my = (private_t *)b->ptr;
+
+    al_append_bytes(my->al_out, in, inl, my->data_label);
+
+    return inl;
+}
+
+static
+int b_puts(BIO *b, const char *s)
+{
+    return b_write(b, s, strlen(s));
+}
+
+static
+BIO_METHOD b_method = {
+    BIO_TYPE_NONE,
+    "SIO SSL",
+    b_write,
+    b_read,
+    b_puts,
+    NULL,
+    b_ctrl,
+    b_new,
+    b_free,
+    NULL
+};
+
+/********************************************************************/
+
 /*
  * create stage
  *
@@ -32,8 +197,11 @@
     if (my == NULL)
         return SIO_ERR_MEM;
 
-    my->bio          = NULL;
-    my->eof          = '\0';
+    my->bio           = NULL;
+    my->nbio          = NULL;
+    my->issink        = 1;
+    my->freebio       = 0;
+    my->eof           = '\0';
     my->error        = '\0';
 
     sio_label(sio, SIO_LN_DATA,  &my->data_label);
@@ -60,6 +228,10 @@
         my->bio       = (BIO *)val;
     } else if (!strcmp(name, "inputsize")) {
         my->inputsize = *(int *)val;
+    } else if (!strcmp(name, "issink")) {
+        my->issink    = *(int *)val;
+    } else if (!strcmp(name, "freebio")) {
+        my->freebio   = *(int *)val;
     } else {
         return SIO_ERR_ARG;
     }
@@ -95,24 +267,189 @@
 static
 sio_rc_t siobio_openw(sio_t *sio, al_t *al, void *u)
 {
+    private_t *my = (private_t *)u;
+
+    if (my->bio == NULL)
+        return SIO_ERR_ARG;
+
+    if (al_create(&my->al_in) != AL_OK)
+        return SIO_ERR_INT;
+    if (al_create(&my->al_out) != AL_OK) {
+        al_destroy(my->al_in); my->al_in = NULL;
+        return SIO_ERR_INT;
+    }
+    if (al_create(&my->in_buf) != AL_OK) {
+        al_destroy(my->al_out); my->al_out = NULL;
+        al_destroy(my->al_in); my->al_in = NULL;
+        return SIO_ERR_INT;
+    }
+    if (al_create(&my->out_buf) != AL_OK) {
+        al_destroy(my->in_buf); my->in_buf = NULL;
+        al_destroy(my->al_out); my->al_out = NULL;
+        al_destroy(my->al_in); my->al_in = NULL;
+        return SIO_ERR_INT;
+    }
+
+    if (!my->issink) {
+        my->nbio = BIO_new(&b_method);
+        BIO_ctrl(my->nbio, BIO_C_SET_FD, 0, (void *)my);
+        BIO_push(my->bio, my->nbio);
+    }
+
+    my->writer_reads  = 0;
+    my->reader_writes = 0;
+
+    my->state = INIT;
+
     return SIO_OK;
 }
 
 static
 sio_rc_t siobio_closew(sio_t *sio, al_t *al, void *u)
 {
+    private_t *my = (private_t *)u;
+
+    if (my->nbio != NULL) {
+        BIO_pop(my->bio);
+        BIO_free(my->nbio);
+        my->nbio = NULL;
+    }
+
+    al_destroy(my->out_buf); my->out_buf = NULL;
+    al_destroy(my->in_buf); my->in_buf = NULL;
+    al_destroy(my->al_out); my->al_out = NULL;
+    al_destroy(my->al_in); my->al_in = NULL;
+
     return SIO_OK;
 }
 
+/********************************************************************/
+
+/*
+ * auto-destructor for allocated input buffer
+ */
 static
 void freebiobuf(char *p, size_t n, void *u)
 {
     free(p);
 }
+
+/*
+ * chunk traversal of output buffer
+ *
+ * counts my->total
+ * sets my->should_retry
+ */
 static
-sio_rc_t siobio_input(sio_t *sio, al_t *al, void *u)
+al_rc_t siobio_write_chunk(al_chunk_t *alc, void *u)
 {
     private_t *my = (private_t *)u;
+    int        n  = al_chunk_len(alc);
+    al_rc_t arc   = AL_OK;
+
+    if (al_same_label(alc, my->data_label)) {
+        char *p = al_chunk_ptr(alc, 0);
+        int   i, t;
+
+        for (t=0; t<n; t+=i) {
+            i = my->bio ? BIO_write(my->bio, p+t, n-t) : (n-t);
+            if (i <= 0) {
+                if (!BIO_should_retry(my->bio)) {
+                    my->should_retry = 1;
+                    arc = AL_ERR_EOF;
+                    break;
+                }
+                i = 0;
+            }
+            my->total += i;
+        }
+    } else {
+        my->total += n;
+        my->flush_lower = 1;
+    }
+
+    return arc;
+}
+
+/********************************************************************/
+
+/* UPSTREAM layer */
+
+static
+sio_rc_t siobio_input_upper(private_t *my, al_t *al)
+{
+    /* flush output */
+    if (my->flush_upper)
+        return SIO_SCHED_CROSS;
+
+    if (al_bytes(al) <= 0)
+        return SIO_SCHED_UP;
+
+    al_splice(al, 0, al_bytes(al), NULL, my->al_in);
+
+    if (!my->isoutput) {
+        my->state = LOWER;
+        return SIO_SCHED_LOOP;
+    }
+
+    return SIO_SCHED_CROSS;
+}
+
+static
+sio_rc_t siobio_output_upper(private_t *my, al_t *al)
+{
+    /* flush output */
+    if (my->flush_upper) {
+        my->flush_upper = 0;
+        al_splice(al, al_bytes(al), 0, my->al_out, NULL);
+        return SIO_SCHED_UP;
+    }
+
+    if (my->should_retry) {
+        my->should_retry = 0;
+        al_splice(al, 0, al_bytes(al), my->al_in, NULL);
+        return SIO_SCHED_CROSS;
+    }
+
+    if (my->isoutput) {
+        my->state = LOWER;
+        return SIO_SCHED_LOOP;
+    }
+
+    return SIO_SCHED_CROSS;
+}
+
+/* DOWNSTREAM layer */
+static
+sio_rc_t siobio_input_lower(private_t *my, al_t *al)
+{
+    al_splice(al, al_bytes(al), 0, my->in_buf, NULL);
+    if (al_bytes(al) > 0) {
+        my->state = INIT;
+        return SIO_SCHED_DOWN;
+    }
+
+    my->state = UPPER;
+    return SIO_SCHED_LOOP;
+}
+
+static
+sio_rc_t siobio_output_lower(private_t *my, al_t *al)
+{
+    al_splice(al, 0, al_bytes(al), NULL, my->out_buf);
+    if (al_bytes(my->out_buf) <= 0) {
+        my->state = INIT;
+        return SIO_SCHED_DOWN;
+    }
+
+    my->state = UPPER;
+    return SIO_SCHED_LOOP;
+}
+
+/* BIO layer */
+static
+void siobio_bio_read(private_t *my)
+{
     char *p;
     int n;
 
@@ -120,64 +457,114 @@
     if (n == 0 || n > my->inputsize)
         n = my->inputsize;
     p = malloc(n);
+    assert(p != NULL);
 
-    if (p != NULL)
+    if (my->bio) {
         do {
             n = BIO_read(my->bio, p, n);
         } while (n <= 0 && BIO_should_retry(my->bio));
+    } else
+        n = -2;
 
-    if (p == NULL || n <= 0) {
+    if (n < 0) {
         free(p);
         if (n < -1)
-            al_append_bytes(al, &my->error, sizeof(my->error), my->error_label);
+            al_append_bytes(my->in_buf, &my->error,
+                            sizeof(my->error), my->error_label);
         else
-            al_append_bytes(al, &my->eof, sizeof(my->eof), my->eof_label);
-    } else
-        al_attach_buffer(al, p, n, my->data_label, freebiobuf, NULL);
+            al_append_bytes(my->in_buf, &my->eof,
+                            sizeof(my->eof), my->eof_label);
+    } else if (n > 0)
+        al_attach_buffer(my->in_buf, p, n, my->data_label, freebiobuf, NULL);
+}
+
+static
+void siobio_bio_write(private_t *my)
+{
+    my->should_retry = 0;
+    my->total        = 0;
+    al_traverse_cb(my->out_buf, 0, al_bytes(my->out_buf),
+                   AL_FORWARD, NULL,
+                   siobio_write_chunk, (void *)my);
+    al_splice(my->out_buf, 0, my->total, NULL, NULL);
+
+    if (my->flush_lower) {
+        my->flush_upper = 1;
+        if (BIO_flush(my->bio) > 0)
+            my->flush_lower = 0;
+    }
+}
+
+/********************************************************************/
 
+static
+sio_rc_t siobio_eof(private_t *my, al_t *al)
+{
+    al_splice(al, 0, al_bytes(al), NULL, NULL);
     return SIO_SCHED_DOWN;
 }
 
 static
-al_rc_t siobio_write_chunk(al_chunk_t *alc, void *u)
+sio_rc_t siobio_input(sio_t *sio, al_t *al, void *u)
 {
     private_t *my = (private_t *)u;
-    al_rc_t arc = AL_OK;
+    sio_rc_t rc;
 
-    if (al_same_label(alc, my->data_label)) {
-        char *p = al_chunk_ptr(alc, 0);
-        int   n = al_chunk_len(alc);
-        int   i, t;
-
-        for (t=0; t<n; t+=i) {
-            i = BIO_write(my->bio, p+t, n-t);
-            if (i <= 0) {
-                if (!BIO_should_retry(my->bio)) {
-                    arc = AL_ERR_EOF;
-                    break;
-                }
-                i = 0;
-            }
-            my->total += i;
-        }
-    } else {
-        my->total += al_chunk_len(alc);
+    switch (my->state) {
+    case INIT:
+        my->isoutput = 0;
+        my->state    = LOWER;
+        rc = SIO_SCHED_LOOP;
+        break;
+    case LOWER:
+        siobio_bio_read(my);
+        rc = siobio_input_lower(my, al);
+        break;
+    case UPPER:
+        rc = siobio_input_upper(my, al);
+        break;
     }
 
-    return arc;
+    return rc;
 }
+
 static
 sio_rc_t siobio_output(sio_t *sio, al_t *al, void *u)
 {
     private_t *my = (private_t *)u;
+    sio_rc_t rc;
 
-    my->total = 0;
-    al_traverse_cb(al, 0, al_bytes(al), AL_FORWARD, my->data_label,
-                   siobio_write_chunk, (void *)my);
+    switch (my->state) {
+    case INIT:
+        my->isoutput = 1;
+        my->state    = LOWER;
+        rc = SIO_SCHED_LOOP;
+        break;
+    case LOWER:
+        rc = siobio_output_lower(my, al);
+        siobio_bio_write(my);
+        break;
+    case UPPER:
+        rc = siobio_output_upper(my, al);
+        break;
+    }
 
-    al_splice(al, 0, al_bytes(al), NULL, NULL);
+    return rc;
+}
 
-    return SIO_SCHED_DOWN;
+static
+sio_rc_t siobio_shutdown(sio_t *sio, void *u)
+{
+    private_t *my = (private_t *)u;
+
+    if (my->freebio && my->bio != NULL) {
+        BIO_free(my->bio);
+        my->bio = NULL;
+        my->flush_upper = 1;
+        my->state       = UPPER;
+    }
+
+    return SIO_OK;
 }
 
 sio_module_t sio_module_bio = {
@@ -190,6 +577,7 @@
     siobio_openw,
     siobio_closew,
     siobio_input,
-    siobio_output
+    siobio_output,
+    siobio_shutdown
 };