More skeleton

Signed-off-by: Laurent Bercot <ska@appnovation.com>

7 files changed, 172 insertions, 54 deletions

diff --git a/package/deps.mak b/package/deps.mak
index 81fdadc..ca15915 100644
--- a/package/deps.mak
+++ b/package/deps.mak
@@ -10,6 +10,7 @@ src/libdcache/dcache-internal.h: src/include/shibari/dcache.h
 src/cache/cache.o src/cache/cache.lo: src/cache/cache.c src/cache/shibari-cache-internal.h src/include/shibari/dcache.h
 src/cache/clientaccess.o src/cache/clientaccess.lo: src/cache/clientaccess.c src/cache/shibari-cache-internal.h
 src/cache/conf.o src/cache/conf.lo: src/cache/conf.c src/cache/shibari-cache-internal.h
+src/cache/log.o src/cache/log.lo: src/cache/log.c src/cache/shibari-cache-internal.h
 src/cache/query.o src/cache/query.lo: src/cache/query.c src/cache/shibari-cache-internal.h
 src/cache/shibari-cache.o src/cache/shibari-cache.lo: src/cache/shibari-cache.c src/cache/shibari-cache-internal.h src/include/shibari/common.h src/include/shibari/config.h
 src/cache/tcpconnection.o src/cache/tcpconnection.lo: src/cache/tcpconnection.c src/cache/shibari-cache-internal.h
@@ -53,7 +54,7 @@ src/server/shibari_tdb_find_authority.o src/server/shibari_tdb_find_authority.lo
 src/server/shibari_tdb_read_entry.o src/server/shibari_tdb_read_entry.lo: src/server/shibari_tdb_read_entry.c src/include/shibari/tdb.h
 
 shibari-cache: EXTRA_LIBS := -ls6dns -lskarnet ${SOCKET_LIB} ${SYSCLOCK_LIB}
-shibari-cache: src/cache/shibari-cache.o src/cache/cache.o src/cache/clientaccess.o src/cache/conf.o src/cache/query.o src/cache/tcpconnection.o src/cache/udpqueue.o ${LIBDCACHE} ${LIBSHIBARI_COMMON}
+shibari-cache: src/cache/shibari-cache.o src/cache/cache.o src/cache/clientaccess.o src/cache/conf.o src/cache/log.o src/cache/query.o src/cache/tcpconnection.o src/cache/udpqueue.o ${LIBDCACHE} ${LIBSHIBARI_COMMON}
 ifeq ($(strip $(STATIC_LIBS_ARE_PIC)),)
 libshibari-common.a.xyzzy: src/common/shibari_log_answer.o src/common/shibari_log_exit.o src/common/shibari_log_query.o src/common/shibari_log_queryplus.o src/common/shibari_log_start.o src/common/shibari_util_qtype_num.o src/common/shibari_util_qtype_str.o src/common/shibari_util_rcode_str.o src/common/shibari_util_canon_domain.o src/common/shibari_util_get_prefixlen.o
 else
diff --git a/src/cache/deps-exe/shibari-cache b/src/cache/deps-exe/shibari-cache
index 19a22c6..e363f19 100644
--- a/src/cache/deps-exe/shibari-cache
+++ b/src/cache/deps-exe/shibari-cache
@@ -1,6 +1,7 @@
 cache.o
 clientaccess.o
 conf.o
+log.o
 query.o
 tcpconnection.o
 udpqueue.o
diff --git a/src/cache/log.c b/src/cache/log.c
new file mode 100644
index 0000000..bc67384
--- /dev/null
+++ b/src/cache/log.c
@@ -0,0 +1,30 @@
+/* ISC license. */
+
+#include <skalibs/fmtscan.h>
+#include <skalibs/strerr.h>
+#include <skalibs/ip46.h>
+
+#include "shibari-cache-internal.h"
+
+void log_tcptimeout (uint16_t i)
+{
+  if (g->verbosity >= 3)
+  {
+  }
+}
+
+void log_newtcp4 (char const *ip, uint16_t port)
+{
+  if (g->verbosity >= 3)
+  {
+  }
+}
+
+#ifdef SKALIBS_IPV6_ENABLED
+void log_newtcp6 (char const *ip, uint16_t port)
+{
+  if (g->verbosity >= 3)
+  {
+  }
+}
+#endif
diff --git a/src/cache/query.c b/src/cache/query.c
index ad606e6..6c4bda8 100644
--- a/src/cache/query.c
+++ b/src/cache/query.c
@@ -1,13 +1,41 @@
 /* ISC license. */
 
