Implement sendfile using blocking io in asynch threads

Move the command handling to outputv in preparation for
header and trailer inclusion in the sendfile api.

Use the standard efile communication functions for sendfile.

4 files changed, 54 insertions, 179 deletions

diff --git a/erts/emulator/drivers/common/efile_drv.c b/erts/emulator/drivers/common/efile_drv.c
index 509c4fe48c..7e194a3787 100644
--- a/erts/emulator/drivers/common/efile_drv.c
+++ b/erts/emulator/drivers/common/efile_drv.c
@@ -629,113 +629,6 @@ static struct t_data *cq_deq(file_descriptor *desc) {
     return d;
 }
 
-
-/*********************************************************************
- * Command queue functions
- */
-
-static ErlDrvTermData am_ok;
-static ErlDrvTermData am_error;
-static ErlDrvTermData am_efile_reply;
-
-#define INIT_ATOM(NAME) am_ ## NAME = driver_mk_atom(#NAME)
-
-#define LOAD_ATOM_CNT 2
-#define LOAD_ATOM(vec, i, atom) \
-  (((vec)[(i)] = ERL_DRV_ATOM), \
-  ((vec)[(i)+1] = (atom)), \
-  ((i)+LOAD_ATOM_CNT))
-
-#define LOAD_INT_CNT 2
-#define LOAD_INT(vec, i, val) \
-  (((vec)[(i)] = ERL_DRV_INT), \
-  ((vec)[(i)+1] = (ErlDrvTermData)(val)), \
-  ((i)+LOAD_INT_CNT))
-
-#define LOAD_INT64_CNT 2
-#define LOAD_INT64(vec, i, val) \
-  (((vec)[(i)] = ERL_DRV_INT64), \
-  ((vec)[(i)+1] = (ErlDrvTermData)(val)), \
-  ((i)+LOAD_INT64_CNT))
-
-#define LOAD_PORT_CNT 2
-#define LOAD_PORT(vec, i, port) \
-  (((vec)[(i)] = ERL_DRV_PORT), \
-  ((vec)[(i)+1] = (port)), \
-  ((i)+LOAD_PORT_CNT))
-
-#define LOAD_PID_CNT 2
-#define LOAD_PID(vec, i, pid) \
-  (((vec)[(i)] = ERL_DRV_PID), \
-  ((vec)[(i)+1] = (pid)), \
-  ((i)+LOAD_PID_CNT))
-
-#define LOAD_TUPLE_CNT 2
-#define LOAD_TUPLE(vec, i, size) \
-  (((vec)[(i)] = ERL_DRV_TUPLE), \
-  ((vec)[(i)+1] = (size)), \
-  ((i)+LOAD_TUPLE_CNT))
-
-/* send:
-**   {efile_reply, Pid, Port, {ok, int64()}}
-*/
-
-static int ef_send_ok_int64(file_descriptor *desc, ErlDrvTermData caller,
-			    ErlDrvSInt64 *n)
-{
-    ErlDrvTermData spec[2*LOAD_ATOM_CNT + LOAD_PID_CNT + LOAD_PORT_CNT
-			+ LOAD_INT64_CNT + 2*LOAD_TUPLE_CNT];
-    int i = 0;
-
-    i = LOAD_ATOM(spec, i, am_efile_reply);
-    i = LOAD_PID(spec, i, caller);
-    i = LOAD_PORT(spec, i, driver_mk_port(desc->port));
-    i = LOAD_ATOM(spec, i, am_ok);
-    i = LOAD_INT64(spec, i, n);
-    i = LOAD_TUPLE(spec, i, 2);
-    i = LOAD_TUPLE(spec, i, 4);
-    ASSERT(i == sizeof(spec)/sizeof(*spec));
-
-    return driver_send_term(desc->port, caller, spec, i);
-}
-
-static ErlDrvTermData error_atom(int err)
-{
-    char errstr[256];
-    char* s;
-    char* t;
-
-    for (s = erl_errno_id(err), t = errstr; *s; s++, t++)
-	*t = tolower(*s);
-    *t = '\0';
-    return driver_mk_atom(errstr);
-}
-
-/* send:
-**   {efile_reply, Pid, Port, {error, posix_error()}
-*/
-
-static int ef_send_posix_error(file_descriptor *desc, ErlDrvTermData caller,
-			       int e)
-{
-    ErlDrvTermData spec[3*LOAD_ATOM_CNT + LOAD_PID_CNT + LOAD_PORT_CNT
-			+ 2*LOAD_TUPLE_CNT];
-    int i = 0;
-
-    i = LOAD_ATOM(spec, i, am_efile_reply);
-    i = LOAD_PID(spec, i, caller);
-    i = LOAD_PORT(spec, i, driver_mk_port(desc->port));
-    i = LOAD_ATOM(spec, i, am_error);
-    /* TODO: safe? set of error codes should be limited and safe */
-    i = LOAD_ATOM(spec, i, error_atom(e));
-    i = LOAD_TUPLE(spec, i, 2);
-    i = LOAD_TUPLE(spec, i, 4);
-    ASSERT(i == sizeof(spec)/sizeof(*spec));
-
-    desc->caller = 0;
-    return driver_send_term(desc->port, caller, spec, i);
-}
-
 /*********************************************************************
  * Driver entry point -> init
  */
