hopefully fix the infinite loop in gatling

io.h

@@ -78,6 +78,9 @@ int64 io_canwrite();
 /* return next descriptor with expired timeout */
 int64 io_timeouted();
 
+/* is this fd over its timeout? */
+int io_timedout(int64 d);
+
 /* 1 means: have IO_FD_CANWRITE, IO_FD_BLOCK and IO_FD_NONBLOCK,
  * will be incremented if API is extended in the future */
 #define HAVE_IO_FD_FLAGS 1
@@ -124,11 +127,23 @@ int64 io_receivefd(int64 sock);
 
 int io_starteventloopthread(unsigned int threads);
 
+#define HAVE_IO_QUEUEFORREAD
+/* Artificially queue a file descriptor as readable.
+ * The next call to io_canread will return this descriptor. */
+int io_queueforread(int64 d);
+/* Artificially queue a file descriptor as writable.
+ * The next call to io_canread will return this descriptor. */
+int io_queueforwrite(int64 d);
+
 typedef int64 (*io_write_callback)(int64 s,const void* buf,uint64 n);
 
 /* used internally, but hey, who knows */
 int64 io_mmapwritefile(int64 out,int64 in,uint64 off,uint64 bytes,io_write_callback writecb);
 
+/* only needed for debugging, will print some stats into the buffer to
+ * aid in debugging the state machine if a descriptor loops or so */
+unsigned int io_debugstring(int64 s,char* buf,unsigned int bufsize);
+
 #ifdef __MINGW32__
 #include_next <io.h>
 #endif

io/io_debugstring.c

@@ -0,0 +1,62 @@
+#include "io_internal.h"
+#include <iarray.h>
+#include <fmt.h>
+#include <time.h>
+
+unsigned int io_debugstring(int64 s,char* buf,unsigned int bufsize) {
+  struct timeval tv;
+  unsigned int i;
+  io_entry* e;
+  gettimeofday(&tv,NULL);
+  e=iarray_get(&io_fds,s);
+  if (!e) return 0;
+  if (bufsize<100) return 0;
+  i=fmt_str(buf,"first_readable ");
+  i+=fmt_long(buf+i,first_readable);
+  i+=fmt_str(buf+i,", first_writeable ");
+  i+=fmt_long(buf+i,first_writeable);
+#ifdef HAVE_SIGIO
+  i+=fmt_str(buf+i,", alt_firstread ");
+  i+=fmt_long(buf+i,alt_firstread);
+  i+=fmt_str(buf+i,", alt_firstwrite ");
+  i+=fmt_long(buf+i,alt_firstwrite);
+#endif
+  i+=fmt_str(buf+i,"\n");
+
+  i+=fmt_str(buf+i,"fd # ");
+  i+=fmt_ulong(buf+i,s);
+  i+=fmt_str(buf+i,": ");
+  if (bufsize-i<100) return 0;
+  i+=fmt_str(buf+i,"timeout ");
+  i+=fmt_long(buf+i,e->timeout.sec.x-4611686018427387914ULL-tv.tv_sec);
+  i+=fmt_str(buf+i,".");
+  i+=fmt_ulong(buf+i,e->timeout.nano);
+  i+=fmt_str(buf+i," ");
+  if (bufsize-i<100) return 0;
+  if (e->wantread) i+=fmt_str(buf+i,"wr ");
+  if (e->wantwrite) i+=fmt_str(buf+i,"ww ");
+  if (e->canread) i+=fmt_str(buf+i,"cr ");
+  if (e->canwrite) i+=fmt_str(buf+i,"cw ");
+  if (e->nonblock) i+=fmt_str(buf+i,"nb ");
+  if (!e->inuse) i+=fmt_str(buf+i,"!inuse ");
+  if (e->kernelwantread) i+=fmt_str(buf+i,"kwr ");
+  if (e->kernelwantwrite) i+=fmt_str(buf+i,"kww ");
+  if (e->epolladded) i+=fmt_str(buf+i,"ea ");
+  if (e->mmapped) {
+    i+=fmt_str(buf+i,"mmap(");
+    i+=fmt_xlong(buf+i,(unsigned long)e->mmapped);
+    i+=fmt_str(buf+i,",");
+    i+=fmt_xlong(buf+i,(unsigned long)e->maplen);
+    i+=fmt_str(buf+i,"@");
+    i+=fmt_xlonglong(buf+i,(unsigned long)e->mapofs);
+  }
+  if (bufsize-i<100) return 0;
+  i+=fmt_str(buf+i,"next_read ");
+  i+=fmt_long(buf+i,e->next_read);
+  i+=fmt_str(buf+i," next_write ");
+  i+=fmt_long(buf+i,e->next_write);
+  i+=fmt_str(buf+i," cookie ");
+  i+=fmt_xlonglong(buf+i,(unsigned long)e->cookie);
+  buf[i]=0;
+  return i;
+}

io/io_eagain.c