+#include <stdint.h>
+
+#include <s6-dns/s6dns-engine.h>
+
 #include "shibari-cache-internal.h"
 
-void query_fail (query *q)
+static uint16_t query_delete (query *q)
+{
+  uint16_t newi = q->prev ;
+  QUERY(newi)->next = q->next ;
+  QUERY(q->next)->prev = q->prev ;
+  q->xindex = UINT16_MAX ;
+  return newi ;
+}
+
+uint16_t query_abort (uint16_t id)
 {
+  query *q = QUERY(id) ;
+  s6dns_engine_recycle(&q->dt) ;
+  return query_delete(q) ;
 }
 
-void query_success (query *q)
+uint16_t query_fail (uint16_t id)
 {
+  query *q = QUERY(id) ;
+
+  if (q->source == 2) tcpconnection_removequery(TCPCONNECTION(q->i), id) ;
+  return query_delete(q) ;
+}
+
+uint16_t query_succeed (uint16_t id)
+{
+  query *q = QUERY(id) ;
+
+  if (q->source == 2) tcpconnection_removequery(TCPCONNECTION(q->i), id) ;
+  return query_delete(q) ;
 }
 
 int query_new (uint8_t source, uint16_t i, char const *ip, uint16_t port, char const *s, uint16_t len)
diff --git a/src/cache/shibari-cache-internal.h b/src/cache/shibari-cache-internal.h
index cd21fd8..6c42055 100644
--- a/src/cache/shibari-cache-internal.h
+++ b/src/cache/shibari-cache-internal.h
@@ -42,6 +42,15 @@ extern int conf_get_uint64 (char const *, uint64_t *) ;
 extern char const *conf_get_string (char const *) ;
 
 
+ /* log */
+
+extern void log_newtcp4 (char const *, uint16_t) ;
+#ifdef SKALIBS_IPV6_ENABLED
+extern void log_newtcp6 (char const *, uint16_t) ;
+extern void log_tcptimeout (uint16_t) ;
+#endif
+
+
  /* query */
 
 typedef struct query_s query, *query_ref ;
@@ -51,18 +60,19 @@ struct query_s
   uint16_t prev ;
   uint16_t next ;
   uint16_t xindex ;
-  uint16_t source ;
   uint16_t i ;
   uint16_t port ;
+  uint8_t source ;
   char ip[SKALIBS_IP_SIZE] ;
 } ;
-#define QUERY_ZERO { .dt = S6DNS_ENGINE_ZERO, .prev = 0, .next = 0, .xindex = UINT16_MAX, .source = 0, .i = 0, .port = 0, .ip = { 0 } }
+#define QUERY_ZERO { .dt = S6DNS_ENGINE_ZERO, .prev = 0, .next = 0, .xindex = UINT16_MAX, .i = 0, .port = 0, .source = 0, .ip = { 0 } }
 #define nq (genset_n(&g->queries) - 1)
 #define QUERY(i) genset_p(query, &g->queries, (i))
 #define qstart (QUERY(g->qsentinel)->next)
 
-extern void query_fail (query *) ;
-extern void query_success (query *) ;
+extern uint16_t query_abort (uint16_t) ;
+extern uint16_t query_fail (uint16_t) ;
+extern uint16_t query_succeed (uint16_t) ;
 extern int query_new (uint8_t, uint16_t, char const *, uint16_t, char const *, uint16_t) ;
 
 
@@ -76,18 +86,20 @@ struct tcpconnection_s
   uint32_t instate ;
   tain rdeadline ;
   tain wdeadline ;
+  genalloc queries ;  /* uint16_t */
   uint16_t prev ;
   uint16_t next ;
   uint16_t xindex ;
 } ;