@@ -751,10 +644,6 @@ file_init(void)
 			    : 0);
     driver_system_info(&sys_info, sizeof(ErlDrvSysInfo));
 
-    INIT_ATOM(ok);
-    INIT_ATOM(error);
-    INIT_ATOM(efile_reply);
-
     return 0;
 }
 
@@ -1871,7 +1760,7 @@ static void do_sendfile(file_descriptor *d)
 			  ERL_DRV_USE|ERL_DRV_WRITE, 1);
 	} else {
 	    printf("==> sendfile DONE eagain=%d\n", d->sendfile.eagain);
-	    ef_send_ok_int64(d, d->caller, &d->sendfile.written);
+	    reply_Uint(d, d->sendfile.written);
 	}
     } else if (errInfo.posix_errno == EAGAIN || errInfo.posix_errno == EINTR) {
 	if (chunksize > 0) {
@@ -2301,8 +2190,6 @@ file_async_ready(ErlDrvData e, ErlDrvThreadData data)
 	  free_preadv(data);
 	  break;
       case FILE_SENDFILE:
-	  /* Return 'ok' and let prim_file:sendfile wait for message */
-	  reply_ok(desc);
 	  driver_select(desc->port, (ErlDrvEvent)desc->sendfile.out_fd,
 			ERL_DRV_USE|ERL_DRV_WRITE, 1);
 	  free_data(data);
@@ -2654,34 +2541,6 @@ file_output(ErlDrvData e, char* buf, int count)
         goto done;
     }
 
-    case FILE_SENDFILE:
-	{
-	    d = EF_SAFE_ALLOC(sizeof(struct t_data));
-	    d->fd = fd;
-	    d->command = command;
-	    d->invoke = invoke_sendfile;
-	    d->free = free_data;
-	    d->level = 2;
-	    desc->sendfile.out_fd = get_int32((uchar*) buf);
-	    /* TODO: are off_t and size_t 64bit on all platforms?
-	       off_t is 32bit on win32 msvc. maybe configurable in msvc.
-	       Maybe use '#if SIZEOF_SIZE_T == 4'? */
-	    desc->sendfile.offset = get_int64(((uchar*) buf)
-					     + sizeof(Sint32));
-	    desc->sendfile.count = get_int64(((uchar*) buf)
-					    + sizeof(Sint32)
-					    + sizeof(Sint64));
-	    desc->sendfile.chunksize = get_int64(((uchar*) buf)
-					    + sizeof(Sint32)
-					    + 2*sizeof(Sint64));
-	    desc->sendfile.written = 0;
-	    desc->sendfile.eagain = 0;
-	    /* TODO: shouldn't d->command be enough? */
-	    desc->command = command;
-	    desc->caller = driver_caller(desc->port);
-	    goto done;
-	}
-
     }
 
     /*
@@ -3475,6 +3334,34 @@ file_outputv(ErlDrvData e, ErlIOVec *ev) {
 	    goto done;
 	} /* case FILE_OPT_DELAYED_WRITE: */
     } ASSERT(0); goto done; /* case FILE_SETOPT: */
