Add realtime and norealtime options, cork/uncork responses

9 files changed, 115 insertions, 34 deletions

diff --git a/doc/tipidee.conf.html b/doc/tipidee.conf.html
index f7753da..581b05f 100644
--- a/doc/tipidee.conf.html
+++ b/doc/tipidee.conf.html
@@ -786,6 +786,43 @@ data into records; in that case, autochunk is not needed and only brings
 a small performance penalty. </li>
 </ul>
 
+<div id="realtime">
+<h4> <tt>realtime</tt> </h4>
+</div>
+
+<p>
+ <code> realtime <em>directory</em> </code> <br>
+ <code> realtime <em>file</em> </code>
+</p>
+
+<ul>
+ <li> This directive is useful if <em>file</em> is actually a resource providing
+a large amount of content that you want to serve to the client with as little
+latency as possible, even if it means sacrificing a little throughput. If the
+system's TCP stack supports the functionality, it will minimize the delays in the
+stack so the data is sent to the client as soon as it becomes available, without
+trying to optimizing sending buffers or anything. </li>
+ <li> It is generally not recommended; you should leave that option disabled
+(which is the default). But if you have a resource that benefits from real-time
+transmission (e.g. a streaming service) then you might want to set the option for
+this resource. </li>
+</ul>
+
+<div id="norealtime">
+<h4> <tt>norealtime</tt> </h4>
+</div>
+
+<p>
+ <code> norealtime <em>directory</em> </code> <br>
+ <code> norealtime <em>file</em> </code>
+</p>
+
+<ul>
+ <li> This is the opposite, saying that content streamed from resources under
+<em>directory</em>, or specific script <em>file</em>, will <em>not</em>
+manipulate the TCP stack to send the content in real time. </li>
+</ul>
+
 <div id="file-type">
 <h4> <tt>file-type</tt> </h4>
 </div>
diff --git a/src/config/lexparse.c b/src/config/lexparse.c
index e87c501..b8d8a5d 100644
--- a/src/config/lexparse.c
+++ b/src/config/lexparse.c
@@ -55,6 +55,8 @@ enum directivevalue_e
   T_NOAUTH,
   T_AUTOCHUNK,
   T_NOAUTOCHUNK,
+  T_REALTIME,
+  T_NOREALTIME,
   T_FILETYPE,
   T_CUSTOMRESPONSE
 } ;