-#define TCPCONNECTION_ZERO { .out = BUFALLOC_ZERO, .in = STRALLOC_ZERO, .instate = 0, .rdeadline = TAIN_INFINITE, .wdeadline = TAIN_INFINITE, .prev = 0, .next = 0, .xindex = UINT16_MAX }
+#define TCPCONNECTION_ZERO { .out = BUFALLOC_ZERO, .in = STRALLOC_ZERO, .instate = 0, .rdeadline = TAIN_INFINITE, .wdeadline = TAIN_INFINITE, .queries = GENALLOC_ZERO, .prev = 0, .next = 0, .xindex = UINT16_MAX }
 #define ntcp (genset_n(&g->tcpconnections) - 1)
 #define TCPCONNECTION(i) genset_p(tcpconnection, &g->tcpconnections, (i))
 #define tcpstart (TCPCONNECTION(g->tcpsentinel)->next)
 
-extern void tcpconnection_drop (tcpconnection *) ;
+extern void tcpconnection_removequery (tcpconnection *, uint16_t) ;
+extern uint16_t tcpconnection_delete (tcpconnection *) ;
 extern int tcpconnection_flush (tcpconnection *) ;
-extern int tcpconnection_new (uint8_t, uint16_t, int, char const *, uint16_t) ;
+extern void tcpconnection_new (int) ;
 
 
  /* udpqueue */
diff --git a/src/cache/shibari-cache.c b/src/cache/shibari-cache.c
index f7e4256..cd0e4f4 100644
--- a/src/cache/shibari-cache.c
+++ b/src/cache/shibari-cache.c
@@ -125,7 +125,7 @@ int main (int argc, char const *const *argv)
   uint16_t n4 = 0, n6 = 0, maxtcp, maxqueries ;
   char const *ip4 = 0, *ip6 = 0 ;
   unsigned int cont = 2 ;
-  int sfd = -1 ;
+  int spfd = -1 ;
   unsigned int notif = 0 ;
   uid_t uid = 0 ;
   gid_t gid = 0 ;
@@ -170,8 +170,8 @@ int main (int argc, char const *const *argv)
   if (!cdb_init(&g->confdb, conffile)) strerr_diefu2sys(111, "open ", conffile) ;
   conf_init(conffile, &n4, &n6, &ip4, &ip6, &maxtcp, &maxqueries) ;
 
-  sfd = selfpipe_init() ;
-  if (sfd == -1) strerr_diefu1sys(111, "create selfpipe") ;
+  spfd = selfpipe_init() ;
+  if (spfd == -1) strerr_diefu1sys(111, "create selfpipe") ;
   if (!sig_altignore(SIGPIPE)) strerr_diefu1sys(111, "ignore SIGPIPE") ;
   {
     sigset_t set ;
@@ -197,18 +197,16 @@ int main (int argc, char const *const *argv)
 
     memset(udpq4, 0, n4 * sizeof(udpqueue)) ;
     memset(udpq6, 0, n6 * sizeof(udpqueue)) ;
+    memset(tcpconnection_storage, 0, (maxtcp + 1) * sizeof(tcpconnection)) ;
+    memset(query_storage, 0, (maxqueries + 1) * sizeof(query)) ;
     GENSET_init(&g->tcpconnections, tcpconnection, tcpconnection_storage, tcpconnection_freelist, maxtcp + 1) ;
     g->tcpsentinel = genset_new(&g->tcpconnections) ;
     GENSET_init(&g->queries, query, query_storage, query_freelist, maxqueries + 1) ;
     g->qsentinel = genset_new(&g->queries) ;
     {
-      static const tcpconnection tcpconnection_zero = TCPCONNECTION_ZERO ;
-      static const query query_zero = QUERY_ZERO ;
       tcpconnection *p = TCPCONNECTION(g->tcpsentinel) ;
       query *q = QUERY(g->qsentinel) ;
-      *p = tcpconnection_zero ;
       p->prev = p->next = g->tcpsentinel ;
-      *q = query_zero ;
       q->prev = q->next = g->qsentinel ;
     }
 
@@ -278,7 +276,7 @@ int main (int argc, char const *const *argv)
 
      /* preparation */
 
-      x[0].fd = sfd ;
+      x[0].fd = spfd ;
       x[0].events = IOPAUSE_READ ;
       if (cont == 1 && tain_less(&lameduckt, &deadline)) deadline = lameduckt ;
 
@@ -355,7 +353,7 @@ int main (int argc, char const *const *argv)
       }
 
 