+    case FILE_SENDFILE:
+	{
+	struct t_data *d;
+	    d = EF_SAFE_ALLOC(sizeof(struct t_data));
+	    d->fd = desc->fd;
+	    d->command = command;
+	    d->invoke = invoke_sendfile;
+	    d->free = free_data;
+	    d->level = 2;
+	    desc->sendfile.out_fd = get_int32((uchar*) buf);
+	    /* TODO: are off_t and size_t 64bit on all platforms?
+	       off_t is 32bit on win32 msvc. maybe configurable in msvc.
+	       Maybe use '#if SIZEOF_SIZE_T == 4'? */
+	    desc->sendfile.offset = get_int64(((uchar*) buf)
+					     + sizeof(Sint32));
+	    desc->sendfile.count = get_int64(((uchar*) buf)
+					    + sizeof(Sint32)
+					    + sizeof(Sint64));
+	    desc->sendfile.chunksize = get_int64(((uchar*) buf)
+					    + sizeof(Sint32)
+					    + 2*sizeof(Sint64));
+	    desc->sendfile.written = 0;
+	    desc->sendfile.eagain = 0;
+	    /* TODO: shouldn't d->command be enough? */
+	    desc->command = command;
+	    desc->caller = driver_caller(desc->port);
+	    goto done;
+	}
     
     } /* switch(command) */
     
diff --git a/erts/emulator/drivers/common/erl_efile.h b/erts/emulator/drivers/common/erl_efile.h
index 3c6c2ec2db..864b867955 100644
--- a/erts/emulator/drivers/common/erl_efile.h
+++ b/erts/emulator/drivers/common/erl_efile.h
@@ -162,5 +162,5 @@ int efile_symlink(Efile_error* errInfo, char* old, char* new);
 int efile_may_openfile(Efile_error* errInfo, char *name);
 int efile_fadvise(Efile_error* errInfo, int fd, Sint64 offset, Sint64 length,
 		  int advise);
