Merge branch 'sendbuf'

Conflicts:
	README.txt

6 files changed, 328 insertions, 75 deletions

diff --git a/README.txt b/README.txt
index f110667..0f969cf 100644
--- a/README.txt
+++ b/README.txt
@@ -14,6 +14,10 @@ gg_trackproc - glouglou probe client in C that track process activity
 TODO
 ====
 
-libglouglou: packet concat: sendbuf
+libglouglou: put public send functions after stop/disconnect
+
+libglouglou: use BSD queue.h, in $LIBDIR/libglouglou/contrib/
 
 libglouglou + glougloud + gg_trackproc: droppriv and chroot
+
+libglouglou: function handlers per packet: _encode, _decode, _getsize
diff --git a/libglouglou/Makefile b/libglouglou/Makefile
index 7aab71a..b47507f 100644
--- a/libglouglou/Makefile
+++ b/libglouglou/Makefile
@@ -6,7 +6,7 @@ LIBDIR=$(PREFIX)/lib
 LIBNAME=libglouglou
 
 TARGET  = ${LIBNAME}.so
-SOURCES = libglouglou.c
+SOURCES = libglouglou.c sendbuf.c
 HEADERS = libglouglou.h
 OBJECTS = $(SOURCES:.c=.o)
 
diff --git a/libglouglou/libglouglou.c b/libglouglou/libglouglou.c
index c2c284a..5a08e07 100644
--- a/libglouglou/libglouglou.c
+++ b/libglouglou/libglouglou.c
@@ -13,6 +13,7 @@
 #include <signal.h>
 
 #include "libglouglou.h"
+#include "sendbuf.h"
 
 #define error(fmt, ...) \
   if (_verbosity >= 0) \