-     /* exit condition */
+     /* normal exit condition */
 
       if (cont < 2 && !r && !nq) break ;
 
@@ -370,17 +368,17 @@ int main (int argc, char const *const *argv)
 
       if (!r)
       {
-        if (cont == 1 && !tain_future(&lameduckt)) break ;
+        if (cont == 1 && !tain_future(&lameduckt)) break ;  /* too lame */
         for (uint16_t i = qstart ; i != g->qsentinel ; i = QUERY(i)->next)
-        {
-          query *p = QUERY(i) ;
-          if (s6dns_engine_timeout_g(&p->dt)) { i = p->prev ; query_fail(p) ; }
-        }
+          if (s6dns_engine_timeout_g(&QUERY(i)->dt)) i = query_fail(i) ;
         for (uint16_t i = tcpstart ; i != g->tcpsentinel ; i = TCPCONNECTION(i)->next)
         {
           tcpconnection *p = TCPCONNECTION(i) ;
           if (!tain_future(&p->rdeadline) || !tain_future(&p->wdeadline))
-           tcpconnection_drop(p) ;
+          {
+            log_tcptimeout(i) ;
+            i = tcpconnection_delete(p) ;
+          }
         }
         for (uint16_t i = 0 ; i < n4 ; i++)
           if (!tain_future(&udpq4[i].deadline)) udpqueue_drop(udpq4 + i) ;
@@ -429,22 +427,15 @@ int main (int argc, char const *const *argv)
         {
           tcpconnection *p = TCPCONNECTION(i) ;
           if (p->xindex < UINT16_MAX && x[p->xindex].revents & IOPAUSE_WRITE)
-          {
-            if (tcpconnection_flush(p) == -1)
-            {
-              i = p->prev ;
-              tcpconnection_drop(p) ;
-            }
-          }
+            if (tcpconnection_flush(p) == -1) i = tcpconnection_delete(p) ;
         }
 
         for (uint16_t i = qstart ; i != g->qsentinel ; i = QUERY(i)->next)
         {
-          query *p = QUERY(i) ;
-          if (p->xindex == UINT16_MAX) continue ;
-          r = s6dns_engine_event_g(&p->dt) ;
-          if (r) i = p->prev ;
-          if (r == -1) query_fail(p) ; else query_success(p) ;
+          if (QUERY(i)->xindex == UINT16_MAX) continue ;
+          r = s6dns_engine_event_g(&QUERY(i)->dt) ;
+          if (r < 0) i = query_fail(i) ;
+          else if (r > 0) i = query_succeed(i) ;
         }
 
         for (uint16_t i = 0 ; i < n4 ; i++)
@@ -524,20 +515,24 @@ int main (int argc, char const *const *argv)
           tcpconnection *p = TCPCONNECTION(i) ;
           if (p->xindex < UINT16_MAX && x[p->xindex].revents & IOPAUSE_READ)
           {
-            int l = sanitize_read(mininetstring_read(bufalloc_fd(&p->out), &p->in, &p->instate)) ;
-            if (l == -1) { i = p->prev ; tcpconnection_drop(p) ; }
-            if (l <= 0) continue ;
-            if (p->in.len < 12 || p->in.len > 65536) { i = p->prev ; tcpconnection_drop(p) ; continue ; }
-            if (!query_new(2, i, 0, 0, p->in.s, p->in.len))
+            uint16_t n = MAXSAME ;
+            while (n-- && nq < maxqueries)
             {
-              if (g->verbosity)
+              int l = sanitize_read(mininetstring_read(bufalloc_fd(&p->out), &p->in, &p->instate)) ;
+              if (l == -1) { i = tcpconnection_delete(p) ; break ; }
+              if (!l) break ;
+              if (p->in.len < 12 || p->in.len > 65536) { i = tcpconnection_delete(p) ; break ; }
+              if (!query_new(2, i, 0, 0, p->in.s, p->in.len))
               {
-                char fmt[UINT16_FMT] ;
-                fmt[uint16_fmt(fmt, i)] = 0 ;
-                strerr_warnwu2sys("process TCP query on connection ", fmt) ;
+                if (g->verbosity)
+                {
+                  char fmt[UINT16_FMT] ;
+                  fmt[uint16_fmt(fmt, i)] = 0 ;
+                  strerr_warnwu2sys("process TCP query on connection ", fmt) ;
+                }
               }
+              p->in.len = 0 ;
             }
-            p->in.len = 0 ;
           }
         }
 
