implement resizable I/O buffers and tune buffer sizes to normal requests

git-svn-id: http://arthurdejong.org/svn/nss-pam-ldapd/nss-ldapd@630 ef36b2f9-881f-0410-afb5-c4e39611909c

5 files changed, 110 insertions, 41 deletions

diff --git a/common/tio.c b/common/tio.c
index afd1555..c37fb46 100644
--- a/common/tio.c
+++ b/common/tio.c
@@ -41,24 +41,19 @@
 #define ETIME ETIMEDOUT
 #endif /* ETIME */
 
-/* buffer size for both read and write buffers */
-/* TODO: pass this along with the open function */
-/* Note that this size should not be larger than SSIZE_MAX because otherwise
-   write() of such blocks is undefined */
-#define TIO_BUFFERSIZE (4*1024)
-
 /* structure that holds a buffer
    the buffer contains the data that is between the application and the
    file descriptor that is used for efficient transfer
    the buffer is built up as follows:
    |.....********......|
-         ^start
-         ^--len--^       */
+         ^start        ^size
+         ^--len--^           */
 struct tio_buffer {
   uint8_t *buffer;
-  /* the size is TIO_BUFFERSIZE */
-  size_t len; /* the number of bytes used in the buffer (from the start) */
-  int start; /* the start of the buffer (space before the start is unused) */
+  size_t size;      /* the size of the buffer */
+  size_t maxsize;   /* the maximum size of the buffer */
+  size_t start;     /* the start of the data (before start is unused) */
+  size_t len;       /* size of the data (from the start) */
 };
 
 /* structure that holds all the state for files */
@@ -132,7 +127,9 @@ static inline int tio_tv_remaining(struct timeval *tv, const struct timeval *dea
 }
 
 /* open a new TFILE based on the file descriptor */