@@ -9,10 +9,12 @@ void io_eagain(int64 d) {
     if (d==alt_firstread) {
       debug_printf(("io_eagain: dequeueing %lld from alt read queue (next is %ld)\n",d,e->next_read));
       alt_firstread=e->next_read;
+      e->next_read=-1;
     }
     if (d==alt_firstwrite) {
       debug_printf(("io_eagain: dequeueing %lld from alt write queue (next is %ld)\n",d,e->next_write));
       alt_firstwrite=e->next_write;
+      e->next_write=-1;
     }
 #endif
   }

io/io_eagain_read.c

@@ -8,8 +8,8 @@ void io_eagain_read(int64 d) {
     if (d==alt_firstread) {
       debug_printf(("io_eagain: dequeueing %lld from alt read queue (next is %ld)\n",d,e->next_read));
       alt_firstread=e->next_read;
+      e->next_read=-1;
     }
 #endif
-    e->next_read=-1;
   }
 }

io/io_eagain_write.c

@@ -8,8 +8,8 @@ void io_eagain_write(int64 d) {
     if (d==alt_firstwrite) {
       debug_printf(("io_eagain: dequeueing %lld from alt write queue (next is %ld)\n",d,e->next_write));
       alt_firstwrite=e->next_write;
+      e->next_write=-1;
     }
 #endif
-    e->next_write=-1;
   }
 }

io/io_fd_canwrite.3

@@ -7,7 +7,7 @@ io_fd_canwrite \- prepare descriptor for io_wait
 int \fBio_fd\fP(int64 fd);
 int \fBio_fd_canwrite\fP(int64 fd);
 .SH DESCRIPTION
-io_fd_canwrite is just like io_fd, except that assumes the descriptor
+io_fd_canwrite is just like io_fd, except that it assumes the descriptor
 is writable, which may save a syscall or two.  This assumption is true
 in most cases, because the kernel buffers writes.  Noteworthy cases in
 which you need to use io_fd instead of io_fd_canwrite are unconnected

io/io_fd_flags.3

@@ -0,0 +1,40 @@
+.TH io_fd_flags 3
+.SH NAME
+io_fd_flags \- prepare descriptor for io_wait
+.SH SYNTAX
+.B #include <io.h>
+
+int \fBio_fd\fP(int64 fd);
+
+#ifdef HAVE_IO_FD_FLAGS
+
+int \fBio_fd_flags\fP(int64 fd);
+.SH DESCRIPTION
+io_fd_flags behaves just like io_fd, but certain flags can be
+bitwise-ORed to it to alter its behavior:
+
+.RS 0
+.IP IO_FD_CANWRITE
+tell io_fd that the descriptor is writable.  This is useful so
+io_wantwrite can queue the descriptor immediately and there is no need
+to query the operating system event reporting mechanism.
+.IP IO_FD_BLOCK
+tell io_fd that the descriptor is blocking.
+.IP IO_FD_NONBLOCK
+tell io_fd that the descriptor is non-blocking.
+.RE
+
+Normally, io_fd calls fcntl to ask the operating system whether the
+descriptor is blocking or not.  The frameworks needs to know because it
+alters how io_tryread and io_trywrite handle the socket.  Never pass
+both IO_FD_BLOCK and IO_FD_NONBLOCK at the same time.
+
+Newly connected stream sockets are always writable if the connection is
+established, so it is usually safe to pass IO_FD_CANWRITE.  The main
+exception case where IO_FD_CANWRITE should not be passed is on a
+non-blocking socket where a connect() is pending.  Then you need to poll
+for writability to get notified when the connection is established.
+.SH "RETURN VALUE"
+io_fd_flags returns 1 on success, 0 on error.
+.SH "SEE ALSO"
+io_fd(3), io_fd_canwrite(3)

io/io_timedout.c

@@ -0,0 +1,8 @@
+#include "io_internal.h"
+
+int io_timedout(int64 d) {
+  tai6464 now;
+  io_entry* e=iarray_get(&io_fds,d);
+  taia_now(&now);
+  return (e && e->timeout.sec.x && taia_less(&e->timeout,&now));
+}

io/io_tryread.c

@@ -116,9 +116,9 @@ int64 io_tryread(int64 d,char* buf,int64 len) {
     if (d==alt_firstread) {
       debug_printf(("io_tryread: dequeueing %ld from alt read queue (next is %ld)\n",d,e->next_read));
       alt_firstread=e->next_read;
+      e->next_read=-1;
     }
 #endif
-    e->next_read=-1;
   }
   return r;
 }

io/io_trywrite.c

@@ -110,9 +110,9 @@ int64 io_trywrite(int64 d,const char* buf,int64 len) {
     if (d==alt_firstwrite) {
       debug_printf(("io_trywrite: dequeueing %ld from alt write queue (next is %ld)\n",d,e->next_write));
       alt_firstwrite=e->next_write;
+      e->next_write=-1;
     }
 #endif
-    e->next_write=-1;
   }
   return r;
 }

io_internal.h

@@ -37,7 +37,6 @@ my_extern HANDLE io_comport;
 
 typedef struct {
   tai6464 timeout;
-  int fd;
   unsigned int wantread:1;	/* does the app want to read/write? */
   unsigned int wantwrite:1;
   unsigned int canread:1;	/* do we know we can read/write? */