@@ -24,18 +25,19 @@
   if (_verbosity >= 2) \
     printf("libgg: %s: XXX: " fmt "\n", __func__, ##__VA_ARGS__)
 
-int               server_send(struct gg_server *, struct gg_user *,
-                              void *, int);
 void              cb_srv_receive(evutil_socket_t, short, void *);
-int               user_send(struct gg_user *, void *, int);
 struct gg_user   *user_add(struct gg_server *, struct sockaddr_in *);
 void              user_del(struct gg_server *, struct gg_user *);
 struct gg_user *  user_find(struct gg_server *, struct sockaddr_in *);
+int               user_send(struct gg_user *, void *, int);
+int               cb_usr_send(void *, int, void *);
 int               client_send(struct gg_client *, void *, int);
 void              cb_cli_receive(evutil_socket_t, short, void *);
 void              cb_cli_timer(evutil_socket_t, short, void *);
+int               cb_cli_send(void *, int, void *);
 struct gg_packet *pkt_decode(char **, int *);
-struct gg_packet *pkt_encode(struct gg_packet *, int *);
+int               pkt_getsize(struct gg_packet *);
+int               pkt_encode(struct gg_packet *, struct gg_packet *);
 
 int _verbosity = 0;
 
@@ -68,6 +70,7 @@ gg_server_start(struct event_base *ev_base, char *ip, int port,
   srv->handle_conn = handle_conn;
   srv->handle_packet = handle_packet;
   srv->usrdata = usrdata;
+
   s = socket(AF_INET, SOCK_DGRAM, 0);
   if (s < 0)
     goto err;
@@ -102,21 +105,41 @@ int
 gg_server_send(struct gg_server *srv, struct gg_packet *pkt, struct gg_user *usr)
 {
   static struct gg_packet *newpkt;
+  static struct gg_packet pktbuf;
+  struct gg_user *u;
   int size;
+  int res = 0;
+
+  /* optimisation to use sendbuf_gettoken */
+  if (!usr && srv->user_count == 1)
+    usr = LIST_FIRST(&srv->user_list);
+
+  if (usr) {
+    size = pkt_getsize(pkt);
+    newpkt = sendbuf_gettoken(usr->sbuf, size);
+    if (!newpkt)
+      return -1;
+    return pkt_encode(pkt, newpkt);
+  } else {
+    size = pkt_encode(pkt, &pktbuf);
+    LIST_FOREACH(u, &srv->user_list, entry) {
+      res = res + sendbuf_append(u->sbuf, &pktbuf, size);
+    }
+  }
 
-  newpkt = pkt_encode(pkt, &size);
-  if (!newpkt)
-    return -1;
-  return server_send(srv, usr, newpkt, size);
+  return res;
 }
 
 void
 gg_server_stop(struct gg_server *srv)
 {
-  if (srv->sock) {
-    server_send(srv, NULL, "", 0);
-    close(srv->sock);
+  struct gg_user *usr;
+
+  while ((usr = LIST_FIRST(&srv->user_list))) {
+    user_del(srv, usr);
   }
+  if (srv->sock)
+    close(srv->sock);
   if (srv->ev)
     event_del(srv->ev);
   free(srv);
@@ -126,19 +149,6 @@ gg_server_stop(struct gg_server *srv)
  * Server - private
  */
 
-int
-server_send(struct gg_server *srv, struct gg_user *usr, void *data, int size)
-{
-  struct gg_user *u;
-
-  if (usr)
-    return user_send(usr, data, size);
-  else
-    LIST_FOREACH(u, &srv->user_list, entry)
-      user_send(u, data, size); /* XXX return error if at least one error ? */
-  return 0;
-}
-
 void cb_srv_receive(evutil_socket_t fd, short what, void *arg)
 {
   struct gg_server *srv;
@@ -189,23 +199,39 @@ struct gg_user *
 user_add(struct gg_server *srv, struct sockaddr_in *remote)
 {
   struct gg_user *usr;
+  struct sendbuf *sbuf;
 
   usr = xcalloc(1, sizeof(struct gg_user));
   usr->id = srv->user_id_count;
   srv->user_id_count++;
+  srv->user_count++;
   usr->sock = srv->sock;
   addrcpy(&usr->addr, remote);
+
+  sbuf = sendbuf_new(srv->ev_base, PACKET_SNDBUF_MAX, 200, cb_usr_send, usr);
+  if (!sbuf)
+    goto err;
+  usr->sbuf = sbuf;
+
   LIST_INSERT_HEAD(&srv->user_list, usr, entry);
   verbose("Add user %d !", usr->id);
 
   return usr;
+
+err:
+  user_del(srv, usr);
+  return NULL;
 }
 
 void
 user_del(struct gg_server *srv, struct gg_user *usr)
 {
   verbose("Del user %d !", usr->id);
+  if (usr->sbuf)
+    sendbuf_free(usr->sbuf);
+  user_send(usr, "", 0);
   LIST_REMOVE(usr, entry);
+  srv->user_count--;
   free(usr);
 }
 
@@ -227,12 +253,22 @@ user_find(struct gg_server *srv, struct sockaddr_in *remote)
 int
 user_send(struct gg_user *usr, void *data, int size)
 {
-  if (sendto(usr->sock, data, size, 0,
-      (struct sockaddr *)&usr->addr, sizeof(struct sockaddr_in)) == -1) {
+  int sent;
+
+  sent = sendto(usr->sock, data, size, 0, (struct sockaddr *)&usr->addr,
+                sizeof(struct sockaddr_in));
+  if (sent == -1)
     error("failed: %s", strerror(errno));
-    return -1;
-  }
-  return 0;
+  return sent;
+}
+
+int
+cb_usr_send(void *data, int size, void *usrdata)
+{
+  struct gg_user *usr;
+
+  usr = usrdata;
+  return user_send(usr, data, size);
 }
 
 /* 
@@ -253,6 +289,7 @@ gg_client_connect(struct event_base *ev_base, char *ip, int port,
 {
   struct gg_client *cli;
   struct sockaddr_in sock_addr;
+  struct sendbuf *sbuf;
   struct event *ev;
   struct timeval tv;
   int s;
@@ -290,6 +327,11 @@ gg_client_connect(struct event_base *ev_base, char *ip, int port,
   tv.tv_sec = 0;
   evtimer_add(ev, &tv);
 
+  sbuf = sendbuf_new(cli->ev_base, PACKET_SNDBUF_MAX, 200, cb_cli_send, cli);
+  if (!sbuf)
+    goto err;
+  cli->sbuf = sbuf;
+
   return cli;
 
 err:
@@ -304,15 +346,18 @@ gg_client_send(struct gg_client *cli, struct gg_packet *pkt)
   static struct gg_packet *newpkt;
   int size;
 
-  newpkt = pkt_encode(pkt, &size);
+  size = pkt_getsize(pkt);
+  newpkt = sendbuf_gettoken(cli->sbuf, size);
   if (!newpkt)
     return -1;
-  return client_send(cli, newpkt, size);
+  return pkt_encode(pkt, newpkt);
 }
 
 void
 gg_client_disconnect(struct gg_client *cli)
 {
+  if (cli->sbuf)
+    sendbuf_free(cli->sbuf);
   if (cli->sock) {
     client_send(cli, "", 0);
     close(cli->sock);
@@ -331,12 +376,13 @@ gg_client_disconnect(struct gg_client *cli)
 int
 client_send(struct gg_client *cli, void *data, int size)
 {
-  if (sendto(cli->sock, data, size, 0,
-      (struct sockaddr *)&cli->addr, sizeof(struct sockaddr_in)) == -1) {
+  int sent;
+
+  sent = sendto(cli->sock, data, size, 0, (struct sockaddr *)&cli->addr,
+                sizeof(struct sockaddr_in));
+  if (sent == -1)
     error("failed: %s", strerror(errno));
-    return -1;
-  }
-  return 0;
+  return sent;
 }
 
 void cb_cli_receive(evutil_socket_t fd, short what, void *arg)
@@ -412,6 +458,15 @@ void cb_cli_timer(evutil_socket_t fd, short what, void *arg)
   event_add(cli->ev_timer, &tv);
 }
 
+int
+cb_cli_send(void *data, int size, void *usrdata)
+{
+  struct gg_client *cli;
+
+  cli = usrdata;
+  return client_send(cli, data, size);
+}
+
 /*
  * Packets - private
  */
@@ -519,72 +574,99 @@ invalid:
  * encodes a packet and returns the size before sending it on the wire.
  * it assumes that the packet contains correct content.
  */
-struct gg_packet *
-pkt_encode(struct gg_packet *pkt, int *len)
+int
+pkt_encode(struct gg_packet *pkt, struct gg_packet *newpkt)
 {
-  static struct gg_packet newpkt;
-  int packet_len;
-
   if (pkt->type < PACKET_TYPE_MIN || pkt->type > PACKET_TYPE_MAX)
     invalid("type");
-  packet_len = gg_packet_props[pkt->type].size; // XXX never overflow ?
 
-  newpkt.ver = pkt->ver;
-  newpkt.type = pkt->type;
+  newpkt->ver = pkt->ver;
+  newpkt->type = pkt->type;
   switch(pkt->type) {
     case PACKET_NEWCONN:
-      newpkt.newconn_id = htons(pkt->newconn_id);
-      newpkt.newconn_src = htonl(pkt->newconn_src);
-      newpkt.newconn_dst = htonl(pkt->newconn_dst);
-      newpkt.newconn_proto = pkt->newconn_proto;
-      newpkt.newconn_size = pkt->newconn_size;
+      newpkt->newconn_id = htons(pkt->newconn_id);
+      newpkt->newconn_src = htonl(pkt->newconn_src);
+      newpkt->newconn_dst = htonl(pkt->newconn_dst);
+      newpkt->newconn_proto = pkt->newconn_proto;
+      newpkt->newconn_size = pkt->newconn_size;
       break;
     case PACKET_DELCONN:
-      newpkt.delconn_id = htons(pkt->delconn_id);
+      newpkt->delconn_id = htons(pkt->delconn_id);
       break;
     case PACKET_DATA:
-      newpkt.data_connid = htons(pkt->data_connid);
-      newpkt.data_size = pkt->data_size;
+      newpkt->data_connid = htons(pkt->data_connid);
+      newpkt->data_size = pkt->data_size;
       break;
     case PACKET_NAME:
       if (pkt->name_len > GG_PKTARG_MAX)
         goto invalid;
-      packet_len = packet_len + pkt->name_len;
-      newpkt.name_addr = htonl(pkt->name_addr);
-      newpkt.name_len = pkt->name_len;
-      strncpy((char *)newpkt.name_fqdn, (char *)pkt->name_fqdn,
+      newpkt->name_addr = htonl(pkt->name_addr);
+      newpkt->name_len = pkt->name_len;
+      strncpy((char *)newpkt->name_fqdn, (char *)pkt->name_fqdn,
               pkt->name_len);
       break;
     case PACKET_FORK:
-      newpkt.fork_pid = htonl(pkt->fork_pid);
-      newpkt.fork_ppid = htonl(pkt->fork_ppid);
-      newpkt.fork_cpid = htonl(pkt->fork_cpid);
-      newpkt.fork_tgid = htonl(pkt->fork_tgid);
+      newpkt->fork_pid = htonl(pkt->fork_pid);
+      newpkt->fork_ppid = htonl(pkt->fork_ppid);
+      newpkt->fork_cpid = htonl(pkt->fork_cpid);
+      newpkt->fork_tgid = htonl(pkt->fork_tgid);
       break;
     case PACKET_EXEC:
       if (pkt->exec_cmdlen > GG_PKTARG_MAX)
         goto invalid;
-      packet_len = packet_len + pkt->exec_cmdlen;
-      newpkt.exec_pid = htonl(pkt->exec_pid);
-      newpkt.exec_cmdlen = pkt->exec_cmdlen;
-      strncpy((char *)newpkt.exec_cmd, (char *)pkt->exec_cmd,
+      newpkt->exec_pid = htonl(pkt->exec_pid);
+      newpkt->exec_cmdlen = pkt->exec_cmdlen;
+      strncpy((char *)newpkt->exec_cmd, (char *)pkt->exec_cmd,
               pkt->exec_cmdlen);
       break;
     case PACKET_EXIT:
-      newpkt.exit_pid = htonl(pkt->exit_pid);
-      newpkt.exit_tgid = htonl(pkt->exit_tgid);
-      newpkt.exit_ecode = pkt->exit_ecode;
+      newpkt->exit_pid = htonl(pkt->exit_pid);
+      newpkt->exit_tgid = htonl(pkt->exit_tgid);
+      newpkt->exit_ecode = pkt->exit_ecode;
       break;
     default:
       error("Unsupported packet type");
-      return NULL;
+      return -1;
   }
-  *len = packet_len;
-  return &newpkt;
+  return pkt_getsize(newpkt);
 
 invalid:
   error("invalid packet");
-  return NULL;
+  return -1;
+}
+
+int
+pkt_getsize(struct gg_packet *pkt)
+{
+  int packet_len;
+
+  packet_len = gg_packet_props[pkt->type].size; // XXX never overflow ?
+  switch(pkt->type) {
+    case PACKET_NEWCONN:
+    case PACKET_DELCONN:
+    case PACKET_DATA:
+    case PACKET_FORK:
+    case PACKET_EXIT:
+      break;
+    case PACKET_NAME:
+      if (pkt->name_len > GG_PKTARG_MAX)
+        goto invalid;
+      packet_len = packet_len + pkt->name_len;
+      break;
+    case PACKET_EXEC:
+      if (pkt->exec_cmdlen > GG_PKTARG_MAX)
+        goto invalid;
+      packet_len = packet_len + pkt->exec_cmdlen;
+      break;
+    default:
+      error("Unsupported packet type");
+      return -1;
+  }
+  return packet_len;
+
+invalid:
+  error("invalid packet");
+  return -1;
 }
 
 /*
diff --git a/libglouglou/libglouglou.h b/libglouglou/libglouglou.h
index 8ee890f..989bdff 100644
--- a/libglouglou/libglouglou.h
+++ b/libglouglou/libglouglou.h
@@ -12,6 +12,7 @@
 
 #define PACKET_VERSION 1
 #define PACKET_BUFFER_SIZE 16384
+#define PACKET_SNDBUF_MAX 500
 #define GG_PKTARG_MAX 30
 
 #define PACKET_HEADER_SIZE 2
@@ -117,7 +118,7 @@ struct gg_user {
   int id;
   int sock;
   struct sockaddr_in addr;
-  char *buf[16384];
+  struct sendbuf *sbuf;
 };
 
 struct gg_server {
@@ -130,7 +131,8 @@ struct gg_server {
   int (*handle_conn)(struct gg_server *, struct gg_user *);
   int (*handle_packet)(struct gg_server *, struct gg_user *, struct gg_packet *);
   void *usrdata;
-  LIST_HEAD(, gg_user)	user_list;
+  LIST_HEAD(, gg_user) user_list;
+  int user_count;
   int user_id_count;
 };
 
@@ -151,6 +153,7 @@ struct gg_client {
   int (*handle_conn)(struct gg_client *);
   int (*handle_packet)(struct gg_client *, struct gg_packet *);
   void *usrdata;
+  struct sendbuf *sbuf;
 };
 
 struct gg_server *gg_server_start(struct event_base *, char *, int,
diff --git a/libglouglou/sendbuf.c b/libglouglou/sendbuf.c
new file mode 100644
index 0000000..15a7d23
--- /dev/null
+++ b/libglouglou/sendbuf.c
@@ -0,0 +1,141 @@
+#include <stdlib.h>
+#include <string.h>
+
+#include "sendbuf.h"
+
+static int flushbuf(struct sendbuf *);
+static void cb_timer(evutil_socket_t, short, void *);
+
+/*
+ * Public
+ */
+
+/*
+ * Create a sendbuf
+ */
+struct sendbuf *
+sendbuf_new(struct event_base *ev_base, int buffer_size, int msec_max,
+            int (*send_func)(void *, int, void *), void *usrdata)
+{
+  struct sendbuf *sbuf = NULL;
+  struct event *ev_timer;
+
+  sbuf = calloc(1, sizeof(struct sendbuf));
+  if (!sbuf)
+    return NULL;
+  sbuf->ev_base = ev_base;
+  sbuf->msec_max = msec_max;
+  sbuf->buffer_size = buffer_size;
+  sbuf->send_func = send_func;
+  sbuf->usrdata = usrdata;
+  sbuf->buffer = malloc(sbuf->buffer_size);
+  if (!sbuf->buffer)
+    goto err;
+
+  ev_timer = evtimer_new(ev_base, cb_timer, sbuf);
+  sbuf->ev_timer = ev_timer;
+  sbuf->ev_timer_tv.tv_usec = msec_max * 1000;
+  evtimer_add(ev_timer, &sbuf->ev_timer_tv);
+
+  return sbuf;
+
+err:
+  sendbuf_free(sbuf);
+  return NULL;
+}
+
+void
+sendbuf_free(struct sendbuf *sbuf)
+{
+  if (sbuf->ev_timer)
+    event_del(sbuf->ev_timer);
+  if (sbuf->buffer && sbuf->send_func)
+    flushbuf(sbuf);
+  if (sbuf->buffer)
+    free(sbuf->buffer);
+  free(sbuf);
+}
+
+/*
+ * Append to the token buffer data to be sent
+ * uses a memcpy, in contrary to sendbuf_gettoken().
+ * return size on success, -1 on error
+ */
+int
+sendbuf_append(struct sendbuf *sbuf, void *token, int size)
+{
+  if (sbuf->buffer_pos + size >= sbuf->buffer_size)
+    if (flushbuf(sbuf) == -1)
+      return -1;
+
+  memcpy(sbuf->buffer + sbuf->buffer_pos, token, size);
+  sbuf->buffer_pos = sbuf->buffer_pos + size;
+
+  return size;
+}
+
+/*
+ * Returns a token buffer to write data to be sent
+ * avoids memcpy, in contrary to sendbuf_append().
+ * might return NULL if the sendbuf is temporary full
+ */
+void *
+sendbuf_gettoken(struct sendbuf *sbuf, int size)
+{
+  void *token;
+
+  if (sbuf->buffer_pos + size >= sbuf->buffer_size)
+    if (flushbuf(sbuf) == -1)
+      return NULL;
+
+  token = sbuf->buffer + sbuf->buffer_pos;
+  sbuf->buffer_pos = sbuf->buffer_pos + size;
+
+  return token;
+}
+
+/*
+ * Private
+ */
+
+/*
+ * Note that you can still add data to the buffer even if flushing is in
+ * progress
+ * returns 0 on success or -1 on error
+ */
+static int
+flushbuf(struct sendbuf *sbuf)
+{
+  int tosend, sent;
+
+  if (sbuf->buffer_pos == 0)
+    return 0;
+
+  sbuf->flushing = 1;
+
+  tosend = sbuf->buffer_pos - sbuf->flushing_pos;
+  sent = sbuf->send_func(sbuf->buffer + sbuf->flushing_pos,
+                         tosend, sbuf->usrdata);
+  if (sent == -1) {
+    // XXX handle error
+    return -1;
+  } else if (sent < tosend) {
+    sbuf->flushing_pos = sbuf->flushing_pos + sent;
+    return -1;
+  }
+  sbuf->buffer_pos = 0;
+
+  sbuf->flushing = 0;
+  sbuf->flushing_pos = 0;
+  return 0;
+}
+
+static void
+cb_timer(evutil_socket_t fd, short what, void *arg)
+{
+  struct sendbuf *sbuf;
+
+  sbuf = arg;
+  flushbuf(sbuf);
+  evtimer_add(sbuf->ev_timer, &sbuf->ev_timer_tv);
+}
diff --git a/libglouglou/sendbuf.h b/libglouglou/sendbuf.h
new file mode 100644
index 0000000..0290202
--- /dev/null
+++ b/libglouglou/sendbuf.h
@@ -0,0 +1,23 @@
+#include <event.h>
+#include <sys/queue.h>
+
+struct sendbuf {
+  struct event_base *ev_base;
+  struct event      *ev_timer;
+  struct timeval     ev_timer_tv;
+  int   msec_max;
+  int   buffer_size;
+  void *buffer;
+  int   buffer_pos; /* next to use in buffer */
+  int   flushing;
+  int   flushing_pos; /* next to send in buffer */
+  int (*send_func)(void *, int, void *);
+  void *usrdata;
+};
+
+struct sendbuf *sendbuf_new(struct event_base *, int, int,
+                            int (*send_func)(void *, int, void *),
+                            void *);
+void            sendbuf_free(struct sendbuf *);
+int             sendbuf_append(struct sendbuf *, void *, int);
+void           *sendbuf_gettoken(struct sendbuf *, int);