-TFILE *tio_fdopen(int fd,struct timeval *readtimeout,struct timeval *writetimeout)
+TFILE *tio_fdopen(int fd,struct timeval *readtimeout,struct timeval *writetimeout,
+                  size_t initreadsize,size_t maxreadsize,
+                  size_t initwritesize,size_t maxwritesize)
 {
   struct tio_fileinfo *fp;
   fp=(struct tio_fileinfo *)malloc(sizeof(struct tio_fileinfo));
@@ -140,24 +137,29 @@ TFILE *tio_fdopen(int fd,struct timeval *readtimeout,struct timeval *writetimeou
     return NULL;
   fp->fd=fd;
   /* initialize read buffer */
-  fp->readbuffer.buffer=(uint8_t *)malloc(TIO_BUFFERSIZE);
+  fp->readbuffer.buffer=(uint8_t *)malloc(initreadsize);
   if (fp->readbuffer.buffer==NULL)
   {
     free(fp);
     return NULL;
   }
+  fp->readbuffer.size=initreadsize;
+  fp->readbuffer.maxsize=maxreadsize;
   fp->readbuffer.start=0;
   fp->readbuffer.len=0;
   /* initialize write buffer */
-  fp->writebuffer.buffer=(uint8_t *)malloc(TIO_BUFFERSIZE);
+  fp->writebuffer.buffer=(uint8_t *)malloc(initwritesize);
   if (fp->writebuffer.buffer==NULL)
   {
     free(fp->readbuffer.buffer);
     free(fp);
     return NULL;
   }
+  fp->writebuffer.size=initwritesize;
+  fp->writebuffer.maxsize=maxwritesize;
   fp->writebuffer.start=0;
   fp->writebuffer.len=0;
+  /* initialize other attributes */
   fp->readtimeout.tv_sec=readtimeout->tv_sec;
   fp->readtimeout.tv_usec=readtimeout->tv_usec;
   fp->writetimeout.tv_sec=writetimeout->tv_sec;
@@ -214,6 +216,8 @@ int tio_read(TFILE *fp, void *buf, size_t count)
 {
   struct timeval deadline;
   int rv;
+  uint8_t *tmp;
+  size_t newsz;
   /* have a more convenient storage type for the buffer */
   uint8_t *ptr=(uint8_t *)buf;
   /* build a time by which we should be finished */
@@ -235,7 +239,7 @@ int tio_read(TFILE *fp, void *buf, size_t count)
       return 0;
     }
     /* empty what we have and continue from there */
-    if (fp->readbuffer.len > 0)
+    if (fp->readbuffer.len>0)
     {
       if (ptr!=NULL)
       {
@@ -243,23 +247,42 @@ int tio_read(TFILE *fp, void *buf, size_t count)
         ptr+=fp->readbuffer.len;
       }
       count-=fp->readbuffer.len;
-    }
-    /* if we have room in the buffer for more don't clear the buffer */
-    if ((fp->read_resettable)&&((fp->readbuffer.start+fp->readbuffer.len)<TIO_BUFFERSIZE))
-    {
       fp->readbuffer.start+=fp->readbuffer.len;
+      fp->readbuffer.len=0;
     }
-    else
+    /* after this point until the read fp->readbuffer.len is 0 */
+    if (!fp->read_resettable)
     {
+      /* the stream is not resettable, re-use the buffer */
       fp->readbuffer.start=0;
-      fp->read_resettable=0;
     }
-    fp->readbuffer.len=0;
+    else if (fp->readbuffer.start>=(fp->readbuffer.size-4))
+    {
+      /* buffer is running empty, try to grow buffer */
+      if (fp->readbuffer.size<fp->readbuffer.maxsize)
+      {
+        newsz=fp->readbuffer.size*2;
+        if (newsz>fp->readbuffer.maxsize)
+          newsz=fp->readbuffer.maxsize;
+        tmp=realloc(fp->readbuffer.buffer,newsz);
+        if (tmp!=NULL)
+        {
+          fp->readbuffer.buffer=tmp;
+          fp->readbuffer.size=newsz;
+        }
+      }
+      /* if buffer still does not contain enough room, clear resettable */
+      if (fp->readbuffer.start>=(fp->readbuffer.size-4))
+      {
+        fp->readbuffer.start=0;
+        fp->read_resettable=0;
+      }
+    }
     /* wait until we have input */
     if (tio_select(fp->fd,1,&deadline))
       return -1;
     /* read the input in the buffer */
-    rv=read(fp->fd,fp->readbuffer.buffer+fp->readbuffer.start,TIO_BUFFERSIZE-fp->readbuffer.start);
+    rv=read(fp->fd,fp->readbuffer.buffer+fp->readbuffer.start,fp->readbuffer.size-fp->readbuffer.start);
     /* check for errors */
     if ((rv==0)||((rv<0)&&(errno!=EINTR)&&(errno!=EAGAIN)))
       return -1; /* something went wrong with the read */
@@ -330,12 +353,14 @@ FIXME: we have a race condition here (setting and restoring the signal mask), th
 int tio_write(TFILE *fp, const void *buf, size_t count)
 {
   size_t fr;
+  uint8_t *tmp;
+  size_t newsz;
   const uint8_t *ptr=(const uint8_t *)buf;
   /* keep filling the buffer until we have bufferred everything */
   while (count>0)
   {
     /* figure out free size in buffer */
-    fr=TIO_BUFFERSIZE-(fp->writebuffer.start+fp->writebuffer.len);
+    fr=fp->writebuffer.size-(fp->writebuffer.start+fp->writebuffer.len);
     if (count <= fr)
     {
       /* the data fits in the buffer */
@@ -343,9 +368,23 @@ int tio_write(TFILE *fp, const void *buf, size_t count)
       fp->writebuffer.len+=count;
       return 0;
     }
-    else if (fr > 0)
+    /* try to grow the buffer */
+    if (fp->writebuffer.size<fp->writebuffer.maxsize)
+    {
+      newsz=fp->writebuffer.size*2;
+      if (newsz>fp->writebuffer.maxsize)
+        newsz=fp->writebuffer.maxsize;
+      tmp=realloc(fp->writebuffer.buffer,newsz);
+      if (tmp!=NULL)
+      {
+        fp->writebuffer.buffer=tmp;
+        fp->writebuffer.size=newsz;
+        continue; /* try again */
+      }
+    }
+    /* fill the buffer with data that will fit */
+    if (fr > 0)
     {
-      /* fill the buffer */
       memcpy(fp->writebuffer.buffer+fp->writebuffer.start+fp->writebuffer.len,ptr,fr);
       fp->writebuffer.len+=fr;
       ptr+=fr;
diff --git a/common/tio.h b/common/tio.h
index 9f63c1b..d6c8d9c 100644
--- a/common/tio.h
+++ b/common/tio.h
@@ -48,7 +48,9 @@ typedef struct tio_fileinfo TFILE;
 /* Open a new TFILE based on the file descriptor. The timeout is set for any
    operation. The timeout value is copied so may be dereferenced after the
    call. */
-TFILE *tio_fdopen(int fd,struct timeval *readtimeout,struct timeval *writetimeout)
+TFILE *tio_fdopen(int fd,struct timeval *readtimeout,struct timeval *writetimeout,
+                  size_t initreadsize,size_t maxreadsize,
+                  size_t initwritesize,size_t maxwritesize)
   LIKE_MALLOC MUST_USE;
 
 /* Read the specified number of bytes from the stream. */
diff --git a/nslcd/nslcd.c b/nslcd/nslcd.c
index 06dfff1..c61cd15 100644
--- a/nslcd/nslcd.c
+++ b/nslcd/nslcd.c
@@ -63,6 +63,12 @@
 #include "compat/attrs.h"
 #include "compat/getpeercred.h"
 
+/* buffer sizes for I/O */
+#define READBUFFER_MINSIZE 32
+#define READBUFFER_MAXSIZE 64
+#define WRITEBUFFER_MINSIZE 64
+#define WRITEBUFFER_MAXSIZE 64*1024
+
 /* the definition of the environment */
 extern char **environ;
 
@@ -345,7 +351,9 @@ static void handleconnection(int sock,MYLDAP_SESSION *session)
   writetimeout.tv_sec=5; /* clients could be taking some time to process the results */
   writetimeout.tv_usec=0;
   /* create a stream object */
-  if ((fp=tio_fdopen(sock,&readtimeout,&writetimeout))==NULL)
+  if ((fp=tio_fdopen(sock,&readtimeout,&writetimeout,
+                     READBUFFER_MINSIZE,READBUFFER_MAXSIZE,
+                     WRITEBUFFER_MINSIZE,WRITEBUFFER_MAXSIZE))==NULL)
   {
     log_log(LOG_WARNING,"cannot create stream for writing: %s",strerror(errno));
     (void)close(sock);
diff --git a/nss/common.c b/nss/common.c
index 00aebf2..4466f1a 100644
--- a/nss/common.c
+++ b/nss/common.c
@@ -39,6 +39,12 @@
 #include "common.h"
 #include "common/tio.h"
 
+/* buffer sizes for I/O */
+#define READBUFFER_MINSIZE 1024
+#define READBUFFER_MAXSIZE 32*1024
+#define WRITEBUFFER_MINSIZE 32
+#define WRITEBUFFER_MAXSIZE 32
+
 /* returns a socket to the server or NULL on error (see errno),
    socket should be closed with fclose() */
 TFILE *nslcd_client_open()
@@ -67,7 +73,9 @@ TFILE *nslcd_client_open()
   writetimeout.tv_sec=1; /* nslcd could be loaded with requests */
   writetimeout.tv_usec=500000;
   /* create a stream object */
-  if ((fp=tio_fdopen(sock,&readtimeout,&writetimeout))==NULL)
+  if ((fp=tio_fdopen(sock,&readtimeout,&writetimeout,
+                     READBUFFER_MINSIZE,READBUFFER_MAXSIZE,
+                     WRITEBUFFER_MINSIZE,WRITEBUFFER_MAXSIZE))==NULL)
   {
     (void)close(sock);
     return NULL;
diff --git a/tests/test_tio.c b/tests/test_tio.c
index 4138181..82586e4 100644
--- a/tests/test_tio.c
+++ b/tests/test_tio.c
@@ -71,7 +71,7 @@ static void *help_tiowriter(void *arg)
   timeout.tv_sec=hargs->timeout;
   timeout.tv_usec=0;
   /* open the file */
-  fp=tio_fdopen(hargs->fd,&timeout,&timeout);
+  fp=tio_fdopen(hargs->fd,&timeout,&timeout,4*1024,8*1024,4*1024,8*1024);
   assertok(fp!=NULL);
   /* write the blocks */
   i=0;
@@ -103,7 +103,7 @@ static void *help_tioreader(void *arg)
   timeout.tv_sec=hargs->timeout;
   timeout.tv_usec=0;
   /* open the file */
-  fp=tio_fdopen(hargs->fd,&timeout,&timeout);
+  fp=tio_fdopen(hargs->fd,&timeout,&timeout,4*1024,8*1024,4*1024,8*1024);
   assertok(fp!=NULL);
   /* read the blocks */
   i=0;
@@ -213,19 +213,19 @@ static void test_reset(void)
   TFILE *fp;
   struct timeval timeout;
   size_t i,j,k,save;
-  uint8_t buf[10];
+  uint8_t buf[20];
   /* set up the socket pair */
   assertok(socketpair(AF_UNIX,SOCK_STREAM,0,sp)==0);
   /* start the writer thread */
   wargs.fd=sp[0];
-  wargs.blocksize=4*1024; /* the current TIO_BUFFERSIZE */
-  wargs.blocks=5;
+  wargs.blocksize=4*1024;
+  wargs.blocks=10;
   wargs.timeout=2;
-  assertok(pthread_create(&wthread,NULL,help_tiowriter,&wargs)==0);
+  assertok(pthread_create(&wthread,NULL,help_normwriter,&wargs)==0);
   /* set up read handle */
   timeout.tv_sec=2;
   timeout.tv_usec=0;
-  fp=tio_fdopen(sp[1],&timeout,&timeout);
+  fp=tio_fdopen(sp[1],&timeout,&timeout,2*1024,4*1024,2*1024,4*1024);
   assertok(fp!=NULL);
   /* perform 20 reads */
   i=0;
@@ -239,17 +239,29 @@ static void test_reset(void)
   /* mark and perform another 2 reads */
   tio_mark(fp);
   save=i;
-  for (k=0;k<2;k++)
+  for (k=20;k<22;k++)
   {
     assertok(tio_read(fp,buf,sizeof(buf))==0);
     /* check the buffer */
     for (j=0;j<sizeof(buf);j++)
       assert(buf[j]==(uint8_t)(i++));
   }
-  /* reset and perform the same 2 reads again and 500 more */
+  /* check that we can reset */
+  assertok(tio_reset(fp)==0);
+  /* perform 204 reads (partially the same as before) */
+  i=save;
+  for (k=20;k<224;k++)
+  {
+    assert(tio_read(fp,buf,sizeof(buf))==0);
+    /* check the buffer */
+    for (j=0;j<sizeof(buf);j++)
+      assert(buf[j]==(uint8_t)(i++));
+  }
+  /* check that we can reset */
   assertok(tio_reset(fp)==0);
+  /* perform 502 reads (partially the same) */
   i=save;
-  for (k=0;k<502;k++)
+  for (k=20;k<522;k++)
   {
     assert(tio_read(fp,buf,sizeof(buf))==0);
     /* check the buffer */
@@ -259,7 +271,7 @@ static void test_reset(void)
   /* check that reset is no longer possible */
   assertok(tio_reset(fp)!=0);
   /* read the remainder of the data 1526 reads */
-  for (k=0;k<1526;k++)
+  for (k=522;k<2048;k++)
   {
     assertok(tio_read(fp,buf,sizeof(buf))==0);
     /* check the buffer */
@@ -277,7 +289,7 @@ int main(int UNUSED(argc),char UNUSED(*argv[]))
 {
   /* normal read-writes */
   test_blocks(400,11,11,400);
-  test_blocks(4*1024,11,4*11,1024);
+  test_blocks(10*1024,11,10*11,1024);
   test_blocks(5*1023,20,20*1023,5);
   /* reader closes file sooner */
 /*  test_blocks(2*6*1023,20,20*1023,5); */