@@ -557,7 +552,8 @@ int main (int argc, char const *const *argv)
                 strerr_diefu1sys(111, "create new TCP connection") ;
               }
               if (!clientaccess_ip4(ip)) { close(fd) ; continue ; }
-              tcpconnection_new(0, i, fd, ip, port) ;
+              tcpconnection_new(fd) ;
+              log_newtcp4(ip, port) ;
             }
           }
         }
@@ -579,7 +575,8 @@ int main (int argc, char const *const *argv)
                 strerr_diefu1sys(111, "create new TCP connection") ;
               }
               if (!clientaccess_ip6(ip)) { close(fd) ; continue ; }
-              tcpconnection_new(1, i, fd, ip, port) ;
+              tcpconnection_new(fd) ;
+              log_newtcp6(ip, port) ;
             }
           }
         }
diff --git a/src/cache/tcpconnection.c b/src/cache/tcpconnection.c
index bc77781..71a4299 100644
--- a/src/cache/tcpconnection.c
+++ b/src/cache/tcpconnection.c
@@ -1,17 +1,66 @@
 /* ISC license. */
 
+#include <stdint.h>
+#include <errno.h>
+
+#include <skalibs/allreadwrite.h>
+#include <skalibs/error.h>
+#include <skalibs/tai.h>
+#include <skalibs/djbunix.h>
+#include <skalibs/bufalloc.h>
+#include <skalibs/genalloc.h>
+
 #include "shibari-cache-internal.h"
 
-void tcpconnection_drop (tcpconnection *tc)
+void tcpconnection_removequery (tcpconnection *p, uint16_t id)
+{
+  uint16_t *tab = genalloc_s(uint16_t, &p->queries) ;
+  uint16_t n = genalloc_len(uint16_t, &p->queries) ;
+  uint16_t i = 0 ;
+  for (; i < n ; i++) if (id == tab[i]) break ;
+  if (i >= n) return ;
+  tab[i] = tab[--n] ;
+  genalloc_setlen(uint16_t, &p->queries, n) ;
+}
+
+uint16_t tcpconnection_delete (tcpconnection *p)
+{
+  uint16_t newi = p->prev ;
+  p->out.x.len = 0 ;
+  p->in.len = 0 ;
+  p->instate = 0 ;
+  fd_close(p->out.fd) ;
+  for (uint16_t i = 0 ; i < genalloc_len(uint16_t, &p->queries) ; i++)
+    query_abort(genalloc_s(uint16_t, &p->queries)[i]) ;
+  genalloc_setlen(uint16_t, &p->queries, 0) ;
+  TCPCONNECTION(newi)->next = p->next ;
+  TCPCONNECTION(p->next)->prev = p->prev ;
+  p->xindex = UINT16_MAX ;
+  return newi ;
+}
+
+int tcpconnection_flush (tcpconnection *p)
 {
+  return bufalloc_flush(&p->out) ? 1 :
+    error_isagain(errno) ? 0 : -1 ;
 }
 
-int tcpconnection_flush (tcpconnection *tc)
+static void tcpconnection_init (tcpconnection *p, int fd)
 {
-  return 1 ;
+  if (!p->out.op) bufalloc_init(&p->out, &fd_write, fd) ;
+  else { p->out.fd = fd ; p->out.x.len = 0 ; }
+  tain_add_g(&p->rdeadline, &tain_infinite_relative) ;
+  tain_add_g(&p->wdeadline, &tain_infinite_relative) ;
 }
 
-int tcpconnection_new (uint8_t source, uint16_t i, int fd, char const *ip, uint16_t port)
+void tcpconnection_new (int fd)
 {
-  return 1 ;
+  uint16_t n = genset_new(&g->tcpconnections) ;
+  tcpconnection *sentinel = TCPCONNECTION(g->tcpsentinel) ;
+  tcpconnection *p = TCPCONNECTION(n) ;
+  tcpconnection_init(p, fd) ;
+  p->prev = g->tcpsentinel ;
+  p->next = sentinel->next ;
+  TCPCONNECTION(sentinel->next)->prev = n ;
+  sentinel->next = n ;
 }