Merge branch 'lukas/ose/master-17.0/OTP-11334'

* lukas/ose/master-17.0/OTP-11334:
  ose: Fix erts assert failed printouts
  ose: fix for packet_bytes in fd/spawn driver.
  ose: Prepare slave for running on OSE
  ose: Fix bug when hunting for signal proxy
  ose: Implement tcp inet driver for OSE
  ose: Add ifdefs for HAVE_UDP
  ose: Yielding has to be done differently for background processes.
  ose: Print faults in aio sys driver calls
  ose: Prinout errno when to_erl read fails
  ose: erlang display goes to ramlog printf
  ose: Initiate stdin/stdout/stderr
  ose: Break lmconf into one per load module
  ose: Reset busy port when pdq empty
  ose: Restore the owner of the signal

4 files changed, 233 insertions, 133 deletions

diff --git a/erts/emulator/sys/ose/default.lmconf b/erts/emulator/sys/ose/beam.lmconf
index a66b0ece56..4ad46b01d9 100644
--- a/erts/emulator/sys/ose/default.lmconf
+++ b/erts/emulator/sys/ose/beam.lmconf
@@ -4,12 +4,13 @@ OSE_LM_POOL_SIZE=0x200000
 OSE_LM_MAIN_NAME=main
 OSE_LM_MAIN_STACK_SIZE=0xF000
 OSE_LM_MAIN_PRIORITY=20
+## Has to be of a type that allows MAM
 OSE_LM_PROGRAM_TYPE=APP_RAM
 OSE_LM_DATA_INIT=YES
 OSE_LM_BSS_INIT=YES
 OSE_LM_EXEC_MODEL=SHARED
 HEAP_MAX_SIZE=1000000000
-HEAP_SMALL_BUF_INIT_SIZE=64000000
+HEAP_SMALL_BUF_INIT_SIZE=20971520
 HEAP_LARGE_BUF_THRESHOLD=16000000
 HEAP_LOCK_TYPE=2
 
diff --git a/erts/emulator/sys/ose/erl_main.c b/erts/emulator/sys/ose/erl_main.c
index 03119c3fec..23a9bc93a4 100644
--- a/erts/emulator/sys/ose/erl_main.c
+++ b/erts/emulator/sys/ose/erl_main.c
@@ -30,6 +30,8 @@
 int
 main(int argc, char **argv) {
 
+  (void)stdin;(void)stdout;(void)stderr;
+
   /* When starting using pm_create -c ARGV="-- -root ..", argv[0] is the first
      part of ARGV and not the name of the executable. So we shuffle some
      pointers here to make erl_start happy. */
diff --git a/erts/emulator/sys/ose/erl_poll.c b/erts/emulator/sys/ose/erl_poll.c
index ca1ed6e53a..7d2a3d1e0b 100644
--- a/erts/emulator/sys/ose/erl_poll.c
+++ b/erts/emulator/sys/ose/erl_poll.c
@@ -551,7 +551,12 @@ int erts_poll_wait(ErtsPollSet ps,
                 fd.id, fd.signo, current_process());
              erts_send_error_to_logger_nogl(dsbufp);
 	     timeout = 0;
-	     ASSERT(0);
+             /* Under normal circumstances the signal is deallocated by the
+              * driver that issued the select operation. But in this case
+              * there's no driver waiting for such signal so we have to
+              * deallocate it here */
+             if (sig)
+                 free_buf(&sig);
 	  } else {
 	    int i;
 	    struct erts_sys_fd_type *fd = NULL;
@@ -737,6 +742,7 @@ union SIGNAL *erl_drv_ose_get_signal(ErlDrvEvent drv_ev) {
       ev->msgs = msg->next;
       ethr_mutex_unlock(&ev->mtx);
       erts_free(ERTS_ALC_T_FD_SIG_LIST,msg);
+      restore(sig);
       return sig;
     }
 }
diff --git a/erts/emulator/sys/ose/sys.c b/erts/emulator/sys/ose/sys.c
index c892cc69c7..5b950a7dae 100644
--- a/erts/emulator/sys/ose/sys.c
+++ b/erts/emulator/sys/ose/sys.c
@@ -195,7 +195,9 @@ static volatile int children_died;
       write_buff += sizeof(struct aiocb *);                             \
       memcpy(write_buff,BUFF,SIZE+1);                                   \
       SET_AIO(*write_req,FD,SIZE,write_buff);                           \