@@ -365,7 +367,7 @@ static inline void parse_rproxy (char const *s, size_t const *word, size_t n, ch
 
 static void parse_bitattr (char const *s, size_t const *word, size_t n, char const *domain, size_t domainlen, mdt const *md, uint8_t bit, int h)
 {
-  static char const *const attr[3][2] = { { "noncgi", "cgi" }, { "nonnph", "nph", }, { "noauth", "basic-auth" } } ;
+  static char const *const attr[4][2] = { { "noncgi", "cgi" }, { "nonnph", "nph", }, { "noauth", "basic-auth" }, { "norealtime", "realtime" } } ;
   if (n != 1)
     strerr_dief8x(1, "too ", n > 1 ? "many" : "few", " arguments to directive ", attr[bit][h], " in file ", g.storage.s + md->filepos, " line ", md->linefmt) ;
   if (!domain)
@@ -505,9 +507,11 @@ static inline void process_line (char const *s, size_t const *word, size_t n, st
     { .name = "noautochunk", .value = T_NOAUTOCHUNK },
     { .name = "noncgi", .value = T_NONCGI },
     { .name = "nonnph", .value = T_NONNPH },
+    { .name = "norealtime", .value = T_NOREALTIME },
     { .name = "noredirect", .value = T_NOREDIRECT },
     { .name = "nph", .value = T_NPH },
     { .name = "nph-prefix", .value = T_NPHPREFIX },
+    { .name = "realtime", .value = T_REALTIME },
     { .name = "redirect", .value = T_REDIRECT },
     { .name = "rproxy", .value = T_RPROXY },
   } ;
@@ -606,6 +610,12 @@ static inline void process_line (char const *s, size_t const *word, size_t n, st
     case T_NOAUTOCHUNK :
       parse_bitattr(s, word, n, domain->s, domain->len, md, 3, 0) ;
       break ;
+    case T_REALTIME :
+      parse_bitattr(s, word, n, domain->s, domain->len, md, 4, 1) ;
+      break ;
+    case T_NOREALTIME :
+      parse_bitattr(s, word, n, domain->s, domain->len, md, 4, 0) ;
+      break ;
     case T_FILETYPE :
       parse_filetype(s, word, n, domain->s, domain->len, md) ;
       break ;
diff --git a/src/include/tipidee/resattr.h b/src/include/tipidee/resattr.h
index da4fbc7..c5f01af 100644
--- a/src/include/tipidee/resattr.h
+++ b/src/include/tipidee/resattr.h
@@ -11,8 +11,9 @@
 #define TIPIDEE_RA_FLAG_NPH 0x0002
 #define TIPIDEE_RA_FLAG_BA 0x0004
 #define TIPIDEE_RA_FLAG_AUTOCHUNK 0x0008
+#define TIPIDEE_RA_FLAG_REALTIME 0x0010
 
-#define TIPIDEE_RA_BITS 4
+#define TIPIDEE_RA_BITS 5
 
 typedef struct tipidee_resattr_s tipidee_resattr, *tipidee_resattr_ref ;
 struct tipidee_resattr_s
diff --git a/src/tipideed/cgi.c b/src/tipideed/cgi.c
index 63f43e7..d5daf56 100644
--- a/src/tipideed/cgi.c
+++ b/src/tipideed/cgi.c
@@ -170,7 +170,7 @@ static inline int local_redirect (tipidee_rql *rql, char const *docroot, char co
   return 1 ;
 }
 
-static inline int do_cgi (tipidee_rql *rql, char const *docroot, char const *const *argv, char const *const *envp, char const *body, size_t bodylen, char *uribuf, int autochunk)
+static inline int do_cgi (tipidee_rql *rql, char const *docroot, char const *const *argv, char const *const *envp, char const *body, size_t bodylen, char *uribuf, int autochunk, uint32_t raflags)
 {
   char const *reason = "OK" ;
   char const *location ;
@@ -365,11 +365,17 @@ static inline int do_cgi (tipidee_rql *rql, char const *docroot, char const *con
       buffer_rseek(&b, len) ;
       rbodylen -= len ;
     }
+    if (!(raflags & TIPIDEE_RA_FLAG_REALTIME)) cork(1) ;
     if (!buffer_timed_flush_g(buffer_1, &deadline))
       strerr_diefu1sys(111, "write to stdout") ;
-    if (rbodylen) stream_fixed(x[0].fd, rbodylen, argv[0]) ;
+    if (rbodylen) stream_fixed(x[0].fd, rbodylen, argv[0], raflags) ;
+    if (!rbodylen) uncork(1) ;  // stream_fixed() auto-uncorks
+  }
+  else if (autochunk)
+  {
+    uncork(1) ;
+    stream_autochunk(&b, argv[0]) ;
   }
-  else if (autochunk) stream_autochunk(&b, argv[0]) ;
   else
   {
     size_t len ;
@@ -379,9 +385,11 @@ static inline int do_cgi (tipidee_rql *rql, char const *docroot, char const *con
     if (buffer_timed_putv_g(buffer_1, v, 2, &deadline) < len)
       strerr_diefu1sys(111, "write to stdout") ;
     buffer_rseek(&b, len) ;
+    if (!(raflags & TIPIDEE_RA_FLAG_REALTIME)) cork(1) ;
     if (!buffer_timed_flush_g(buffer_1, &deadline))
       strerr_diefu1sys(111, "write to stdout") ;
-    stream_infinite(x[0].fd, argv[0]) ;
+    stream_infinite(x[0].fd, argv[0], raflags) ;
+    if (!(raflags & TIPIDEE_RA_FLAG_REALTIME)) uncork(1) ;
   }
 
   close(x[0].fd) ;
@@ -400,5 +408,5 @@ int respond_cgi (tipidee_rql *rql, char const *docroot, char const *fn, size_t d
   g.sa.len = sabase ;
   return ra->flags & TIPIDEE_RA_FLAG_NPH ?
     do_nph(rql, docroot, argv, envp, body, bodylen) :
-    do_cgi(rql, docroot, argv, envp, body, bodylen, uribuf, rql->http_minor && ra->flags & TIPIDEE_RA_FLAG_AUTOCHUNK) ;
+    do_cgi(rql, docroot, argv, envp, body, bodylen, uribuf, rql->http_minor && ra->flags & TIPIDEE_RA_FLAG_AUTOCHUNK, ra->flags) ;
 }
diff --git a/src/tipideed/errors.c b/src/tipideed/errors.c
index edde934..c797701 100644
--- a/src/tipideed/errors.c
+++ b/src/tipideed/errors.c
@@ -95,7 +95,7 @@ void response_error_plus (tipidee_rql const *rql, char const *docroot, unsigned
           tipidee_response_header_write(buffer_1, plus, plusn) ;
           tipidee_response_header_end(buffer_1) ;
           tipidee_log_answer(g.logv, rql, status, st.st_size) ;
-          send_file(fd, st.st_size, g.sa.s + pos) ;
+          send_file(fd, st.st_size, g.sa.s + pos, 0) ;
           fd_close(fd) ;
           return ;
         }
diff --git a/src/tipideed/regular.c b/src/tipideed/regular.c
index 3d82750..0514c8c 100644
--- a/src/tipideed/regular.c
+++ b/src/tipideed/regular.c
@@ -44,7 +44,7 @@ int respond_regular (tipidee_rql const *rql, char const *docroot, char const *fn
     tipidee_response_file_G(buffer_1, rql, 200, "OK", st, st->st_size, ra->content_type, g.rhdr, g.rhdrn, 2 | !g.cont) ;
     tipidee_response_header_end(buffer_1) ;
     tipidee_log_answer(g.logv, rql, 200, st->st_size) ;
-    send_file(fd, st->st_size, fn) ;
+    send_file(fd, st->st_size, fn, ra->flags) ;
     fd_close(fd) ;
   }
   return 0 ;
@@ -83,7 +83,7 @@ int respond_partial (tipidee_rql const *rql, char const *docroot, char const *fn
   tipidee_response_partial_G(buffer_1, rql, st, start, len, ra->content_type, g.rhdr, g.rhdrn, 2 | !g.cont) ;
   tipidee_response_header_end(buffer_1) ;
   tipidee_log_answer(g.logv, rql, 206, len) ;
-  send_file_range(fd, start, len, fn) ;
+  send_file_range(fd, start, len, fn, ra->flags) ;
   fd_close(fd) ;
   return 0 ;
 }
diff --git a/src/tipideed/send_file.c b/src/tipideed/send_file.c
index 6e55c7b..f9fbb9c 100644
--- a/src/tipideed/send_file.c
+++ b/src/tipideed/send_file.c
@@ -50,7 +50,7 @@ static int s_flush (void *p)
   return 1 ;
 }
 
-void send_file_range (int fd, uint64_t offset, uint64_t n, char const *fn)
+void send_file_range (int fd, uint64_t offset, uint64_t n, char const *fn, uint32_t flags)
 {
   struct sendfile_s sf = { .pos = offset, .n = n, .fd = fd } ;
   tain deadline ;
@@ -59,6 +59,7 @@ void send_file_range (int fd, uint64_t offset, uint64_t n, char const *fn)
     strerr_diefu2sys(111, "write", " to stdout") ;
   if (!timed_flush_g(&sf, &s_getfd, &s_isnonempty, &s_flush, &deadline))
     strerr_diefu3sys(111, "sendfile ", fn, " to stdout") ;
+  (void)flags ;
 }
 
 #else
@@ -88,6 +89,7 @@ struct spliceinfo_s
 {
   ssize_t n ;
   uint32_t last : 1 ;
+  uint32_t realtime : 1 ;
 } ;
 
 static int getfd (void *b)
@@ -107,7 +109,7 @@ static int flush (void *b)
   struct spliceinfo_s *si = b ;
   while (si->n)
   {
-    ssize_t r = splice(g.p[0], 0, 1, 0, si->n, SPLICE_F_NONBLOCK | (si->last ? 0 : SPLICE_F_MORE)) ;
+    ssize_t r = splice(g.p[0], 0, 1, 0, si->n, SPLICE_F_NONBLOCK | (si->last || si->realtime ? 0 : SPLICE_F_MORE)) ;
     if (r == -1) return 0 ;
     if (!r) return 1 ;
     si->n -= r ;
@@ -115,10 +117,10 @@ static int flush (void *b)
   return 1 ;
 }
 
-void send_file_range (int fd, uint64_t offset, uint64_t n, char const *fn)
+void send_file_range (int fd, uint64_t offset, uint64_t n, char const *fn, uint32_t flags)
 {
   tain deadline ;
-  struct spliceinfo_s si = { .last = 0 } ;
+  struct spliceinfo_s si = { .last = 0, .realtime = !!(flags & TIPIDEE_RA_FLAG_REALTIME) } ;
   tain_add_g(&deadline, &g.writetto) ;
   if (!buffer_timed_flush_g(buffer_1, &deadline))
     strerr_diefu2sys(111, "write", " to stdout") ;
@@ -159,7 +161,7 @@ void init_splice_pipe (void)
 {
 }
 
-void send_file_range (int fd, uint64_t offset, uint64_t n, char const *fn)
+void send_file_range (int fd, uint64_t offset, uint64_t n, char const *fn, uint32_t flags)
 {
   tain deadline ;
   struct iovec v[2] ;
@@ -184,6 +186,7 @@ void send_file_range (int fd, uint64_t offset, uint64_t n, char const *fn)
   if (!buffer_timed_flush_g(buffer_1, &deadline))
     strerr_diefu1sys(111, "write to stdout") ;
   if (n) goto fillit ;
+  (void)flags ;
 }
 
 #endif
diff --git a/src/tipideed/stream.c b/src/tipideed/stream.c
index a5f78cc..917b0e0 100644
--- a/src/tipideed/stream.c
+++ b/src/tipideed/stream.c
@@ -27,15 +27,9 @@ struct fixed_info_s
 {
   int fd ;
   uint64_t n ;
+  uint32_t realtime : 1 ;
 } ;
 
-static inline void uncork (int fd)
-{
-  static int const val = 0 ;
-  if (setsockopt(fd, SOL_TCP, TCP_CORK, &val, sizeof(int)) == -1 && g.logv)
-    strerr_warnwu1sys("uncork stdout") ;
-}
-
 static int fixed_getfd (void *b)
 {
   struct fixed_info_s *si = b ;
@@ -47,7 +41,7 @@ static ssize_t fixed_get (void *b)
   struct fixed_info_s *si = b ;
   while (si->n)
   {
-    ssize_t r = sanitize_read(splice(si->fd, 0, 1, 0, si->n > SSIZE_MAX ? SSIZE_MAX : si->n, SPLICE_F_NONBLOCK)) ;
+    ssize_t r = sanitize_read(splice(si->fd, 0, 1, 0, si->n > SSIZE_MAX ? SSIZE_MAX : si->n, SPLICE_F_NONBLOCK | (!si->realtime && si->n > 65536 ? SPLICE_F_MORE : 0))) ;
     if (r == -1) break ;
     if (!r) return 0 ;
     si->n -= r ;
@@ -55,16 +49,30 @@ static ssize_t fixed_get (void *b)
   return si->n ? -1 : 1 ;
 }
 
-void stream_fixed (int fd, uint64_t n, char const *fn)
+void cork (int fd)
+{
+  static int const val = 1 ;
+  if (setsockopt(fd, SOL_TCP, TCP_CORK, &val, sizeof(int)) == -1 && g.logv)
+    strerr_warnwu1sys("uncork stdout") ;
+}
+
+void uncork (int fd)
+{
+  static int const val = 0 ;
+  if (setsockopt(fd, SOL_TCP, TCP_CORK, &val, sizeof(int)) == -1 && g.logv)
+    strerr_warnwu1sys("uncork stdout") ;
+}
+
+void stream_fixed (int fd, uint64_t n, char const *fn, uint32_t flags)
 {
   tain deadline ;
-  struct fixed_info_s si = { .fd = fd, .n = n } ;
+  struct fixed_info_s si = { .fd = fd, .n = n, .realtime = !!(flags & TIPIDEE_RA_FLAG_REALTIME) } ;
   tain_add_g(&deadline, tain_less(&g.writetto, &g.cgitto) ? &g.cgitto : &g.writetto) ;
   if (timed_get_g(&si, &fixed_getfd, &fixed_get, &deadline) < 0)
     strerr_diefu3sys(111, "splice from ", fn, " to stdout") ;
 }
 
-void stream_infinite (int fd, char const *fn)
+void stream_infinite (int fd, char const *fn, uint32_t flags)
 {
   ssize_t r ;
 
@@ -74,12 +82,11 @@ void stream_infinite (int fd, char const *fn)
    /* XXX: ignores timeouts, but this is just TOO GOOD to pass up */
    /* no really, it is just optimal in 99.9% of cases, it is WORTH IT */
 
-    while ((r = splice(fd, 0, 1, 0, 65536, SPLICE_F_MORE)) > 0) ;
+    while ((r = splice(fd, 0, 1, 0, 65536, flags & TIPIDEE_RA_FLAG_REALTIME ? 0 : SPLICE_F_MORE)) > 0) ;
 
    /* You WISH you had written that line of code */
 
   if (r == -1) strerr_diefu3sys(111, "splice from ", fn, " to stdout") ;
-  uncork(1) ;
   if (ndelay_on(1) == -1) strerr_diefu1sys(111, "set stdout nonblocking again") ;
 }
 
@@ -94,12 +101,23 @@ void stream_infinite (int fd, char const *fn)
 #include <skalibs/buffer.h>
 #include <skalibs/strerr.h>
 #include <skalibs/siovec.h>
+#include <skalibs/socket.h>
 #include <skalibs/tai.h>
 #include <skalibs/unix-timed.h>
 
 #include "tipideed-internal.h"
 
-void stream_fixed (int fd, uint64_t n, char const *fn)
+void cork (int fd)
+{
+  socket_tcpdelay(fd) ;
+}
+
+void uncork (int fd)
+{
+  socket_tcpnodelay(fd) ;
+}
+
+void stream_fixed (int fd, uint64_t n, char const *fn, uint32_t flags)
 {
   tain deadline ;
   struct iovec v[2] ;
@@ -118,9 +136,10 @@ void stream_fixed (int fd, uint64_t n, char const *fn)
     if (!buffer_timed_flush_g(buffer_1, &deadline))
       strerr_diefu1sys(111, "write to stdout") ;
   }
+  (void)flags ;
 }
 
-void stream_infinite (int fd, char const *fn)
+void stream_infinite (int fd, char const *fn, uint32_t flags)
 {
   tain deadline ;
   struct iovec v[2] ;
@@ -137,6 +156,7 @@ void stream_infinite (int fd, char const *fn)
     if (!buffer_timed_flush_g(buffer_1, &deadline))
       strerr_diefu1sys(111, "write to stdout") ;
   }
+  (void)flags ;
 }
 
 #endif
diff --git a/src/tipideed/tipideed-internal.h b/src/tipideed/tipideed-internal.h
index d3a66f6..e3e014c 100644
--- a/src/tipideed/tipideed-internal.h
+++ b/src/tipideed/tipideed-internal.h
@@ -142,15 +142,17 @@ extern int respond_options (tipidee_rql const *, uint32_t) ;
  /* send_file */
 
 extern void init_splice_pipe (void) ;
-extern void send_file_range (int, uint64_t, uint64_t, char const *) ;
-#define send_file(fd, n, fn) send_file_range(fd, 0, n, fn)
+extern void send_file_range (int, uint64_t, uint64_t, char const *, uint32_t) ;
+#define send_file(fd, n, fn, flags) send_file_range(fd, 0, n, fn, flags)
 
 
  /* stream */
 
-extern void stream_fixed (int, uint64_t, char const *) ;
+extern void cork (int) ;
+extern void uncork (int) ;
+extern void stream_fixed (int, uint64_t, char const *, uint32_t) ;
 extern void stream_autochunk (buffer *, char const *) ;
-extern void stream_infinite (int, char const *) ;
+extern void stream_infinite (int, char const *, uint32_t) ;
 
 
  /* regular */