Introduced the a_Dpip_dsh_tryflush() call

This adds an internal buffer in the dsh data structure.

3 files changed, 94 insertions, 66 deletions

diff --git a/dpi/file.c b/dpi/file.c
index c8c14b14..2a40d2d1 100644
--- a/dpi/file.c
+++ b/dpi/file.c
@@ -92,8 +92,6 @@ typedef struct {
    int err_code;
    int flags;
    int old_style;
-
-   Dstr *dbuf;
 } ClientInfo;
 
 /*
@@ -694,6 +692,7 @@ static void File_get(ClientInfo *client, const char *filename,
  */
 static int File_send_file(ClientInfo *client)
 {
+//#define LBUF 1
 #define LBUF 16*1024
 
    const char *ct;
@@ -746,29 +745,24 @@ static int File_send_file(ClientInfo *client)
 
    } else if (client->state == st_http) {
       /* Send body -- raw file contents */
-      if (client->dbuf->len > 0) {
-         /* send pending data */
-         st = a_Dpip_dsh_trywrite(client->sh, 
-                                  client->dbuf->str, client->dbuf->len);
-         if (st > 0)
-            dStr_erase(client->dbuf, 0, st);
+      if ((st = a_Dpip_dsh_tryflush(client->sh)) < 0) {
+         client->flags |= (st == -3) ? FILE_ERR : 0;
       } else {
-         /* ok to send new data */
+         /* no pending data, let's send new data */
          do {
-            st = read(client->file_fd, buf, LBUF);
-         } while (st < 0 && errno == EINTR);
-         if (st == -1) {
-            MSG("\nexit(1), ERROR while reading from file '%s': %s\n\n",
+            st2 = read(client->file_fd, buf, LBUF);
+         } while (st2 < 0 && errno == EINTR);
+         if (st2 < 0) {
+            MSG("\nERROR while reading from file '%s': %s\n\n",
                 client->filename, dStrerror(errno));
-            exit(1);
-         } else if (st == 0) {
+            client->flags |= FILE_ERR;
+         } else if (st2 == 0) {
             client->state = st_content;
             client->flags |= FILE_DONE;
          } else {
-            /* partial write */
-            st2 = a_Dpip_dsh_trywrite(client->sh, buf, st);
-            if (st2 > 0 && st2 < st)
-               dStr_append_l(client->dbuf, buf + st2, st - st2);
+            /* ok to write */
+            st = a_Dpip_dsh_trywrite(client->sh, buf, st2);
+            client->flags |= (st == -3) ? FILE_ERR : 0;
          }
       }
    }
@@ -889,7 +883,6 @@ static ClientInfo *File_add_client(int sock_fd)
    new_client->err_code = 0;
    new_client->flags = FILE_READ;
    new_client->old_style = OLD_STYLE;
-   new_client->dbuf = dStr_sized_new(8*1024);
 
    dList_append(Clients, new_client);
    return new_client;
@@ -908,7 +901,6 @@ static void File_remove_client(ClientInfo *client)
    File_close(client->file_fd);
    dFree(client->orig_url);
    dFree(client->filename);
-   dStr_free(client->dbuf, TRUE);
    File_dillodir_free(client->d_dir);
 
    dFree(client);
@@ -1053,7 +1045,7 @@ static int File_check_fds(uint_t seconds)
 int main(void)
 {
    struct sockaddr_in sin;
-   socklen_t address_size;
+   socklen_t sin_sz;
    int tmp_fd, c_st, st = 1;
 
    /* Arrange the cleanup function for abnormal terminations */
@@ -1065,6 +1057,7 @@ int main(void)
      signal (SIGTERM, SIG_IGN);
 
    MSG("(v.2) accepting connections...\n");
+   //sleep(20);
 
    /* initialize observed file descriptors */
    FD_ZERO (&read_set);
@@ -1075,7 +1068,7 @@ int main(void)
    Clients = dList_new(512);
 
    /* some OSes may need this... */
-   address_size = sizeof(struct sockaddr_in);
+   sin_sz = sizeof(sin);
 
    /* start the service loop */
    while (!DPIBYE) {
@@ -1092,13 +1085,17 @@ int main(void)
 
       if (FD_ISSET(STDIN_FILENO, &read_set)) {
          /* accept the incoming connection */
-         tmp_fd = accept(STDIN_FILENO, (struct sockaddr *)&sin, &address_size);
+         do {
+            tmp_fd = accept(STDIN_FILENO, (struct sockaddr *)&sin, &sin_sz);
+         } while (tmp_fd < 0 && errno == EINTR);
          if (tmp_fd == -1) {
             MSG(" accept() %s\n", dStrerror(errno));
             break;
          } else {
+            _MSG(" accept() fd=%d\n", tmp_fd);
+            /* Set nonblocking */
+            fcntl(tmp_fd, F_SETFL, O_NONBLOCK | fcntl(tmp_fd, F_GETFL));
             /* Create and initialize a new client */
-            MSG(" accept() fd=%d\n", tmp_fd);
             File_add_client(tmp_fd);
          }
          continue;
diff --git a/dpip/dpip.c b/dpip/dpip.c
index b7f71d79..de0a6989 100644
--- a/dpip/dpip.c
+++ b/dpip/dpip.c
@@ -247,8 +247,8 @@ Dsh *a_Dpip_dsh_new(int fd_in, int fd_out, int flush_sz)
    dsh->fd_out = fd_out;
 
    /* init buffer */
-   dsh->dbuf = dStr_sized_new(8 *1024);
-   dsh->rd_dbuf = dStr_sized_new(8 *1024);
+   dsh->wrbuf = dStr_sized_new(8 *1024);
+   dsh->rdbuf = dStr_sized_new(8 *1024);
    dsh->flush_sz = flush_sz;
    dsh->mode = DPIP_TAG;
    if (fcntl(dsh->fd_in, F_GETFL) & O_NONBLOCK)
@@ -267,9 +267,9 @@ int a_Dpip_dsh_write(Dsh *dsh, int flush, const char *Data, int DataSize)
    int blocking, old_flags, st, sent = 0, ret = 1;
 
    /* append to buf */
-   dStr_append_l(dsh->dbuf, Data, DataSize);
+   dStr_append_l(dsh->wrbuf, Data, DataSize);
 
-   if (!flush || dsh->dbuf->len == 0)
+   if (!flush || dsh->wrbuf->len == 0)
       return 0;
 
    blocking = !(dsh->mode & DPIP_NONBLOCK);
@@ -280,7 +280,7 @@ int a_Dpip_dsh_write(Dsh *dsh, int flush, const char *Data, int DataSize)
    }
 
    while (1) {
-      st = write(dsh->fd_out, dsh->dbuf->str + sent, dsh->dbuf->len - sent);
+      st = write(dsh->fd_out, dsh->wrbuf->str + sent, dsh->wrbuf->len - sent);
       if (st < 0) {
          if (errno == EINTR) {
             continue;
@@ -290,8 +290,8 @@ int a_Dpip_dsh_write(Dsh *dsh, int flush, const char *Data, int DataSize)
          }
       } else {
          sent += st;
-         if (sent == dsh->dbuf->len) {
-            dStr_truncate(dsh->dbuf, 0);
+         if (sent == dsh->wrbuf->len) {
+            dStr_truncate(dsh->wrbuf, 0);
             ret = 0;
             break;
          }
@@ -309,9 +309,9 @@ int a_Dpip_dsh_write(Dsh *dsh, int flush, const char *Data, int DataSize)
 /*
  * Return value: 1..DataSize sent, -1 eagain, or -3 on big Error
  */
-int a_Dpip_dsh_trywrite(Dsh *dsh, const char *Data, int DataSize)
+static int Dpip_dsh_trywrite(Dsh *dsh, const char *Data, int DataSize)
 {
-   int blocking, old_flags, st, ret = -3, sent = 0;
+   int blocking, old_flags, st, ret = -3;
 
    blocking = !(dsh->mode & DPIP_NONBLOCK);
    if (blocking) {
@@ -320,24 +320,19 @@ int a_Dpip_dsh_trywrite(Dsh *dsh, const char *Data, int DataSize)
       fcntl(dsh->fd_in, F_SETFL, O_NONBLOCK | old_flags);
    }
 
-   while (1) {
-      st = write(dsh->fd_out, Data + sent, DataSize - sent);
-      if (st < 0) {
-         if (errno == EINTR) {
-            continue;
-         } else if (errno == EAGAIN) {
-            dsh->status = DPIP_EAGAIN;
-            ret = -1;
-            break;
-         } else {
-            MSG_ERR("[a_Dpip_dsh_trywrite] %s\n", dStrerror(errno));
-            dsh->status = DPIP_ERROR;
-            ret = -3;
-            break;
-         }
+   do {
+      st = write(dsh->fd_out, Data, DataSize);
+   } while (st < 0 && errno == EINTR);
+   if (st < 0) {
+      if (errno == EAGAIN) {
+         dsh->status = DPIP_EAGAIN;
+         ret = -1;
+      } else {
+         MSG_ERR("[Dpip_dsh_trywrite] %s\n", dStrerror(errno));
+         dsh->status = DPIP_ERROR;
       }
-      sent += st;
-      break;
+   } else {
+      ret = st;
    }
 
    if (blocking) {
@@ -345,7 +340,42 @@ int a_Dpip_dsh_trywrite(Dsh *dsh, const char *Data, int DataSize)
       fcntl(dsh->fd_in, F_SETFL, old_flags);
    }
 
-   return (st > 0 ? sent : ret);
+   return ret;
+}
+
+/*
+ * Return value: 0 on success or empty buffer, 
+ *               1..DataSize sent, -1 eagain, or -3 on big Error
+ */
+int a_Dpip_dsh_tryflush(Dsh *dsh)
+{
+   int st;
+
+   if (dsh->wrbuf->len == 0) {
+      st = 0;
+   } else {
+      st = Dpip_dsh_trywrite(dsh, dsh->wrbuf->str, dsh->wrbuf->len);
+      if (st > 0) {
+         /* update internal buffer */
+         dStr_erase(dsh->wrbuf, 0, st);
+      }
+   }
+   return (dsh->wrbuf->len == 0) ? 0 : st;
+}
+
+/*
+ * Return value: 1..DataSize sent, -1 eagain, or -3 on big Error
+ */
+int a_Dpip_dsh_trywrite(Dsh *dsh, const char *Data, int DataSize)
+{
+   int st;
+
+   if ((st = Dpip_dsh_trywrite(dsh, Data, DataSize)) > 0) {
+      /* update internal buffer */
+      if (st < DataSize)
+         dStr_append_l(dsh->wrbuf, Data + st, DataSize - st);
+   }
+   return st;
 }
 
 /*
@@ -393,7 +423,7 @@ static void Dpip_dsh_read_nb(Dsh *dsh)
          break;
       } else {
          /* append to buf */
-         dStr_append_l(dsh->rd_dbuf, buf, st);
+         dStr_append_l(dsh->rdbuf, buf, st);
       }
    }
 
@@ -436,7 +466,7 @@ static void Dpip_dsh_read(Dsh *dsh)
          break;
       } else {
          /* append to buf */
-         dStr_append_l(dsh->rd_dbuf, buf, st);
+         dStr_append_l(dsh->rdbuf, buf, st);
          break;
       }
    }
@@ -467,27 +497,27 @@ char *a_Dpip_dsh_read_token(Dsh *dsh, int blocking)
 
    if (blocking && dsh->mode & DPIP_TAG) {
       /* Only wait for data when the tag is incomplete */
-      if (!strstr(dsh->rd_dbuf->str, DPIP_TAG_END)) {
+      if (!strstr(dsh->rdbuf->str, DPIP_TAG_END)) {
          do {
             Dpip_dsh_read(dsh);
-            p = strstr(dsh->rd_dbuf->str, DPIP_TAG_END);
+            p = strstr(dsh->rdbuf->str, DPIP_TAG_END);
          } while (!p && dsh->status == EAGAIN);
       }
    }
 
    if (dsh->mode & DPIP_TAG) {
       /* return a full tag */
-      if ((p = strstr(dsh->rd_dbuf->str, DPIP_TAG_END))) {
-         ret = dStrndup(dsh->rd_dbuf->str, p - dsh->rd_dbuf->str + 3);
-         dStr_erase(dsh->rd_dbuf, 0, p - dsh->rd_dbuf->str + 3);
+      if ((p = strstr(dsh->rdbuf->str, DPIP_TAG_END))) {
+         ret = dStrndup(dsh->rdbuf->str, p - dsh->rdbuf->str + 3);
+         dStr_erase(dsh->rdbuf, 0, p - dsh->rdbuf->str + 3);
          if (strstr(ret, DPIP_MODE_SWITCH_TAG))
             dsh->mode |= DPIP_LAST_TAG;
       }
    } else {
       /* raw mode, return what we have "as is" */
-      if (dsh->rd_dbuf->len > 0) {
-         ret = dStrndup(dsh->rd_dbuf->str, dsh->rd_dbuf->len);
-         dStr_truncate(dsh->rd_dbuf, 0);
+      if (dsh->rdbuf->len > 0) {
+         ret = dStrndup(dsh->rdbuf->str, dsh->rdbuf->len);
+         dStr_truncate(dsh->rdbuf, 0);
       }
    }
 
@@ -521,8 +551,8 @@ void a_Dpip_dsh_close(Dsh *dsh)
  */
 void a_Dpip_dsh_free(Dsh *dsh)
 {
-   dStr_free(dsh->dbuf, 1);
-   dStr_free(dsh->rd_dbuf, 1);
+   dStr_free(dsh->wrbuf, 1);
+   dStr_free(dsh->rdbuf, 1);
    dFree(dsh);
 }
 
diff --git a/dpip/dpip.h b/dpip/dpip.h
index 453b13e5..083732ff 100644
--- a/dpip/dpip.h
+++ b/dpip/dpip.h
@@ -35,8 +35,8 @@ struct _DpipSocketHandler {
    /* FILE *in;    --Unused. The stream functions block when reading. */
    FILE *out;
 
-   Dstr *dbuf;     /* write buffer */
-   Dstr *rd_dbuf;  /* read buffer */
+   Dstr *wrbuf;    /* write buffer */
+   Dstr *rdbuf;    /* read buffer */
    int flush_sz;   /* max size before flush */
 
    int mode;       /* mode flags: DPIP_TAG | DPIP_LAST_TAG | DPIP_RAW */
@@ -68,6 +68,7 @@ int a_Dpip_check_auth(const char *auth);
 Dsh *a_Dpip_dsh_new(int fd_in, int fd_out, int flush_sz);
 int a_Dpip_dsh_write(Dsh *dsh, int flush, const char *Data, int DataSize);
 int a_Dpip_dsh_write_str(Dsh *dsh, int flush, const char *str);
+int a_Dpip_dsh_tryflush(Dsh *dsh);
 int a_Dpip_dsh_trywrite(Dsh *dsh, const char *Data, int DataSize);
 char *a_Dpip_dsh_read_token(Dsh *dsh, int blocking);
 void a_Dpip_dsh_close(Dsh *dsh);