-      aio_write(write_req);                                             \
+      if (aio_write(write_req))						\
+	ramlog_printf("%s:%d: write failed with %d\n",			\
+		      __FILE__,__LINE__,errno);				\
    }                                                                    \
 } while(0)
 
@@ -210,13 +212,13 @@ static volatile int children_died;
    driver_free(buffer_ptr);                                             \
 } while(0)
 
-/* When we have several schedulers, we need to make sure
- * that scheduler issuing aio_dispatch() is the owner on the signal */
 #define DISPATCH_AIO(sig) do {                                          \
-   restore(sig);                                                        \
-   aio_dispatch(sig);                                                   \
+   if (aio_dispatch(sig))						\
+     ramlog_printf("%s:%d: dispatch failed with %d\n",			\
+		   __FILE__,__LINE__,errno);				\
    } while(0)
 
+#define AIO_PIPE_SIZE 1024
 
 /* debug print macros */
 #define DEBUG_RES 0
@@ -371,6 +373,63 @@ thr_create_prepare_child(void *vtcdp)
 
 #endif /* #ifdef USE_THREADS */
 
+/* The two functions below are stolen from win_con.c
+   They have to use malloc/free/realloc directly becasue
+   we want to do able to do erts_printf very early on.
+ */
+#define VPRINTF_BUF_INC_SIZE 128
+static erts_dsprintf_buf_t *
+grow_vprintf_buf(erts_dsprintf_buf_t *dsbufp, size_t need)
+{
+    char *buf;
+    size_t size;
+
+    ASSERT(dsbufp);
+
+    if (!dsbufp->str) {
+	size = (((need + VPRINTF_BUF_INC_SIZE - 1)
+		 / VPRINTF_BUF_INC_SIZE)
+		* VPRINTF_BUF_INC_SIZE);
+	buf = (char *) malloc(size * sizeof(char));
+    }
+    else {
+	size_t free_size = dsbufp->size - dsbufp->str_len;
+
+	if (need <= free_size)
+	    return dsbufp;
+
+	size = need - free_size + VPRINTF_BUF_INC_SIZE;
+	size = (((size + VPRINTF_BUF_INC_SIZE - 1)
+		 / VPRINTF_BUF_INC_SIZE)
+		* VPRINTF_BUF_INC_SIZE);
+	size += dsbufp->size;
+        buf = (char *) realloc((void *) dsbufp->str,
+			       size * sizeof(char));
+    }
+    if (!buf)
+	return NULL;
+    if (buf != dsbufp->str)
+	dsbufp->str = buf;
+    dsbufp->size = size;
+    return dsbufp;
+}
+
+static int erts_sys_ramlog_printf(char *format, va_list arg_list)
+{
+    int res,i;
+    erts_dsprintf_buf_t dsbuf = ERTS_DSPRINTF_BUF_INITER(grow_vprintf_buf);
+    res = erts_vdsprintf(&dsbuf, format, arg_list);
+    if (res >= 0) {
+      for (i = 0; i < dsbuf.str_len; i+= 50)
+	/* We print 50 characters at a time because otherwise
+	   the ramlog looks broken */
+        ramlog_printf("%.*s",dsbuf.str_len-50 < 0?dsbuf.str_len:50,dsbuf.str+i);
+    }
+    if (dsbuf.str)
+      free((void *) dsbuf.str);
+    return res;
+}
+
 void
 erts_sys_pre_init(void)
 {
@@ -409,6 +468,9 @@ erts_sys_pre_init(void)
     children_died = 0;
 #endif
 #endif /* USE_THREADS */
+
+    erts_printf_stdout_func = erts_sys_ramlog_printf;
+
     erts_smp_atomic_init_nob(&sys_misc_mem_sz, 0);
 }
 