-int efile_sendfile(Efile_error* errInfo, int in_fd, int out_fd, off_t *offset,
+int efile_sendfile(Efile_error* errInfo, int in_fd, int out_fd, off_t offset,
 		   size_t *count);
diff --git a/erts/emulator/drivers/unix/unix_efile.c b/erts/emulator/drivers/unix/unix_efile.c
index 5b001b3819..8b612164da 100644
--- a/erts/emulator/drivers/unix/unix_efile.c
+++ b/erts/emulator/drivers/unix/unix_efile.c
@@ -1471,31 +1471,24 @@ efile_fadvise(Efile_error* errInfo, int fd, Sint64 offset,
 #ifdef HAVE_SENDFILE
 int
 efile_sendfile(Efile_error* errInfo, int in_fd, int out_fd,
-	       off_t *offset, size_t *count)
+	       off_t offset, size_t *ret_nbytes)
 {
 #if defined(__linux__) || (defined(__sun) && defined(__SVR4))
-    ssize_t retval = sendfile(out_fd, in_fd, offset, *count);
-    if (retval >= 0) {
-	if (retval != *count) {
-	    *count = retval;
-	    retval = -1;
-	    errno = EAGAIN;
-	} else {
-	    *count = retval;
-	}
-    } else if (retval == -1 && (errno == EINTR || errno == EAGAIN)) {
-	*count = 0;
-    }
+    ssize_t retval;
+    do {
+	retval = sendfile(out_fd, in_fd, &offset, *ret_nbytes);
+    } while (retval == -1 && (errno == EINTR || errno == EAGAIN));
+    *ret_nbytes = retval;
     return check_error(retval == -1 ? -1 : 0, errInfo);
 #elif defined(DARWIN)
-    off_t len = *count;
+    off_t len = *ret_nbytes;
     int retval = sendfile(in_fd, out_fd, *offset, &len, NULL, 0);
-    *count = len;
+    *ret_nbytes = len;
     return check_error(retval, errInfo);
 #elif defined(__FreeBSD__) || defined(__DragonFly__)
     off_t len = 0;
-    int retval = sendfile(in_fd, out_fd, *offset, *count, NULL, &len, 0);
-    *count = len;
+    int retval = sendfile(in_fd, out_fd, *offset, *nbytes, NULL, &len, 0);
+    *nbytes = len;
     return check_error(retval, errInfo);
 #endif
 }
diff --git a/erts/preloaded/src/prim_file.erl b/erts/preloaded/src/prim_file.erl
index 606d7d5aab..0767067682 100644
--- a/erts/preloaded/src/prim_file.erl
+++ b/erts/preloaded/src/prim_file.erl
@@ -539,30 +539,22 @@ write_file(File, Bin) when (is_list(File) orelse is_binary(File)) ->
     end;
 write_file(_, _) -> 
     {error, badarg}.
-    
+
 
 %% Returns {error, Reason} | {ok, BytesCopied}
-sendfile(_,_,_,_,_,_,_,_,_,_) ->
-    {error, enotsup};
+%sendfile(_,_,_,_,_,_,_,_,_,_) ->
+%    {error, enotsup};
 sendfile(#file_descriptor{module = ?MODULE, data = {Port, _}},
 	 DestFD, Offset, Bytes, ChunkSize, Headers, Trailers,
 	 Nodiskio, MNowait, Sync) ->
 
-    ok = drv_command(Port, <<?FILE_SENDFILE, DestFD:32, Offset:64, Bytes:64,
-			     ChunkSize:64,
-			     (get_bit(Nodiskio)):1,
-			     (get_bit(MNowait)):1,
-			     (get_bit(Sync)):1,0:5,
-			     (encode_hdtl(Headers))/binary,
-			     (encode_hdtl(Trailers))/binary>>),
-    Self = self(),
-    %% Should we use a ref()?
-    receive
-	{efile_reply, Self, Port, {ok, _Written}=OKRes}->
-	    OKRes;
-	{efile_reply, Self, Port, {error, _PosixError}=Error}->
-	    Error
-    end.
+    drv_command(Port, <<?FILE_SENDFILE, DestFD:32, Offset:64, Bytes:64,
+			ChunkSize:64,
+			(get_bit(Nodiskio)):1,
+			(get_bit(MNowait)):1,
+			(get_bit(Sync)):1,0:5,
+			(encode_hdtl(Headers))/binary,
+			(encode_hdtl(Trailers))/binary>>).
 
 get_bit(true) ->
     1;
@@ -578,8 +570,11 @@ encode_hdtl(List) ->
 
 encode_hdtl([], Acc, Cnt) ->
     <<Cnt:8, Acc/binary>>;
+encode_hdtl([Bin|T], Acc, Cnt) when is_binary(Bin) ->
+    encode_hdtl(T, <<(byte_size(Bin)):32, Bin/binary, Acc/binary>>,Cnt + 1);
 encode_hdtl([Bin|T], Acc, Cnt) ->
-    encode_hdtl(T, <<(byte_size(Bin)):32, Bin/binary, Acc/binary>>,Cnt + 1).
+    encode_hdtl(T, <<(iolist_size(Bin)):32, (iolist_to_binary(Bin))/binary,
+		     Acc/binary>>,Cnt + 1).