diff options
author | Lukas Larsson <lukas@erlang-solutions.com> | 2011-12-02 15:08:47 +0100 |
---|---|---|
committer | Lukas Larsson <lukas@erlang-solutions.com> | 2011-12-02 15:08:47 +0100 |
commit | f30ac8e5a84c550734b79a6d66639c4d3489c6fd (patch) | |
tree | 331a81551b21d75c41b5edc25b9077bfe2a8cf70 | |
parent | 7bd9c1f9a68a024958040fe5b77dacc73bb6c5ab (diff) | |
parent | 62fffa75e2003b3f19eb7614307942028a400fd1 (diff) | |
download | otp-f30ac8e5a84c550734b79a6d66639c4d3489c6fd.tar.gz otp-f30ac8e5a84c550734b79a6d66639c4d3489c6fd.tar.bz2 otp-f30ac8e5a84c550734b79a6d66639c4d3489c6fd.zip |
Merge branch 'ta/sendfile/OTP-9240'
* ta/sendfile/OTP-9240: (31 commits)
Add sendfile server printouts
Skip recv/send during tests for fallback platforms
Remove header/trailer support
Remove windows implementation
Expand sendfile documentation
Only allow tcp sockets as target for sendfile
Move sendfile api to file module
Preliminary work on header/trailer
Use free_sendfile explicitly for non-async
Remove debug printouts
Add tests for send/recv/sendfile interactions
Remove tests for file_server sendfile
sendfile caller now has to be the controlling_process
Remove support for file_server, sendfile has to be raw
Set chunk size to 3 GB
Change type of fd to be ErlDrvEvent
Add ifdef's for HAVE_SENDFILE
Fix freebsd support for sendfile
Change nbytes to 64 bit
Implement ignorefd for TCP
...
-rw-r--r-- | erts/configure.in | 16 | ||||
-rw-r--r-- | erts/emulator/drivers/common/efile_drv.c | 202 | ||||
-rw-r--r-- | erts/emulator/drivers/common/erl_efile.h | 17 | ||||
-rw-r--r-- | erts/emulator/drivers/common/inet_drv.c | 67 | ||||
-rw-r--r-- | erts/emulator/drivers/unix/unix_efile.c | 77 | ||||
-rw-r--r-- | erts/preloaded/src/prim_file.erl | 30 | ||||
-rw-r--r-- | erts/preloaded/src/prim_inet.erl | 18 | ||||
-rw-r--r-- | lib/kernel/doc/src/file.xml | 45 | ||||
-rw-r--r-- | lib/kernel/src/file.erl | 139 | ||||
-rw-r--r-- | lib/kernel/src/gen_tcp.erl | 4 | ||||
-rw-r--r-- | lib/kernel/src/inet.erl | 15 | ||||
-rw-r--r-- | lib/kernel/src/inet_int.hrl | 2 | ||||
-rw-r--r-- | lib/kernel/test/Makefile | 3 | ||||
-rw-r--r-- | lib/kernel/test/gen_tcp_api_SUITE.erl | 10 | ||||
-rw-r--r-- | lib/kernel/test/sendfile_SUITE.erl | 278 |
15 files changed, 899 insertions, 24 deletions
diff --git a/erts/configure.in b/erts/configure.in index 378d0ab220..10d303e241 100644 --- a/erts/configure.in +++ b/erts/configure.in @@ -1742,6 +1742,22 @@ dnl fdatasync requires linking against -lrt on SunOS <= 5.10. dnl OpenSolaris 2009.06 is SunOS 5.11 and does not require -lrt. AC_SEARCH_LIBS(fdatasync, [rt]) + +dnl sendfile syscall +case $host_os in + linux*|freebsd*|dragonfly*|darwin*) + AC_CHECK_FUNCS([sendfile]) + ;; + solaris*) + AC_SEARCH_LIBS(sendfile, sendfile, AC_DEFINE(HAVE_SENDFILE, 1)) + ;; + win32) + LIBS="$LIBS -lmswsock" + ;; + *) + ;; +esac + dnl ---------------------------------------------------------------------- dnl Checks for library functions. dnl ---------------------------------------------------------------------- diff --git a/erts/emulator/drivers/common/efile_drv.c b/erts/emulator/drivers/common/efile_drv.c index 901d98c09d..5c52b99348 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 */ @@ -98,7 +99,13 @@ # include "config.h" #endif #include <stdlib.h> + +// Need (NON)BLOCKING macros for sendfile +#ifndef WANT_NONBLOCKING +#define WANT_NONBLOCKING +#endif #include "sys.h" + #include "erl_driver.h" #include "erl_efile.h" #include "erl_threads.h" @@ -225,9 +232,16 @@ static void file_outputv(ErlDrvData, ErlIOVec*); static void file_async_ready(ErlDrvData, ErlDrvThreadData); static void file_flush(ErlDrvData); +#ifdef HAVE_SENDFILE +static void file_ready_output(ErlDrvData data, ErlDrvEvent event); +static void file_stop_select(ErlDrvEvent event, void* _); +#endif /* HAVE_SENDFILE */ enum e_timer {timer_idle, timer_again, timer_write}; +#ifdef HAVE_SENDFILE +enum e_sendfile {sending, not_sending}; +#endif /* HAVE_SENDFILE */ struct t_data; @@ -242,6 +256,9 @@ typedef struct { struct t_data *cq_head; /* Queue of incoming commands */ struct t_data *cq_tail; /* -""- */ enum e_timer timer_state; +#ifdef HAVE_SENDFILE + enum e_sendfile sendfile_state; +#endif /* HAVE_SENDFILE */ size_t read_bufsize; ErlDrvBinary *read_binp; size_t read_offset; @@ -264,7 +281,11 @@ struct erl_drv_entry efile_driver_entry = { file_stop, file_output, NULL, +#ifdef HAVE_SENDFILE + file_ready_output, +#else NULL, +#endif /* HAVE_SENDFILE */ "efile", NULL, NULL, @@ -279,7 +300,13 @@ 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, +#ifdef HAVE_SENDFILE + file_stop_select +#else NULL +#endif /* HAVE_SENDFILE */ }; @@ -398,6 +425,14 @@ struct t_data Sint64 length; int advise; } fadvise; +#ifdef HAVE_SENDFILE + struct { + int out_fd; + off_t offset; + Uint64 nbytes; + Uint64 written; + } sendfile; +#endif /* HAVE_SENDFILE */ } c; char b[1]; }; @@ -485,7 +520,6 @@ static void *ef_safe_realloc(void *op, Uint s) : 0) - #if 0 static void ev_clear(ErlIOVec *ev) { @@ -613,7 +647,6 @@ static struct t_data *cq_deq(file_descriptor *desc) { } - /********************************************************************* * Driver entry point -> init */ @@ -628,6 +661,7 @@ file_init(void) ? atoi(buf) : 0); driver_system_info(&sys_info, sizeof(ErlDrvSysInfo)); + return 0; } @@ -655,6 +689,9 @@ file_start(ErlDrvPort port, char* command) desc->cq_head = NULL; desc->cq_tail = NULL; desc->timer_state = timer_idle; +#ifdef HAVE_SENDFILE + desc->sendfile_state = not_sending; +#endif desc->read_bufsize = 0; desc->read_binp = NULL; desc->read_offset = 0; @@ -893,8 +930,6 @@ static int reply_eof(file_descriptor *desc) { driver_output2(desc->port, &c, 1, NULL, 0); return 0; } - - static void invoke_name(void *data, int (*f)(Efile_error *, char *)) { @@ -1694,6 +1729,66 @@ static void invoke_fadvise(void *data) d->result_ok = efile_fadvise(&d->errInfo, fd, offset, length, advise); } +#ifdef HAVE_SENDFILE +static void invoke_sendfile(void *data) +{ + struct t_data *d = (struct t_data *)data; + int fd = d->fd; + int out_fd = (int)d->c.sendfile.out_fd; + Uint64 nbytes = d->c.sendfile.nbytes; + int result = 0; + d->again = 0; + + result = efile_sendfile(&d->errInfo, fd, out_fd, &d->c.sendfile.offset, &nbytes, NULL); + + d->c.sendfile.written += nbytes; + + if (result == 1) { + if (sys_info.async_threads != 0) { + d->result_ok = 0; + } else if (d->c.sendfile.nbytes == 0 && nbytes != 0) { + d->result_ok = 1; + } else if ((d->c.sendfile.nbytes - nbytes) != 0) { + d->result_ok = 1; + d->c.sendfile.nbytes -= nbytes; + } else { + d->result_ok = 0; + } + } else if (result == 0 && (d->errInfo.posix_errno == EAGAIN + || d->errInfo.posix_errno == EINTR)) { + d->result_ok = 1; + } else { + d->result_ok = -1; + } +} + +static void free_sendfile(void *data) { + EF_FREE(data); +} + +static void file_ready_output(ErlDrvData data, ErlDrvEvent event) +{ + file_descriptor* fd = (file_descriptor*) data; + + switch (fd->d->command) { + case FILE_SENDFILE: + driver_select(fd->port, event, + (int)ERL_DRV_WRITE,(int) 0); + invoke_sendfile((void *)fd->d); + file_async_ready(data, (ErlDrvThreadData)fd->d); + break; + default: + break; + } +} + +static void file_stop_select(ErlDrvEvent event, void* _) +{ + +} +#endif /* HAVE_SENDFILE */ + + static void free_readdir(void *data) { struct t_data *d = (struct t_data *) data; @@ -1755,6 +1850,10 @@ static void cq_execute(file_descriptor *desc) { register void *void_ptr; /* Soft cast variable */ if (desc->timer_state == timer_again) return; +#ifdef HAVE_SENDFILE + if (desc->sendfile_state == sending) + return; +#endif if (! (d = cq_deq(desc))) return; TRACE_F(("x%i", (int) d->command)); @@ -2105,6 +2204,37 @@ file_async_ready(ErlDrvData e, ErlDrvThreadData data) } free_preadv(data); break; +#ifdef HAVE_SENDFILE + case FILE_SENDFILE: + if (d->result_ok == -1) { + desc->sendfile_state = not_sending; + reply_error(desc, &d->errInfo); + if (sys_info.async_threads != 0) { + SET_NONBLOCKING(d->c.sendfile.out_fd); + free_sendfile(data); + } else { + driver_select(desc->port, (ErlDrvEvent)(long)d->c.sendfile.out_fd, + ERL_DRV_USE, 0); + free_sendfile(data); + } + } else if (d->result_ok == 0) { + desc->sendfile_state = not_sending; + reply_Sint64(desc, d->c.sendfile.written); + if (sys_info.async_threads != 0) { + SET_NONBLOCKING(d->c.sendfile.out_fd); + free_sendfile(data); + } else { + driver_select(desc->port, (ErlDrvEvent)(long)d->c.sendfile.out_fd, ERL_DRV_USE, 0); + free_sendfile(data); + } + } else if (d->result_ok == 1) { // If we are using select to send the rest of the data + desc->sendfile_state = sending; + desc->d = d; + driver_select(desc->port, (ErlDrvEvent)(long)d->c.sendfile.out_fd, + ERL_DRV_USE|ERL_DRV_WRITE, 1); + } + break; +#endif default: abort(); } @@ -3245,9 +3375,69 @@ file_outputv(ErlDrvData e, ErlIOVec *ev) { goto done; } /* case FILE_OPT_DELAYED_WRITE: */ } ASSERT(0); goto done; /* case FILE_SETOPT: */ - + + case FILE_SENDFILE: { + +#ifdef HAVE_SENDFILE + struct t_data *d; + Uint32 out_fd, offsetH, offsetL, hd_len, tl_len; + Uint64 nbytes; + char flags; + + if (ev->size < 1 + 7 * sizeof(Uint32) + sizeof(char) + || !EV_GET_UINT32(ev, &out_fd, &p, &q) + || !EV_GET_CHAR(ev, &flags, &p, &q) + || !EV_GET_UINT32(ev, &offsetH, &p, &q) + || !EV_GET_UINT32(ev, &offsetL, &p, &q) + || !EV_GET_UINT64(ev, &nbytes, &p, &q) + || !EV_GET_UINT32(ev, &hd_len, &p, &q) + || !EV_GET_UINT32(ev, &tl_len, &p, &q)) { + /* Buffer has wrong length to contain all the needed values */ + reply_posix_error(desc, EINVAL); + goto done; + } + + if (hd_len != 0 || tl_len != 0 || flags != 0) { + // We do not allow header, trailers and/or flags right now + reply_posix_error(desc, EINVAL); + goto done; + } + + d = EF_SAFE_ALLOC(sizeof(struct t_data)); + d->fd = desc->fd; + d->command = command; + d->invoke = invoke_sendfile; + d->free = NULL; + d->level = 2; + + d->c.sendfile.out_fd = (int) out_fd; + d->c.sendfile.written = 0; + + #if SIZEOF_OFF_T == 4 + if (offsetH != 0) { + reply_posix_error(desc, EINVAL); + goto done; + } + d->c.sendfile.offset = (off_t) offsetL; + #else + d->c.sendfile.offset = ((off_t) offsetH << 32) | offsetL; + #endif + + d->c.sendfile.nbytes = nbytes; + + if (sys_info.async_threads != 0) { + SET_BLOCKING(d->c.sendfile.out_fd); + } + + cq_enq(desc, d); +#else + reply_posix_error(desc, ENOTSUP); +#endif + goto done; + } /* case FILE_SENDFILE: */ + } /* switch(command) */ - + if (lseek_flush_read(desc, &err) < 0) { reply_posix_error(desc, err); goto done; diff --git a/erts/emulator/drivers/common/erl_efile.h b/erts/emulator/drivers/common/erl_efile.h index 3097ded3f1..349ab0e17b 100644 --- a/erts/emulator/drivers/common/erl_efile.h +++ b/erts/emulator/drivers/common/erl_efile.h @@ -118,6 +118,19 @@ typedef struct _Efile_info { */ } Efile_info; + +#ifdef HAVE_SENDFILE +/* + * Described the structure of header/trailers for sendfile + */ +struct t_sendfile_hdtl { + SysIOVec *headers; + int hdr_cnt; + SysIOVec *trailers; + int trl_cnt; +}; +#endif /* HAVE_SENDFILE */ + /* * Functions. */ @@ -162,3 +175,7 @@ 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); +#ifdef HAVE_SENDFILE +int efile_sendfile(Efile_error* errInfo, int in_fd, int out_fd, + off_t *offset, Uint64 *nbytes, struct t_sendfile_hdtl *hdtl); +#endif /* HAVE_SENDFILE */ diff --git a/erts/emulator/drivers/common/inet_drv.c b/erts/emulator/drivers/common/inet_drv.c index db052523a8..7d952b0c71 100644 --- a/erts/emulator/drivers/common/inet_drv.c +++ b/erts/emulator/drivers/common/inet_drv.c @@ -445,6 +445,7 @@ static int my_strncasecmp(const char *s1, const char *s2, size_t n) driver_select(port, e, mode | (on?ERL_DRV_USE:0), on) #define sock_select(d, flags, onoff) do { \ + ASSERT(!onoff || !(d)->is_ignored); \ (d)->event_mask = (onoff) ? \ ((d)->event_mask | (flags)) : \ ((d)->event_mask & ~(flags)); \ @@ -538,6 +539,8 @@ static int my_strncasecmp(const char *s1, const char *s2, size_t n) #define INET_REQ_GETIFADDRS 25 #define INET_REQ_ACCEPT 26 #define INET_REQ_LISTEN 27 +#define INET_REQ_IGNOREFD 28 + /* TCP requests */ /* #define TCP_REQ_ACCEPT 40 MOVED */ /* #define TCP_REQ_LISTEN 41 MERGED */ @@ -725,6 +728,11 @@ static int my_strncasecmp(const char *s1, const char *s2, size_t n) /* Max interface name */ #define INET_IFNAMSIZ 16 +/* INET Ignore states */ +#define INET_IGNORE_NONE 0 +#define INET_IGNORE_READ 1 +#define INET_IGNORE_WRITE 1 << 1 + /* Max length of Erlang Term Buffer (for outputting structured terms): */ #ifdef HAVE_SCTP #define PACKET_ERL_DRV_TERM_DATA_LEN 512 @@ -864,6 +872,9 @@ typedef struct { double send_avg; /* average packet size sent */ subs_list empty_out_q_subs; /* Empty out queue subscribers */ + int is_ignored; /* if a fd is ignored by from the inet_drv, + this should be set to true when the fd is used + outside of inet_drv. */ } inet_descriptor; @@ -7344,6 +7355,8 @@ static ErlDrvData inet_start(ErlDrvPort port, int size, int protocol) sys_memzero((char *)&desc->remote,sizeof(desc->remote)); + desc->is_ignored = 0; + return (ErlDrvData)desc; } @@ -7626,6 +7639,33 @@ static int inet_ctl(inet_descriptor* desc, int cmd, char* buf, int len, return ctl_reply(INET_REP_OK, tbuf, 2, rbuf, rsize); } + case INET_REQ_IGNOREFD: { + DEBUGF(("inet_ctl(%ld): IGNOREFD, IGNORED = %d\r\n", + (long)desc->port,(int)*buf)); + + /* + * FD can only be ignored for connected TCP connections for now, + * possible to add UDP and SCTP support if needed. + */ + if (!IS_CONNECTED(desc)) + return ctl_error(ENOTCONN, rbuf, rsize); + + if (!desc->stype == SOCK_STREAM) + return ctl_error(EINVAL, rbuf, rsize); + + if (*buf == 1 && !desc->is_ignored) { + desc->is_ignored = INET_IGNORE_READ; + sock_select(desc, (FD_READ|FD_WRITE|FD_CLOSE|ERL_DRV_USE_NO_CALLBACK), 0); + } else if (*buf == 0 && desc->is_ignored) { + int flags = (FD_READ|FD_CLOSE|((desc->is_ignored & INET_IGNORE_WRITE)?FD_WRITE:0)); + desc->is_ignored = INET_IGNORE_NONE; + sock_select(desc, flags, 1); + } else + return ctl_error(EINVAL, rbuf, rsize); + + return ctl_reply(INET_REP_OK, NULL, 0, rbuf, rsize); + } + #ifndef VXWORKS case INET_REQ_GETSERVBYNAME: { /* L1 Name-String L2 Proto-String */ @@ -8001,6 +8041,7 @@ static int tcp_inet_ctl(ErlDrvData e, unsigned int cmd, char* buf, int len, char** rbuf, int rsize) { tcp_descriptor* desc = (tcp_descriptor*)e; + switch(cmd) { case INET_REQ_OPEN: { /* open socket and return internal index */ int domain; @@ -8266,13 +8307,14 @@ static int tcp_inet_ctl(ErlDrvData e, unsigned int cmd, char* buf, int len, if (enq_async(INETP(desc), tbuf, TCP_REQ_RECV) < 0) return ctl_error(EALREADY, rbuf, rsize); - if (tcp_recv(desc, n) == 0) { + if (INETP(desc)->is_ignored || tcp_recv(desc, n) == 0) { if (timeout == 0) async_error_am(INETP(desc), am_timeout); else { if (timeout != INET_INFINITY) - driver_set_timer(desc->inet.port, timeout); - sock_select(INETP(desc),(FD_READ|FD_CLOSE),1); + driver_set_timer(desc->inet.port, timeout); + if (!INETP(desc)->is_ignored) + sock_select(INETP(desc),(FD_READ|FD_CLOSE),1); } } return ctl_reply(INET_REP_OK, tbuf, 2, rbuf, rsize); @@ -9012,6 +9054,7 @@ static int tcp_inet_input(tcp_descriptor* desc, HANDLE event) #ifdef DEBUG long port = (long) desc->inet.port; /* Used after driver_exit() */ #endif + ASSERT(!INETP(desc)->is_ignored); DEBUGF(("tcp_inet_input(%ld) {s=%d\r\n", port, desc->inet.s)); if (desc->inet.state == INET_STATE_ACCEPTING) { SOCKET s; @@ -9273,7 +9316,11 @@ static int tcp_sendv(tcp_descriptor* desc, ErlIOVec* ev) DEBUGF(("tcp_sendv(%ld): s=%d, about to send %d,%d bytes\r\n", (long)desc->inet.port, desc->inet.s, h_len, len)); - if (desc->tcp_add_flags & TCP_ADDF_DELAY_SEND) { + + if (INETP(desc)->is_ignored) { + INETP(desc)->is_ignored |= INET_IGNORE_WRITE; + n = 0; + } else if (desc->tcp_add_flags & TCP_ADDF_DELAY_SEND) { n = 0; } else if (IS_SOCKET_ERROR(sock_sendv(desc->inet.s, ev->iov, vsize, &n, 0))) { @@ -9301,7 +9348,8 @@ static int tcp_sendv(tcp_descriptor* desc, ErlIOVec* ev) DEBUGF(("tcp_sendv(%ld): s=%d, Send failed, queuing\r\n", (long)desc->inet.port, desc->inet.s)); driver_enqv(ix, ev, n); - sock_select(INETP(desc),(FD_WRITE|FD_CLOSE), 1); + if (!INETP(desc)->is_ignored) + sock_select(INETP(desc),(FD_WRITE|FD_CLOSE), 1); } return 0; } @@ -9366,7 +9414,10 @@ static int tcp_send(tcp_descriptor* desc, char* ptr, int len) DEBUGF(("tcp_send(%ld): s=%d, about to send %d,%d bytes\r\n", (long)desc->inet.port, desc->inet.s, h_len, len)); - if (desc->tcp_add_flags & TCP_ADDF_DELAY_SEND) { + if (INETP(desc)->is_ignored) { + INETP(desc)->is_ignored |= INET_IGNORE_WRITE; + n = 0; + } else if (desc->tcp_add_flags & TCP_ADDF_DELAY_SEND) { sock_send(desc->inet.s, buf, 0, 0); n = 0; } else if (IS_SOCKET_ERROR(sock_sendv(desc->inet.s,iov,2,&n,0))) { @@ -9397,7 +9448,8 @@ static int tcp_send(tcp_descriptor* desc, char* ptr, int len) n -= h_len; driver_enq(ix, ptr+n, len-n); } - sock_select(INETP(desc),(FD_WRITE|FD_CLOSE), 1); + if (!INETP(desc)->is_ignored) + sock_select(INETP(desc),(FD_WRITE|FD_CLOSE), 1); } return 0; } @@ -9421,6 +9473,7 @@ static int tcp_inet_output(tcp_descriptor* desc, HANDLE event) int ret = 0; ErlDrvPort ix = desc->inet.port; + ASSERT(!INETP(desc)->is_ignored); DEBUGF(("tcp_inet_output(%ld) {s=%d\r\n", (long)desc->inet.port, desc->inet.s)); if (desc->inet.state == INET_STATE_CONNECTING) { diff --git a/erts/emulator/drivers/unix/unix_efile.c b/erts/emulator/drivers/unix/unix_efile.c index 4b3934657c..72911641d3 100644 --- a/erts/emulator/drivers/unix/unix_efile.c +++ b/erts/emulator/drivers/unix/unix_efile.c @@ -33,6 +33,9 @@ #include <sys/types.h> #include <sys/uio.h> #endif +#if defined(HAVE_SENDFILE) && (defined(__linux__) || (defined(__sun) && defined(__SVR4))) +#include <sys/sendfile.h> +#endif #if defined(__APPLE__) && defined(__MACH__) && !defined(__DARWIN__) #define DARWIN 1 @@ -1464,3 +1467,77 @@ efile_fadvise(Efile_error* errInfo, int fd, Sint64 offset, return check_error(0, errInfo); #endif } + +#ifdef HAVE_SENDFILE +#define SENDFILE_CHUNK_SIZE ((1 << 30) -1) + +/* + * sendfile: The implementation of the sendfile system call varies + * a lot on different *nix platforms so to make the api similar in all + * we have to emulate some things in linux and play with variables on + * bsd/darwin. + * + * It could be possible to implement header/trailer in sendfile, though + * you would have to emulate it in linux and on BSD/Darwin some complex + * calculations have to be made when using a non blocking socket to figure + * out how much of the header/file/trailer was sent in each command. + */ + +int +efile_sendfile(Efile_error* errInfo, int in_fd, int out_fd, + off_t *offset, Uint64 *nbytes, struct t_sendfile_hdtl* hdtl) +{ + Uint64 written = 0; +#if defined(__linux__) || (defined(__sun) && defined(__SVR4)) + ssize_t retval; + do { + // check if *nbytes is 0 or greater than the largest size_t + if (*nbytes == 0 || *nbytes > SENDFILE_CHUNK_SIZE) + retval = sendfile(out_fd, in_fd, offset, SENDFILE_CHUNK_SIZE); + else + retval = sendfile(out_fd, in_fd, offset, *nbytes); + if (retval > 0) { + written += retval; + *nbytes -= retval; + } + } while (retval != -1 && retval == SENDFILE_CHUNK_SIZE); + *nbytes = written; + return check_error(retval == -1 ? -1 : 0, errInfo); +#elif defined(DARWIN) + int retval; + off_t len; + do { + // check if *nbytes is 0 or greater than the largest off_t + if(*nbytes > SENDFILE_CHUNK_SIZE) + len = SENDFILE_CHUNK_SIZE; + else + len = *nbytes; + retval = sendfile(in_fd, out_fd, *offset, &len, NULL, 0); + if (retval != -1 || errno == EAGAIN || errno == EINTR) { + *offset += len; + *nbytes -= len; + written += len; + } + } while (len == SENDFILE_CHUNK_SIZE); + *nbytes = written; + return check_error(retval, errInfo); +#elif defined(__FreeBSD__) || defined(__DragonFly__) + off_t len; + int retval; + do { + if (*nbytes > SENDFILE_CHUNK_SIZE) + retval = sendfile(in_fd, out_fd, *offset, SENDFILE_CHUNK_SIZE, + NULL, &len, 0); + else + retval = sendfile(in_fd, out_fd, *offset, *nbytes, NULL, &len, 0); + if (retval != -1 || errno == EAGAIN || errno == EINTR) { + *offset += len; + *nbytes -= len; + written += len; + } + } while(len == SENDFILE_CHUNK_SIZE); + *nbytes = written; + return check_error(retval, errInfo); +#endif +} +#endif /* HAVE_SENDFILE */ diff --git a/erts/preloaded/src/prim_file.erl b/erts/preloaded/src/prim_file.erl index 30b7a5246a..7316e0be99 100644 --- a/erts/preloaded/src/prim_file.erl +++ b/erts/preloaded/src/prim_file.erl @@ -26,7 +26,8 @@ %% Generic file contents operations -export([open/2, close/1, datasync/1, sync/1, advise/4, position/2, truncate/1, - write/2, pwrite/2, pwrite/3, read/2, read_line/1, pread/2, pread/3, copy/3]). + write/2, pwrite/2, pwrite/3, read/2, read_line/1, pread/2, pread/3, + copy/3, sendfile/10]). %% Specialized file operations -export([open/1, open/3]). @@ -98,6 +99,7 @@ -define(FILE_READ_LINE, 29). -define(FILE_FDATASYNC, 30). -define(FILE_ADVISE, 31). +-define(FILE_SENDFILE, 32). %% Driver responses -define(FILE_RESP_OK, 0). @@ -537,7 +539,31 @@ 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(#file_descriptor{module = ?MODULE, data = {Port, _}}, + Dest, Offset, Bytes, _ChunkSize, Headers, Trailers, + _Nodiskio, _MNowait, _Sync) -> + case erlang:port_get_data(Dest) of + Data when Data == inet_tcp; Data == inet6_tcp -> + ok = inet:lock_socket(Dest,true), + {ok, DestFD} = prim_inet:getfd(Dest), + try drv_command(Port, [<<?FILE_SENDFILE, DestFD:32, + 0:8, + Offset:64/unsigned, + Bytes:64/unsigned, + (iolist_size(Headers)):32/unsigned, + (iolist_size(Trailers)):32/unsigned>>, + Headers,Trailers]) + after + ok = inet:lock_socket(Dest,false) + end; + _Else -> + {error,badarg} + end. %%%----------------------------------------------------------------- diff --git a/erts/preloaded/src/prim_inet.erl b/erts/preloaded/src/prim_inet.erl index f144f73d68..0cedd284db 100644 --- a/erts/preloaded/src/prim_inet.erl +++ b/erts/preloaded/src/prim_inet.erl @@ -36,7 +36,8 @@ -export([recvfrom/2, recvfrom/3]). -export([setopt/3, setopts/2, getopt/2, getopts/2, is_sockopt_val/2]). -export([chgopt/3, chgopts/2]). --export([getstat/2, getfd/1, getindex/1, getstatus/1, gettype/1, +-export([getstat/2, getfd/1, ignorefd/2, + getindex/1, getstatus/1, gettype/1, getifaddrs/1, getiflist/1, ifget/3, ifset/3, gethostname/1]). -export([getservbyname/3, getservbyport/3]). @@ -842,6 +843,21 @@ getfd(S) when is_port(S) -> %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% +%% IGNOREFD(insock(),boolean()) -> {ok,integer()} | {error, Reason} +%% +%% steal internal file descriptor +%% +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +ignorefd(S,Bool) when is_port(S) -> + Val = if Bool -> 1; true -> 0 end, + case ctl_cmd(S, ?INET_REQ_IGNOREFD, [Val]) of + {ok, _} -> ok; + Error -> Error + end. + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% +%% %% GETIX(insock()) -> {ok,integer()} | {error, Reason} %% %% get internal socket index diff --git a/lib/kernel/doc/src/file.xml b/lib/kernel/doc/src/file.xml index 7db20e6343..c6a1f25dd9 100644 --- a/lib/kernel/doc/src/file.xml +++ b/lib/kernel/doc/src/file.xml @@ -149,6 +149,9 @@ <datatype> <name name="mode"/> </datatype> + <datatype> + <name name="sendfile_option"/> + </datatype> </datatypes> <funcs> @@ -1574,6 +1577,48 @@ </desc> </func> <func> + <name name="sendfile" arity="2"/> + <fsummary>send a file to a socket</fsummary> + <desc> + <p>Sends the file <c>Filename</c> to <c>Socket</c>. + Returns <c>{ok, BytesSent}</c> if successful, + otherwise <c>{error, Reason}</c>.</p> + </desc> + </func> + <func> + <name name="sendfile" arity="5"/> + <fsummary>send a file to a socket</fsummary> + <desc> + <p>Sends <c>Bytes</c> from the file + referenced by <c>RawFile</c> beginning at <c>Offset</c> to + <c>Socket</c>. + Returns <c>{ok, BytesSent}</c> if successful, + otherwise <c>{error, Reason}</c>. If <c>Bytes</c> is set to + 0 all data after the given <c>Offset</c> is sent.</p> + <p>The file used must be opened using the raw flag, and the process + calling sendfile must be the controlling process of the socket. + See <seealso marker="gen_tcp#controlling_process-2">gen_tcp:controlling_process/2</seealso></p> + <p>If the OS used does not support sendfile, an Erlang fallback + using file:read and gen_tcp:send is used.</p> + <p>The option list can contain the following options: + <taglist> + <tag><c>chunk_size</c></tag> + <item>The chunk size used by the erlang fallback to send + data. If using the fallback, this should be set to a value + which comfortably fits in the systems memory. Default is 20 MB.</item> + </taglist> + </p> + <p>On operating systems with thread support, it is recommended to use + async threads. See the command line flag + <c>+A</c> in <seealso marker="erts:erl">erl(1)</seealso>. If it is not + possible to use async threads for sendfile, it is recommended to use + a relatively small value for the send buffer on the socket. Otherwise + the Erlang VM might loose some of its soft realtime guarantees. + Which size to use depends on the OS/hardware and the requirements + of the application.</p> + </desc> + </func> + <func> <name name="write" arity="2"/> <fsummary>Write to a file</fsummary> <desc> diff --git a/lib/kernel/src/file.erl b/lib/kernel/src/file.erl index 706c60caaf..0b0f91d86a 100644 --- a/lib/kernel/src/file.erl +++ b/lib/kernel/src/file.erl @@ -51,6 +51,9 @@ -export([pid2name/1]). +%% Sendfile functions +-export([sendfile/2,sendfile/5]). + %%% Obsolete exported functions -export([raw_read_file_info/1, raw_write_file_info/2]). @@ -103,7 +106,7 @@ -type date_time() :: calendar:datetime(). -type posix_file_advise() :: 'normal' | 'sequential' | 'random' | 'no_reuse' | 'will_need' | 'dont_need'. - +-type sendfile_option() :: {chunk_size, non_neg_integer()}. %%%----------------------------------------------------------------- %%% General functions @@ -1114,6 +1117,140 @@ change_time(Name, Atime, Mtime) when is_tuple(Atime), is_tuple(Mtime) -> write_file_info(Name, #file_info{atime=Atime, mtime=Mtime}). +%% +%% Send data using sendfile +%% + +-define(MAX_CHUNK_SIZE, (1 bsl 20)*20). %% 20 MB, has to fit in primary memory + +-spec sendfile(RawFile, Socket, Offset, Bytes, Opts) -> + {'ok', non_neg_integer()} | {'error', inet:posix() | badarg | not_owner} when + RawFile :: file:fd(), + Socket :: inet:socket(), + Offset :: non_neg_integer(), + Bytes :: non_neg_integer(), + Opts :: [sendfile_option()]. +sendfile(File, _Sock, _Offet, _Bytes, _Opts) when is_pid(File) -> + {error, badarg}; +sendfile(File, Sock, Offset, Bytes, []) -> + sendfile(File, Sock, Offset, Bytes, ?MAX_CHUNK_SIZE, [], [], + false, false, false); +sendfile(File, Sock, Offset, Bytes, Opts) -> + ChunkSize0 = proplists:get_value(chunk_size, Opts, ?MAX_CHUNK_SIZE), + ChunkSize = if ChunkSize0 > ?MAX_CHUNK_SIZE -> + ?MAX_CHUNK_SIZE; + true -> ChunkSize0 + end, + %% Support for headers, trailers and options has been removed because the + %% Darwin and BSD API for using it does not play nice with + %% non-blocking sockets. See unix_efile.c for more info. + sendfile(File, Sock, Offset, Bytes, ChunkSize, [], [], + false,false,false). + +%% sendfile/2 +-spec sendfile(Filename, Socket) -> + {'ok', non_neg_integer()} | {'error', inet:posix() | badarg | not_owner} + when Filename :: file:name(), + Socket :: inet:socket(). +sendfile(Filename, Sock) -> + case file:open(Filename, [read, raw, binary]) of + {error, Reason} -> + {error, Reason}; + {ok, Fd} -> + Res = sendfile(Fd, Sock, 0, 0, []), + file:close(Fd), + Res + end. + +%% Internal sendfile functions +sendfile(#file_descriptor{ module = Mod } = Fd, Sock, Offset, Bytes, + ChunkSize, Headers, Trailers, Nodiskio, MNowait, Sync) + when is_port(Sock) -> + case Mod:sendfile(Fd, Sock, Offset, Bytes, ChunkSize, Headers, Trailers, + Nodiskio, MNowait, Sync) of + {error, enotsup} -> + sendfile_fallback(Fd, Sock, Offset, Bytes, ChunkSize, + Headers, Trailers); + Else -> + Else + end; +sendfile(_,_,_,_,_,_,_,_,_,_) -> + {error, badarg}. + +%%% +%% Sendfile Fallback +%%% +sendfile_fallback(File, Sock, Offset, Bytes, ChunkSize, + Headers, Trailers) + when Headers == []; is_integer(Headers) -> + case sendfile_fallback(File, Sock, Offset, Bytes, ChunkSize) of + {ok, BytesSent} when is_list(Trailers), + Trailers =/= [], + is_integer(Headers) -> + sendfile_send(Sock, Trailers, BytesSent+Headers); + {ok, BytesSent} when is_list(Trailers), Trailers =/= [] -> + sendfile_send(Sock, Trailers, BytesSent); + {ok, BytesSent} when is_integer(Headers) -> + {ok, BytesSent + Headers}; + Else -> + Else + end; +sendfile_fallback(File, Sock, Offset, Bytes, ChunkSize, Headers, Trailers) -> + case sendfile_send(Sock, Headers, 0) of + {ok, BytesSent} -> + sendfile_fallback(File, Sock, Offset, Bytes, ChunkSize, BytesSent, + Trailers); + Else -> + Else + end. + + +sendfile_fallback(File, Sock, Offset, Bytes, ChunkSize) -> + {ok, CurrPos} = file:position(File, {cur, 0}), + {ok, _NewPos} = file:position(File, {bof, Offset}), + Res = sendfile_fallback_int(File, Sock, Bytes, ChunkSize, 0), + file:position(File, {bof, CurrPos}), + Res. + + +sendfile_fallback_int(File, Sock, Bytes, ChunkSize, BytesSent) + when Bytes > BytesSent; Bytes == 0 -> + Size = if Bytes == 0 -> + ChunkSize; + (Bytes - BytesSent + ChunkSize) > 0 -> + Bytes - BytesSent; + true -> + ChunkSize + end, + case file:read(File, Size) of + {ok, Data} -> + case sendfile_send(Sock, Data, BytesSent) of + {ok,NewBytesSent} -> + sendfile_fallback_int( + File, Sock, Bytes, ChunkSize, + NewBytesSent); + Error -> + Error + end; + eof -> + {ok, BytesSent}; + Error -> + Error + end; +sendfile_fallback_int(_File, _Sock, BytesSent, _ChunkSize, BytesSent) -> + {ok, BytesSent}. + +sendfile_send(Sock, Data, Old) -> + Len = iolist_size(Data), + case gen_tcp:send(Sock, Data) of + ok -> + {ok, Len+Old}; + Else -> + Else + end. + + + %%%----------------------------------------------------------------- %%% Helpers diff --git a/lib/kernel/src/gen_tcp.erl b/lib/kernel/src/gen_tcp.erl index 8ab18c01b4..4d6c7f5f1d 100644 --- a/lib/kernel/src/gen_tcp.erl +++ b/lib/kernel/src/gen_tcp.erl @@ -27,6 +27,7 @@ -export([fdopen/2]). -include("inet_int.hrl"). +-include("file.hrl"). -type option() :: {active, true | false | once} | @@ -302,7 +303,7 @@ unrecv(S, Data) when is_port(S) -> Mod:unrecv(S, Data); Error -> Error - end. + end. %% %% Set controlling process @@ -354,3 +355,4 @@ mod([_|Opts], Address) -> mod(Opts, Address); mod([], Address) -> mod(Address). + diff --git a/lib/kernel/src/inet.erl b/lib/kernel/src/inet.erl index b60c68e3a1..49f64a9236 100644 --- a/lib/kernel/src/inet.erl +++ b/lib/kernel/src/inet.erl @@ -40,6 +40,10 @@ -export([tcp_controlling_process/2, udp_controlling_process/2, tcp_close/1, udp_close/1]). + +%% used by sendfile +-export([lock_socket/2]). + %% used by socks5 -export([setsockname/2, setpeername/2]). @@ -1353,3 +1357,14 @@ stop_timer(Timer) -> end; T -> T end. + + +lock_socket(S,Val) -> + case erlang:port_info(S, connected) of + {connected, Pid} when Pid =/= self() -> + {error, not_owner}; + undefined -> + {error, einval}; + _ -> + prim_inet:ignorefd(S,Val) + end. diff --git a/lib/kernel/src/inet_int.hrl b/lib/kernel/src/inet_int.hrl index f8984b13fe..cf893c73eb 100644 --- a/lib/kernel/src/inet_int.hrl +++ b/lib/kernel/src/inet_int.hrl @@ -85,6 +85,8 @@ -define(INET_REQ_GETIFADDRS, 25). -define(INET_REQ_ACCEPT, 26). -define(INET_REQ_LISTEN, 27). +-define(INET_REQ_IGNOREFD, 28). + %% TCP requests %%-define(TCP_REQ_ACCEPT, 40). MOVED %%-define(TCP_REQ_LISTEN, 41). MERGED diff --git a/lib/kernel/test/Makefile b/lib/kernel/test/Makefile index 82bc3fc6d1..5dcaad3f5e 100644 --- a/lib/kernel/test/Makefile +++ b/lib/kernel/test/Makefile @@ -74,7 +74,8 @@ MODULES= \ wrap_log_reader_SUITE \ cleanup \ zlib_SUITE \ - loose_node + loose_node \ + sendfile_SUITE APP_FILES = \ appinc.app \ diff --git a/lib/kernel/test/gen_tcp_api_SUITE.erl b/lib/kernel/test/gen_tcp_api_SUITE.erl index cbaec2d6dd..a7af00c12a 100644 --- a/lib/kernel/test/gen_tcp_api_SUITE.erl +++ b/lib/kernel/test/gen_tcp_api_SUITE.erl @@ -22,7 +22,7 @@ %% are not tested here, because they are tested indirectly in this and %% and other test suites. --include_lib("test_server/include/test_server.hrl"). +-include_lib("common_test/include/ct.hrl"). -include_lib("kernel/include/inet.hrl"). -export([all/0, suite/0,groups/0,init_per_suite/1, end_per_suite/1, @@ -46,6 +46,8 @@ groups() -> {t_connect, [], [t_connect_timeout, t_connect_bad]}, {t_recv, [], [t_recv_timeout, t_recv_eof]}]. + + init_per_suite(Config) -> Config. @@ -55,9 +57,8 @@ end_per_suite(_Config) -> init_per_group(_GroupName, Config) -> Config. -end_per_group(_GroupName, Config) -> - Config. - +end_per_group(_,_Config) -> + ok. init_per_testcase(_Func, Config) -> Dog = test_server:timetrap(test_server:seconds(60)), @@ -237,7 +238,6 @@ implicit_inet6(S, Addr) -> ?line ok = gen_tcp:close(S1). - %%% Utilities %% Calls M:F/length(A), which should return a timeout error, and complete diff --git a/lib/kernel/test/sendfile_SUITE.erl b/lib/kernel/test/sendfile_SUITE.erl new file mode 100644 index 0000000000..04af16a6b9 --- /dev/null +++ b/lib/kernel/test/sendfile_SUITE.erl @@ -0,0 +1,278 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2011-2011. All Rights Reserved. +%% +%% The contents of this file are subject to the Erlang Public License, +%% Version 1.1, (the "License"); you may not use this file except in +%% compliance with the License. You should have received a copy of the +%% Erlang Public License along with this software. If not, it can be +%% retrieved online at http://www.erlang.org/. +%% +%% Software distributed under the License is distributed on an "AS IS" +%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See +%% the License for the specific language governing rights and limitations +%% under the License. +%% +%% %CopyrightEnd% +%% + +-module(sendfile_SUITE). + +-include_lib("common_test/include/ct.hrl"). +-include_lib("kernel/include/file.hrl"). + +-compile(export_all). + +all() -> + [t_sendfile_small + ,t_sendfile_big + ,t_sendfile_partial + ,t_sendfile_offset + ,t_sendfile_sendafter + ,t_sendfile_recvafter + ,t_sendfile_sendduring + ,t_sendfile_recvduring + ]. + +init_per_suite(Config) -> + Priv = ?config(priv_dir, Config), + SFilename = filename:join(Priv, "sendfile_small.html"), + {ok, DS} = file:open(SFilename,[write,raw]), + file:write(DS,"yo baby yo"), + file:sync(DS), + file:close(DS), + BFilename = filename:join(Priv, "sendfile_big.html"), + {ok, DB} = file:open(BFilename,[write,raw]), + [file:write(DB,[<<0:(10*8*1024*1024)>>]) || _I <- lists:seq(1,51)], + file:sync(DB), + file:close(DB), + [{small_file, SFilename}, + {file_opts,[raw,binary]}, + {big_file, BFilename}|Config]. + +end_per_suite(Config) -> + file:delete(proplists:get_value(big_file, Config)). + +init_per_testcase(TC,Config) when TC == t_sendfile_recvduring; + TC == t_sendfile_sendduring -> + Filename = proplists:get_value(small_file, Config), + + Send = fun(Sock) -> + {_Size, Data} = sendfile_file_info(Filename), + {ok,D} = file:open(Filename, [raw,binary,read]), + prim_file:sendfile(D, Sock, 0, 0, 0, + [],[],false,false,false), + Data + end, + + %% Check if sendfile is supported on this platform + case catch sendfile_send(Send) of + ok -> + Config; + Error -> + ct:log("Error: ~p",[Error]), + {skip,"Not supported"} + end; +init_per_testcase(_Tc,Config) -> + Config. + + +t_sendfile_small(Config) when is_list(Config) -> + Filename = proplists:get_value(small_file, Config), + + Send = fun(Sock) -> + {Size, Data} = sendfile_file_info(Filename), + {ok, Size} = file:sendfile(Filename, Sock), + Data + end, + + ok = sendfile_send(Send). + +t_sendfile_big(Config) when is_list(Config) -> + Filename = proplists:get_value(big_file, Config), + + Send = fun(Sock) -> + {ok, #file_info{size = Size}} = + file:read_file_info(Filename), + {ok, Size} = file:sendfile(Filename, Sock), + Size + end, + + ok = sendfile_send("localhost", Send, 0). + +t_sendfile_partial(Config) -> + Filename = proplists:get_value(small_file, Config), + FileOpts = proplists:get_value(file_opts, Config, []), + + SendSingle = fun(Sock) -> + {_Size, <<Data:5/binary,_/binary>>} = + sendfile_file_info(Filename), + {ok,D} = file:open(Filename,[read|FileOpts]), + {ok,5} = file:sendfile(D,Sock,0,5,[]), + file:close(D), + Data + end, + ok = sendfile_send(SendSingle), + + {_Size, <<FData:5/binary,SData:3/binary,_/binary>>} = + sendfile_file_info(Filename), + {ok,D} = file:open(Filename,[read|FileOpts]), + {ok, <<FData/binary>>} = file:read(D,5), + FSend = fun(Sock) -> + {ok,5} = file:sendfile(D,Sock,0,5,[]), + FData + end, + + ok = sendfile_send(FSend), + + SSend = fun(Sock) -> + {ok,3} = file:sendfile(D,Sock,5,3,[]), + SData + end, + + ok = sendfile_send(SSend), + + {ok, <<SData/binary>>} = file:read(D,3), + + file:close(D). + +t_sendfile_offset(Config) -> + Filename = proplists:get_value(small_file, Config), + FileOpts = proplists:get_value(file_opts, Config, []), + + Send = fun(Sock) -> + {_Size, <<_:5/binary,Data:3/binary,_/binary>> = AllData} = + sendfile_file_info(Filename), + {ok,D} = file:open(Filename,[read|FileOpts]), + {ok,3} = file:sendfile(D,Sock,5,3,[]), + {ok, AllData} = file:read(D,100), + file:close(D), + Data + end, + ok = sendfile_send(Send). + + +t_sendfile_sendafter(Config) -> + Filename = proplists:get_value(small_file, Config), + + Send = fun(Sock) -> + {Size, Data} = sendfile_file_info(Filename), + {ok, Size} = file:sendfile(Filename, Sock), + ok = gen_tcp:send(Sock, <<2>>), + <<Data/binary,2>> + end, + + ok = sendfile_send(Send). + +t_sendfile_recvafter(Config) -> + Filename = proplists:get_value(small_file, Config), + + Send = fun(Sock) -> + {Size, Data} = sendfile_file_info(Filename), + {ok, Size} = file:sendfile(Filename, Sock), + ok = gen_tcp:send(Sock, <<1>>), + {ok,<<1>>} = gen_tcp:recv(Sock, 1), + <<Data/binary,1>> + end, + + ok = sendfile_send(Send). + +t_sendfile_sendduring(Config) -> + Filename = proplists:get_value(big_file, Config), + + Send = fun(Sock) -> + {ok, #file_info{size = Size}} = + file:read_file_info(Filename), + spawn_link(fun() -> + timer:sleep(10), + ok = gen_tcp:send(Sock, <<2>>) + end), + {ok, Size} = file:sendfile(Filename, Sock), + Size+1 + end, + + ok = sendfile_send("localhost", Send, 0). + +t_sendfile_recvduring(Config) -> + Filename = proplists:get_value(big_file, Config), + + Send = fun(Sock) -> + {ok, #file_info{size = Size}} = + file:read_file_info(Filename), + spawn_link(fun() -> + timer:sleep(10), + ok = gen_tcp:send(Sock, <<1>>), + {ok,<<1>>} = gen_tcp:recv(Sock, 1) + end), + {ok, Size} = file:sendfile(Filename, Sock), + timer:sleep(1000), + Size+1 + end, + + ok = sendfile_send("localhost", Send, 0). + +%% TODO: consolidate tests and reduce code +sendfile_send(Send) -> + sendfile_send("localhost",Send). +sendfile_send(Host, Send) -> + sendfile_send(Host, Send, []). +sendfile_send(Host, Send, Orig) -> + spawn_link(?MODULE, sendfile_server, [self(), Orig]), + receive + {server, Port} -> + {ok, Sock} = gen_tcp:connect(Host, Port, + [binary,{packet,0}, + {active,false}]), + Data = Send(Sock), + ok = gen_tcp:close(Sock), + receive + {ok, Bin} -> + Data = Bin, + ok + end + end. + +sendfile_server(ClientPid, Orig) -> + {ok, LSock} = gen_tcp:listen(0, [binary, {packet, 0}, + {active, true}, + {reuseaddr, true}]), + {ok, Port} = inet:port(LSock), + ClientPid ! {server, Port}, + {ok, Sock} = gen_tcp:accept(LSock), + {ok, Bin} = sendfile_do_recv(Sock, Orig), + ClientPid ! {ok, Bin}, + gen_tcp:send(Sock, <<1>>). + +-define(SENDFILE_TIMEOUT, 10000). +%% f(),{ok, S} = gen_tcp:connect("localhost",7890,[binary]),file:sendfile("/ldisk/lukas/otp/sendfiletest.dat",S). +sendfile_do_recv(Sock, Bs) -> + receive + {tcp, Sock, B} -> + case binary:match(B,<<1>>) of + nomatch when is_list(Bs) -> + sendfile_do_recv(Sock, [B|Bs]); + nomatch when is_integer(Bs) -> + sendfile_do_recv(Sock, byte_size(B) + Bs); + _ when is_list(Bs) -> + ct:log("Stopped due to a 1"), + {ok, iolist_to_binary(lists:reverse([B|Bs]))}; + _ when is_integer(Bs) -> + ct:log("Stopped due to a 1"), + {ok, byte_size(B) + Bs} + end; + {tcp_closed, Sock} when is_list(Bs) -> + ct:log("Stopped due to close"), + {ok, iolist_to_binary(lists:reverse(Bs))}; + {tcp_closed, Sock} when is_integer(Bs) -> + ct:log("Stopped due to close"), + {ok, Bs} + after ?SENDFILE_TIMEOUT -> + ct:log("Sendfile timeout"), + timeout + end. + +sendfile_file_info(File) -> + {ok, #file_info{size = Size}} = file:read_file_info(File), + {ok, Data} = file:read_file(File), + {Size, Data}. |