Implement file:sendfile

Allow Erlang code to use sendfile() where available by wrapping it as
file:sendfile/4 and file:sendfile/2.

sendfile(2) - Linux man page:
"sendfile() copies data between one file descriptor and another.
Because this copying is done within the kernel, sendfile() is more
efficient than the combination of read(2) and write(2), which would
require transferring data to and from user space."

1 files changed, 232 insertions, 2 deletions

diff --git a/erts/emulator/drivers/common/efile_drv.c b/erts/emulator/drivers/common/efile_drv.c
index 901d98c09d..509c4fe48c 100644
--- a/erts/emulator/drivers/common/efile_drv.c
+++ b/erts/emulator/drivers/common/efile_drv.c
@@ -55,6 +55,7 @@
 #define FILE_READ_LINE          29
 #define FILE_FDATASYNC          30
 #define FILE_FADVISE            31
+#define FILE_SENDFILE           32
 
 /* Return codes */
 
@@ -217,6 +218,7 @@ typedef unsigned char uchar;
 static ErlDrvData file_start(ErlDrvPort port, char* command);
 static int file_init(void);
 static void file_stop(ErlDrvData);
+static void file_ready_output(ErlDrvData data, ErlDrvEvent event);
 static void file_output(ErlDrvData, char* buf, int len);
 static int file_control(ErlDrvData, unsigned int command, 
 			char* buf, int len, char **rbuf, int rlen);
@@ -224,6 +226,7 @@ static void file_timeout(ErlDrvData);
 static void file_outputv(ErlDrvData, ErlIOVec*);
 static void file_async_ready(ErlDrvData, ErlDrvThreadData);
 static void file_flush(ErlDrvData);
+static void file_stop_select(ErlDrvEvent event, void* _);
 
 
 
@@ -253,6 +256,18 @@ typedef struct {
     ErlDrvPDL       q_mtx;    /* Mutex for the driver queue, known by the emulator. Also used for
 				 mutual exclusion when accessing field(s) below. */
     size_t          write_buffered;
+    ErlDrvTermData caller;      /* recipient of sync reply */
+    /* sendfile call state to retry/resume on event */
+    int command; /* same as d->command. for sendfile. TODO: this seems wrong */
+    struct {
+	int eagain;
+	int out_fd;
+	/* TODO: Use Sint64 instead? What about 32-bit off_t linux */
+	off_t offset;
+	size_t count;
+	size_t chunksize;
+	ErlDrvSInt64 written;
+    } sendfile;
 } file_descriptor;
 
 
@@ -264,7 +279,7 @@ struct erl_drv_entry efile_driver_entry = {
     file_stop,
     file_output,
     NULL,
-    NULL,
+    file_ready_output,
     "efile",
     NULL,
     NULL,
@@ -279,7 +294,9 @@ struct erl_drv_entry efile_driver_entry = {
     ERL_DRV_EXTENDED_MAJOR_VERSION,
     ERL_DRV_EXTENDED_MINOR_VERSION,
     ERL_DRV_FLAG_USE_PORT_LOCKING,
-    NULL
+    NULL,
+    NULL,
+    file_stop_select
 };
 
 
@@ -613,6 +630,111 @@ static struct t_data *cq_deq(file_descriptor *desc) {
 }
 
 
+/*********************************************************************
+ * 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
@@ -628,6 +750,11 @@ file_init(void)
 			    ? atoi(buf)
 			    : 0);
     driver_system_info(&sys_info, sizeof(ErlDrvSysInfo));
+
+    INIT_ATOM(ok);
+    INIT_ATOM(error);
+    INIT_ATOM(efile_reply);
+
     return 0;
 }
 
@@ -1694,6 +1821,74 @@ static void invoke_fadvise(void *data)
     d->result_ok = efile_fadvise(&d->errInfo, fd, offset, length, advise);
 }
 
+
+
+static void do_sendfile(file_descriptor *desc);
+static void file_ready_output(ErlDrvData data, ErlDrvEvent event)
+{
+    file_descriptor* d = (file_descriptor*) data;
+
+    switch (d->command) {
+    case FILE_SENDFILE:
+	driver_select(d->port, (ErlDrvEvent)d->sendfile.out_fd,
+		      ERL_DRV_WRITE, 0);
+	do_sendfile(d);
+	break;
+    default:
+	break;
+    }
+}
+
+static void file_stop_select(ErlDrvEvent event, void* _)
+{
+    /* TODO: close socket? */
+}
+
+static void invoke_sendfile(void *data)
+{
+    ((struct t_data *)data)->again = 0;
+}
+
+static void do_sendfile(file_descriptor *d)
+{
+    int fd = d->fd;
+    int out_fd = d->sendfile.out_fd;
+    off_t offset = d->sendfile.offset;
+    size_t count = d->sendfile.count;
+    size_t chunksize = count < d->sendfile.chunksize
+	? count : d->sendfile.chunksize;
+    int result_ok = 0;
+    Efile_error errInfo;
+
+    result_ok = efile_sendfile(&errInfo, fd, out_fd, &offset, &chunksize);
+
+    if (result_ok) {
+	d->sendfile.offset += chunksize;
+	d->sendfile.written += chunksize;
+	d->sendfile.count -= chunksize;
+	if (d->sendfile.count > 0) {
+	    driver_select(d->port, (ErlDrvEvent)d->sendfile.out_fd,
+			  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);
+	}
+    } else if (errInfo.posix_errno == EAGAIN || errInfo.posix_errno == EINTR) {
+	if (chunksize > 0) {
+	    d->sendfile.offset += chunksize;
+	    d->sendfile.written += chunksize;
+	    d->sendfile.count -= chunksize;
+	}
+	d->sendfile.eagain++;
+
+	driver_select(d->port, (ErlDrvEvent)d->sendfile.out_fd,
+		      ERL_DRV_USE|ERL_DRV_WRITE, 1);
+    } else {
+	printf("==> sendfile ERROR %s\n", erl_errno_id(errInfo.posix_errno));
+	ef_send_posix_error(d, d->caller, errInfo.posix_errno);
+    }
+}
+
 static void free_readdir(void *data)
 {
     struct t_data *d = (struct t_data *) data;
@@ -2105,6 +2300,13 @@ 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);
+	  break;
       default:
 	abort();
     }
@@ -2452,6 +2654,34 @@ 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;
+	}
+
     }
 
     /*