@@ -650,7 +712,7 @@ static void stop_select(ErlDrvEvent, void*);
 static PROCESS
 get_signal_proxy_pid(void) {
    union SIGNAL *sig;
-   SIGSELECT any_sig[] = {0};
+   SIGSELECT any_sig[] = {1,ERTS_SIGNAL_OSE_DRV_ATTACH};
 
    if (!sig_proxy_pid) {
       sig = alloc(sizeof(union SIGNAL), ERTS_SIGNAL_OSE_DRV_ATTACH);
@@ -685,7 +747,7 @@ resolve_signal(union SIGNAL* sig) {
 struct erl_drv_entry spawn_driver_entry = {
     spawn_init,
     spawn_start,
-    erl_stop,
+    NULL, /* erl_stop, */
     output,
     ready_input,
     ready_output,
@@ -784,7 +846,11 @@ set_driver_data(ErlDrvPort port_num,
 
     /* READ */
     if (read_write & DO_READ) {
-       efs_examine_fd(ifd, FLIB_FD_HANDLE, &driver_data[ifd].handle, 0);
+       EfsStatus res = efs_examine_fd(ifd, FLIB_FD_HANDLE,
+				      &driver_data[ifd].handle, 0);
+       if (res != EFS_SUCCESS)
+	 ramlog_printf("%s:%d: efs_examine_fd(%d) failed with %d\n",
+		       __FILE__,__LINE__,ifd,errno);
        driver_data[ifd].ifd = ifd;
        driver_data[ifd].packet_bytes = packet_bytes;
        driver_data[ifd].port_num = port_num;
@@ -792,10 +858,9 @@ set_driver_data(ErlDrvPort port_num,
 
        /* async read struct */
        memset(&driver_data[ifd].aiocb, 0, sizeof(struct aiocb));
-       driver_data[ifd].aiocb.aio_buf = driver_alloc(255);
+       driver_data[ifd].aiocb.aio_buf = driver_alloc(AIO_PIPE_SIZE);
        driver_data[ifd].aiocb.aio_fildes = ifd;
-       driver_data[ifd].aiocb.aio_nbytes = 255;
-
+       driver_data[ifd].aiocb.aio_nbytes = (packet_bytes?packet_bytes:AIO_PIPE_SIZE);
        driver_data[ifd].alive = 1;
        driver_data[ifd].status = 0;
        driver_data[ifd].input_event =
@@ -826,7 +891,9 @@ set_driver_data(ErlDrvPort port_num,
        (void) driver_select(port_num, driver_data[ifd].input_event,
 			     (ERL_DRV_READ | ERL_DRV_USE), 1);
 
-       aio_read(&driver_data[ifd].aiocb);
+       if (aio_read(&driver_data[ifd].aiocb))
+	 ramlog_printf("%s:%d: aio_read(%d) failed with %d\n",
+		       __FILE__,__LINE__,ifd,errno);
     }
     else { /* WRITE ONLY */
        efs_examine_fd(ofd, FLIB_FD_HANDLE, &driver_data[ofd].handle, 0);
@@ -926,7 +993,7 @@ spawn_start(ErlDrvPort port_num, char* name, SysDriverOpts* opts)
 {
     int ifd[2];
     int ofd[2];
-    static uint32_t ticker = 0;
+    static uint32_t ticker = 1;
     PmStatus pm_status;
     OSDOMAIN domain = PM_NEW_DOMAIN;
     PROCESS progpid, mainbid, mainpid;
@@ -938,39 +1005,53 @@ spawn_start(ErlDrvPort port_num, char* name, SysDriverOpts* opts)
     int handle_size;
     char *ptr;
 
-    /* handle arguments */
-    ptr = strchr(name, ' ');
-    if (ptr != NULL) {
-       *ptr ='\0';
-       ptr++;
-       args = ptr;
+   
+    args = driver_alloc(strlen(name)+1);
+    strcpy(args, name);
+    /* We need to handle name in three parts 
+     * - install handle (must be unique)
+     * - install binary (needed for ose_pm_install_load_module())
+     * - full path (as argument to the spawned applications env.var
+     */
+
+    /* full path including arguments */
+    args = driver_alloc(strlen(name)+1);
+    strcpy(args, name);
+
+    /* handle path */
+    tmp_handle = strrchr(name, '/');
+    if (tmp_handle == NULL) {
+       tmp_handle = name;
     }
     else {
-       args = NULL;
+       tmp_handle++;
     }
 
-    /* create an install handle */
-    ptr = strrchr(name, '/');
+    /* handle args */
+    ptr = strchr(tmp_handle, ' ');
     if (ptr != NULL) {
-       ptr++;
-       tmp_handle = ptr;
+       *ptr = '\0';
+       handle_size = ptr - tmp_handle;
     }
     else {
-       tmp_handle = name;
+       handle_size = strlen(name)+1;
     }
-    handle_size = strlen(tmp_handle)+1;
-    handle_size += (ticker<10)?3:((ticker<100)?4:5);
 
+    /* make room for ticker */
+    handle_size += (ticker<10)?3:((ticker<100)?4:5);
     handle = driver_alloc(handle_size);
-    snprintf(handle, handle_size, "%s_%d", tmp_handle, ticker);
-
+    
     do {
-       snprintf(handle, handle_size, "%s_%d", tmp_handle, ticker++);
+       snprintf(handle, handle_size, "%s_%d", tmp_handle, ticker);
        pm_status = ose_pm_install_load_module(0, "ELF", name, handle,
                                               0, 0, NULL);
-
+       ticker++;
     } while (pm_status == PM_EINSTALL_HANDLE_ALREADY_INSTALLED);
-    DEBUG_CHECK_RES(pm_status, PM_SUCCESS);
+
+    if (pm_status != PM_SUCCESS) {
+       errno = ENOSYS; /* FIXME add comment */
+       return ERL_DRV_ERROR_ERRNO; 
+    }
 
     /* Create Program */
     pm_status = ose_pm_create_program(&domain, handle, 0, 0,
@@ -1143,17 +1224,13 @@ static void erl_stop(ErlDrvData drv_data)
       if (data->ifd != data->ofd) { /* read and write */
          nbio_stop_fd(data->port_num, data->input_event);
          nbio_stop_fd(data->port_num, data->output_event);
-         driver_select(data->port_num, data->input_event, ERL_DRV_USE, 0);
-         driver_select(data->port_num, data->output_event, ERL_DRV_USE, 0);
       }
       else { /* write only */
          nbio_stop_fd(data->port_num, data->output_event);
-         driver_select(data->port_num, data->output_event, ERL_DRV_USE, 0);
       }
    }
    else { /* read only */
       nbio_stop_fd(data->port_num, data->input_event);
-      driver_select(data->port_num, data->input_event, ERL_DRV_USE, 0);
    }
    close(data->ifd);
    close(data->ofd);
@@ -1177,18 +1254,31 @@ static void output(ErlDrvData drv_data, char* buf, ErlDrvSizeT len)
     lbp = lb + (4-(data->packet_bytes));
 
     if ((sz = driver_sizeq(data->port_num)) > 0) {
-	driver_enq(data->port_num, lbp, data->packet_bytes);
-	driver_enq(data->port_num, buf, len);
-	if (sz + len + data->packet_bytes >= (1 << 13))
+       if (data->packet_bytes != 0) {
+          driver_enq(data->port_num, lbp, data->packet_bytes);
+       }
+       driver_enq(data->port_num, buf, len);
+
+       if (sz + len + data->packet_bytes >= (1 << 13))
 	    set_busy_port(data->port_num, 1);
     }
     else {
-       driver_enq(data->port_num, buf, len);  /* n is the skip value */
-
+       char *pbbuf;
+       if (data->packet_bytes != 0) {
+          pbbuf = malloc(len + data->packet_bytes);
+          int i;
+          for (i = 0; i < data->packet_bytes; i++) {
+             *pbbuf++ = *lbp++;
+          }
+          strncpy(pbbuf, buf, len);
+          pbbuf -= data->packet_bytes;
+       }
        driver_select(data->port_num, data->output_event,
 		      ERL_DRV_WRITE|ERL_DRV_USE, 1);
-
-       WRITE_AIO(data->ofd, len, buf);
+       WRITE_AIO(data->ofd, 
+             (data->packet_bytes ? len+data->packet_bytes : len), 
+             (data->packet_bytes ? pbbuf : buf));
+       if (data->packet_bytes != 0) free(pbbuf);
     }
     return; /* 0; */
 }
@@ -1204,12 +1294,12 @@ static int port_inp_failure(ErlDrvPort port_num, ErlDrvEvent ready_fd, int res)
     ASSERT(res <= 0);
 
     erl_drv_ose_event_fetch(ready_fd,&sig_no, NULL, (void **)&fd);
-
     /* As we need to handle two signals, we do this in two steps */
     if (driver_data[*fd].alive) {
        report_exit_status(driver_data[*fd].report_exit, 0); /* status? */
     }
     else {
+       driver_select(port_num,ready_fd,DO_READ|DO_WRITE,0);
        clear_fd_data(*fd);
        driver_report_exit(driver_data[*fd].port_num, driver_data[*fd].status);
        /* As we do not really know if the spawn has crashed or exited nicely
@@ -1248,6 +1338,10 @@ static void ready_input(ErlDrvData drv_data, ErlDrvEvent ready_fd)
        }
        else {
           res = sig->fm_read_reply.actual;
+          if (res == 0) {
+             port_inp_failure(data->port_num, ready_fd, res);
+             break;
+          }
 
           if (data->packet_bytes == 0) {
              if (res < 0) {
@@ -1258,6 +1352,7 @@ static void ready_input(ErlDrvData drv_data, ErlDrvEvent ready_fd)
              else if (res == 0) {
                 /* read of 0 bytes, eof, otherside of pipe is assumed dead */
                 port_inp_failure(data->port_num, ready_fd, res);
+                break;
              }
              else {
                 buf = driver_alloc(res);
@@ -1267,100 +1362,91 @@ static void ready_input(ErlDrvData drv_data, ErlDrvEvent ready_fd)
                 driver_output(data->port_num, (char*) buf, res);
                 driver_free(buf);
              }
+                /* clear the previous read */
+                memset(data->aiocb.aio_buf, 0, res);
+
+                /* issue a new read */
+                DISPATCH_AIO(sig);
+                aio_read(&data->aiocb);
           }
-          /* We try to read the remainder */
-          else if (fd_data[data->ifd].remain > 0) {
-             if (res < 0) {
-                if ((errno != EINTR) && (errno != ERRNO_BLOCK)) {
-                   port_inp_failure(data->port_num, ready_fd, res);
+          else if (data->packet_bytes && fd_data[data->ifd].remain > 0) {
+             /* we've read a partial package, or a header */
+
+             if (res == fd_data[data->ifd].remain) { /* we are done! */
+                char *buf = data->aiocb.aio_buf;
+                int i;
+
+                /* do we have anything buffered? */
+                if (fd_data[data->ifd].buf != NULL) {
+                   memcpy(fd_data[data->ifd].buf + fd_data[data->ifd].sz,
+                          buf, res);
+                   buf = fd_data[data->ifd].buf;
                 }
-             }
-             else if (res == 0) {
-                port_inp_failure(data->port_num, ready_fd, res);
-             }
-             else if (res == fd_data[data->ifd].remain) { /* we're done  */
-                driver_output(data->port_num,
-                              fd_data[data->ifd].buf,
-                              fd_data[data->ifd].sz);
+
+                fd_data[data->ifd].sz += res;
+                driver_output(data->port_num, buf, (fd_data[data->ifd].sz>0?fd_data[data->ifd].sz:res));
                 clear_fd_data(data->ifd);
-             }
-             else { /*  if (res < fd_data[fd].remain) */
-                fd_data[data->ifd].cpos += res;
-                fd_data[data->ifd].remain -= res;
-             }
-          }
-          else if (fd_data[data->ifd].remain == 0) { /* clean fd */
-             if (res < 0) {
-                if ((errno != EINTR) && (errno != ERRNO_BLOCK)) {
-                   port_inp_failure(data->port_num, ready_fd, res);
+
+                /* clear the previous read */
+                memset(data->aiocb.aio_buf, 0, res);
+             
+                /* issue a new read */
+                DISPATCH_AIO(sig);
+                data->aiocb.aio_nbytes = data->packet_bytes;
+
+                if (data->aiocb.aio_buf == NULL) {
+                   port_inp_failure(data->port_num, ready_fd, -1);
                 }
+                aio_read(&data->aiocb);
              }
-             else if (res == 0) {	/* eof */
-                port_inp_failure(data->port_num, ready_fd, res);
-             }
-             else if (res < data->packet_bytes - fd_data[data->ifd].psz) {
-                memcpy(fd_data[data->ifd].pbuf+fd_data[data->ifd].psz,
-                      (void *)data->aiocb.aio_buf, res);
-                fd_data[data->ifd].psz += res;
-             }
-             else  { /* if (res >= packet_bytes) */
-                unsigned char* cpos = (unsigned char*)data->aiocb.aio_buf;
-                int bytes_left = res;
-
-                while (1) {
-                   int psz = fd_data[data->ifd].psz;
-                   char* pbp = fd_data[data->ifd].pbuf + psz;
-
-                   while (bytes_left && (psz < data->packet_bytes)) {
-                      *pbp++ = *cpos++;
-                      bytes_left--;
-                      psz++;
-                   }
-
-                   if (psz < data->packet_bytes) {
-                      fd_data[data->ifd].psz = psz;
-                      break;
-                   }
-                   fd_data[data->ifd].psz = 0;
-
-                   switch (data->packet_bytes) {
-                      case 1: h = get_int8(fd_data[data->ifd].pbuf);  break;
-                      case 2: h = get_int16(fd_data[data->ifd].pbuf); break;
-                      case 4: h = get_int32(fd_data[data->ifd].pbuf); break;
-                      default: ASSERT(0); return; /* -1; */
-                   }
-
-                   if (h <= (bytes_left)) {
-                      driver_output(data->port_num, (char*) cpos, h);
-                      cpos += h;
-                      bytes_left -= h;
-                      continue;
-                   }
-                   else {		/* The last message we got was split */
-                      char *buf = erts_alloc_fnf(ERTS_ALC_T_FD_ENTRY_BUF, h);
-                      if (!buf) {
-                         errno = ENOMEM;
-                         port_inp_failure(data->port_num, ready_fd, -1);
-                      }
-                      else {
-                         erts_smp_atomic_add_nob(&sys_misc_mem_sz, h);
-                         sys_memcpy(buf, cpos, bytes_left);
-                         fd_data[data->ifd].buf = buf;
-                         fd_data[data->ifd].sz = h;
-                         fd_data[data->ifd].remain = h - bytes_left;
-                         fd_data[data->ifd].cpos = buf + bytes_left;
-                      }
-                      break;
-                   }
+             else if(res < fd_data[data->ifd].remain) { /* received part of a package */
+                if (fd_data[data->ifd].sz == 0) {
+
+                   fd_data[data->ifd].sz += res;
+                   memcpy(fd_data[data->ifd].buf, data->aiocb.aio_buf, res);
+                   fd_data[data->ifd].remain -= res;
+                }
+                else {
+                   memcpy(fd_data[data->ifd].buf + fd_data[data->ifd].sz,
+                          data->aiocb.aio_buf, res);
+                   fd_data[data->ifd].sz += res;
+                   fd_data[data->ifd].remain -= res;
+                }
+                /* clear the previous read */
+                memset(data->aiocb.aio_buf, 0, res);
+
+                /* issue a new read */
+                DISPATCH_AIO(sig);
+                data->aiocb.aio_nbytes = fd_data[data->ifd].remain;
+
+                if (data->aiocb.aio_buf == NULL) {
+                    port_inp_failure(data->port_num, ready_fd, -1);
                 }
+                aio_read(&data->aiocb);
              }
           }
+          else if (data->packet_bytes && fd_data[data->ifd].remain == 0) { /* we've recieved a header */
+
+             /* analyze the header FIXME  */
+             switch (data->packet_bytes) {
+                case 1: h = get_int8(data->aiocb.aio_buf);  break;
+                case 2: h = get_int16(data->aiocb.aio_buf); break;
+                case 4: h = get_int32(data->aiocb.aio_buf); break;
+             }
 
-          /* reset the read buffer and init next asynch read */
-          DISPATCH_AIO(sig);
-          memset((void *)data->aiocb.aio_buf, 0, 255);
+             fd_data[data->ifd].buf = erts_alloc_fnf(ERTS_ALC_T_FD_ENTRY_BUF, h + data->packet_bytes);
+             fd_data[data->ifd].remain = ((h + data->packet_bytes) - res);
 
-          if (res > 0) {
+             /* clear the previous read */
+             memset(data->aiocb.aio_buf, 0, data->packet_bytes);
+
+             /* issue a new read */
+             DISPATCH_AIO(sig);
+             data->aiocb.aio_nbytes = h;
+
+             if (data->aiocb.aio_buf == NULL) {
+                port_inp_failure(data->port_num, ready_fd, -1);
+             }
              aio_read(&data->aiocb);
           }
        }
@@ -1400,11 +1486,16 @@ static void ready_output(ErlDrvData drv_data, ErlDrvEvent ready_fd)
              DISPATCH_AIO(sig);
              FREE_AIO(sig->fm_write_reply.buffer);
              res = driver_deq(data->port_num, iov[0].iov_len);
-             if (res > 0) {
+             if (res > 0) { 
                 iov = driver_peekq(data->port_num, &vlen);
                 WRITE_AIO(data->ofd, iov[0].iov_len, iov[0].iov_base);
              }
          }
+         else if (vlen == 0) {
+            DISPATCH_AIO(sig);
+            FREE_AIO(sig->fm_write_reply.buffer);
+         }
+
       }
       sig = erl_drv_ose_get_signal(ready_fd);
    }
@@ -1650,10 +1741,10 @@ erl_assert_error(const char* expr, const char* func,
 {
     fflush(stdout);
     fprintf(stderr, "%s:%d:%s() Assertion failed: %s\n",
-	    file, func, line, expr);
+	    file, line, func, expr);
     fflush(stderr);
     ramlog_printf("%s:%d:%s() Assertion failed: %s\n",
-		  file, func, line, expr);
+		  file, line, func, expr);
 
     abort();
 }