mirror/guile
mirror/guile
1
Fork 0
mirror of https://git.savannah.gnu.org/git/guile.git synced 2025-05-01 04:10:18 +02:00
Code Activity

Support for non-blocking I/O

Browse source 
* doc/ref/api-io.texi (I/O Extensions): Document read_wait_fd /
  write_wait_fd members.
  (Non-Blocking I/O): New section.
* libguile/fports.c (fport_read, fport_write): Return -1 if the
  operation would block.
  (fport_wait_fd, scm_make_fptob): Add read/write wait-fd
  implementation.
* libguile/ports-internal.h (scm_t_port_type): Add read_wait_fd /
  write_wait_fd.
* libguile/ports.c (default_read_wait_fd, default_write_wait_fd): New
  functions.
  (scm_make_port_type): Initialize default read/write wait fd impls.
  (trampoline_to_c_read, trampoline_to_scm_read)
  (trampoline_to_c_write, trampoline_to_scm_write): To Scheme, a return
  of #f indicates EWOULDBLOCk.
  (scm_set_port_read_wait_fd, scm_set_port_write_wait_fd): New
  functions.
  (port_read_wait_fd, port_write_wait_fd, scm_port_read_wait_fd)
  (scm_port_write_wait_fd, port_poll, scm_port_poll): New functions.
  (scm_i_read_bytes, scm_i_write_bytes): Poll if the read or write would
  block.
* libguile/ports.h (scm_set_port_read_wait_fd)
  (scm_set_port_write_wait_fd): Add declarations.
* module/ice-9/ports.scm: Shunt port-poll and port-{read,write}-wait-fd
  to the internals module.
* module/ice-9/sports.scm (current-write-waiter):
  (current-read-waiter): Implement.
* test-suite/tests/ports.test: Adapt non-blocking test to new behavior.
* NEWS: Add entry.
This commit is contained in:
Andy Wingo 2016-05-20 14:51:51 +02:00
parent 8b6f4df3f4
commit 534139e458
9 changed files with 344 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

46
libguile/fports.c
View file

@ -573,14 +573,24 @@ fport_print (SCM exp, SCM port, scm_print_state *pstate SCM_UNUSED)
static size_t
fport_read (SCM port, SCM dst, size_t start, size_t count)
{
long res;
scm_t_fport *fp = SCM_FSTREAM (port);
signed char *ptr = SCM_BYTEVECTOR_CONTENTS (dst) + start;
ssize_t ret;
SCM_SYSCALL (res = read (fp->fdes, ptr, count));
if (res == -1)
scm_syserror ("fport_read");
return res;
retry:
ret = read (fp->fdes, ptr, count);
if (ret < 0)
{
if (errno == EINTR)
{
SCM_ASYNC_TICK;
goto retry;
}
if (errno == EWOULDBLOCK || errno == EAGAIN)
return -1;
scm_syserror ("fport_read");
}
return ret;
}
static size_t
@ -588,11 +598,23 @@ fport_write (SCM port, SCM src, size_t start, size_t count)
{
int fd = SCM_FPORT_FDES (port);
signed char *ptr = SCM_BYTEVECTOR_CONTENTS (src) + start;
ssize_t ret;
if (full_write (fd, ptr, count) < count)
scm_syserror ("fport_write");
retry:
ret = write (fd, ptr, count);
if (ret < 0)
{
if (errno == EINTR)
{
SCM_ASYNC_TICK;
goto retry;
}
if (errno == EWOULDBLOCK || errno == EAGAIN)
return -1;
scm_syserror ("fport_write");
}
return count;
return ret;
}
static scm_t_off
@ -637,6 +659,12 @@ fport_random_access_p (SCM port)
return SCM_FDES_RANDOM_P (SCM_FSTREAM (port)->fdes);
}
static int
fport_wait_fd (SCM port)
{
return SCM_FSTREAM (port)->fdes;
}
/* Query the OS to get the natural buffering for FPORT, if available. */
static void
fport_get_natural_buffer_sizes (SCM port, size_t *read_size, size_t *write_size)
@ -660,6 +688,8 @@ scm_make_fptob ()
scm_set_port_close (ptob, fport_close);
scm_set_port_seek (ptob, fport_seek);
scm_set_port_truncate (ptob, fport_truncate);
scm_set_port_read_wait_fd (ptob, fport_wait_fd);
scm_set_port_write_wait_fd (ptob, fport_wait_fd);
scm_set_port_input_waiting (ptob, fport_input_waiting);
scm_set_port_random_access_p (ptob, fport_random_access_p);
scm_set_port_get_natural_buffer_sizes (ptob, fport_get_natural_buffer_sizes);

3
libguile/ports-internal.h
View file

@ -44,6 +44,9 @@ struct scm_t_port_type
SCM scm_read;
SCM scm_write;
int (*read_wait_fd) (SCM port);
int (*write_wait_fd) (SCM port);
scm_t_off (*seek) (SCM port, scm_t_off OFFSET, int WHENCE);
void (*close) (SCM port);

179
libguile/ports.c
View file

@ -33,6 +33,7 @@
#include <fcntl.h> /* for chsize on mingw */
#include <assert.h>
#include <iconv.h>
#include <poll.h>
#include <uniconv.h>
#include <unistr.h>
#include <striconveh.h>
@ -126,6 +127,18 @@ default_random_access_p (SCM port)
return SCM_PORT_TYPE (port)->seek != NULL;
}
static int
default_read_wait_fd (SCM port)
{
scm_misc_error ("read_wait_fd", "unimplemented", SCM_EOL);
}
static int
default_write_wait_fd (SCM port)
{
scm_misc_error ("write_wait_fd", "unimplemented", SCM_EOL);
}
scm_t_port_type *
scm_make_port_type (char *name,
size_t (*read) (SCM port, SCM dst, size_t start,
@ -144,6 +157,8 @@ scm_make_port_type (char *name,
desc->c_write = write;
desc->scm_read = read ? trampoline_to_c_read_subr : SCM_BOOL_F;
desc->scm_write = write ? trampoline_to_c_write_subr : SCM_BOOL_F;
desc->read_wait_fd = default_read_wait_fd;
desc->write_wait_fd = default_write_wait_fd;
desc->random_access_p = default_random_access_p;
scm_make_port_classes (desc);
@ -154,7 +169,7 @@ static SCM
trampoline_to_c_read (SCM port, SCM dst, SCM start, SCM count)
#define FUNC_NAME "port-read"
{
size_t c_start, c_count;
size_t c_start, c_count, ret;
SCM_VALIDATE_OPPORT (1, port);
c_start = scm_to_size_t (start);
@ -162,24 +177,25 @@ trampoline_to_c_read (SCM port, SCM dst, SCM start, SCM count)
SCM_ASSERT_RANGE (2, start, c_start <= c_count);
SCM_ASSERT_RANGE (3, count, c_start+c_count <= scm_c_bytevector_length (dst));
return scm_from_size_t
(SCM_PORT_TYPE (port)->c_read (port, dst, c_start, c_count));
ret = SCM_PORT_TYPE (port)->c_read (port, dst, c_start, c_count);
return ret == (size_t) -1 ? SCM_BOOL_F : scm_from_size_t (ret);
}
#undef FUNC_NAME
static size_t
trampoline_to_scm_read (SCM port, SCM dst, size_t start, size_t count)
{
return scm_to_size_t
(scm_call_4 (SCM_PORT_TYPE (port)->scm_read, port, dst,
scm_from_size_t (start), scm_from_size_t (count)));
SCM ret = scm_call_4 (SCM_PORT_TYPE (port)->scm_read, port, dst,
scm_from_size_t (start), scm_from_size_t (count));
return scm_is_true (ret) ? scm_to_size_t (ret) : (size_t) -1;
}
static SCM
trampoline_to_c_write (SCM port, SCM src, SCM start, SCM count)
#define FUNC_NAME "port-write"
{
size_t c_start, c_count;
size_t c_start, c_count, ret;
SCM_VALIDATE_OPPORT (1, port);
c_start = scm_to_size_t (start);
@ -187,17 +203,18 @@ trampoline_to_c_write (SCM port, SCM src, SCM start, SCM count)
SCM_ASSERT_RANGE (2, start, c_start <= c_count);
SCM_ASSERT_RANGE (3, count, c_start+c_count <= scm_c_bytevector_length (src));
return scm_from_size_t
(SCM_PORT_TYPE (port)->c_write (port, src, c_start, c_count));
ret = SCM_PORT_TYPE (port)->c_write (port, src, c_start, c_count);
return ret == (size_t) -1 ? SCM_BOOL_F : scm_from_size_t (ret);
}
#undef FUNC_NAME
static size_t
trampoline_to_scm_write (SCM port, SCM src, size_t start, size_t count)
{
return scm_to_size_t
(scm_call_4 (SCM_PORT_TYPE (port)->scm_write, port, src,
scm_from_size_t (start), scm_from_size_t (count)));
SCM ret = scm_call_4 (SCM_PORT_TYPE (port)->scm_write, port, src,
scm_from_size_t (start), scm_from_size_t (count));
return scm_is_true (ret) ? scm_to_size_t (ret) : (size_t) -1;
}
void
@ -214,6 +231,18 @@ scm_set_port_scm_write (scm_t_port_type *ptob, SCM write)
ptob->c_write = trampoline_to_scm_write;
}
void
scm_set_port_read_wait_fd (scm_t_port_type *ptob, int (*get_fd) (SCM))
{
ptob->read_wait_fd = get_fd;
}
void
scm_set_port_write_wait_fd (scm_t_port_type *ptob, int (*get_fd) (SCM))
{
ptob->write_wait_fd = get_fd;
}
void
scm_set_port_print (scm_t_port_type *ptob,
int (*print) (SCM exp, SCM port, scm_print_state *pstate))
@ -1230,6 +1259,116 @@ SCM_DEFINE (scm_set_port_conversion_strategy_x, "set-port-conversion-strategy!",
#undef FUNC_NAME
/* Non-blocking I/O. */
static int
port_read_wait_fd (SCM port)
{
scm_t_port_type *ptob = SCM_PORT_TYPE (port);
return ptob->read_wait_fd (port);
}
static int
port_write_wait_fd (SCM port)
{
scm_t_port_type *ptob = SCM_PORT_TYPE (port);
return ptob->write_wait_fd (port);
}
SCM_INTERNAL SCM scm_port_read_wait_fd (SCM);
SCM_DEFINE (scm_port_read_wait_fd, "port-read-wait-fd", 1, 0, 0,
(SCM port), "")
#define FUNC_NAME s_scm_port_read_wait_fd
{
int fd;
port = SCM_COERCE_OUTPORT (port);
SCM_VALIDATE_OPINPORT (1, port);
fd = port_read_wait_fd (port);
return fd < 0 ? SCM_BOOL_F : scm_from_int (fd);
}
#undef FUNC_NAME
SCM_INTERNAL SCM scm_port_write_wait_fd (SCM);
SCM_DEFINE (scm_port_write_wait_fd, "port-write-wait-fd", 1, 0, 0,
(SCM port), "")
#define FUNC_NAME s_scm_port_write_wait_fd
{
int fd;
port = SCM_COERCE_OUTPORT (port);
SCM_VALIDATE_OPOUTPORT (1, port);
fd = port_write_wait_fd (port);
return fd < 0 ? SCM_BOOL_F : scm_from_int (fd);
}
#undef FUNC_NAME
static int
port_poll (SCM port, short events, int timeout)
#define FUNC_NAME "port-poll"
{
struct pollfd pollfd[2];
int nfds = 0, rv = 0;
if (events & POLLIN)
{
pollfd[nfds].fd = port_read_wait_fd (port);
pollfd[nfds].events = events & (POLLIN | POLLPRI);
pollfd[nfds].revents = 0;
nfds++;
}
if (events & POLLOUT)
{
pollfd[nfds].fd = port_write_wait_fd (port);
pollfd[nfds].events = events & (POLLOUT | POLLPRI);
pollfd[nfds].revents = 0;
nfds++;
}
if (nfds == 2 && pollfd[0].fd == pollfd[1].fd)
{
pollfd[0].events |= pollfd[1].events;
nfds--;
}
SCM_SYSCALL (rv = poll (pollfd, nfds, timeout));
if (rv < 0)
SCM_SYSERROR;
return rv;
}
#undef FUNC_NAME
SCM_INTERNAL SCM scm_port_poll (SCM, SCM, SCM);
SCM_DEFINE (scm_port_poll, "port-poll", 2, 1, 0,
(SCM port, SCM events, SCM timeout),
"")
#define FUNC_NAME s_scm_port_poll
{
short c_events = 0;
int c_timeout;
port = SCM_COERCE_OUTPORT (port);
SCM_VALIDATE_PORT (1, port);
SCM_VALIDATE_STRING (2, events);
c_timeout = SCM_UNBNDP (timeout) ? -1 : SCM_NUM2INT (3, timeout);
if (scm_i_string_contains_char (events, 'r'))
c_events |= POLLIN;
if (scm_i_string_contains_char (events, '!'))
c_events |= POLLPRI;
if (scm_i_string_contains_char (events, 'w'))
c_events |= POLLIN;
return scm_from_int (port_poll (port, c_events, c_timeout));
}
#undef FUNC_NAME
/* Input. */
@ -1330,8 +1469,15 @@ scm_i_read_bytes (SCM port, SCM dst, size_t start, size_t count)
assert (count <= SCM_BYTEVECTOR_LENGTH (dst));
assert (start + count <= SCM_BYTEVECTOR_LENGTH (dst));
retry:
filled = ptob->c_read (port, dst, start, count);
if (filled == (size_t) -1)
{
port_poll (port, POLLIN, -1);
goto retry;
}
assert (filled <= count);
return filled;
@ -2508,7 +2654,14 @@ scm_i_write_bytes (SCM port, SCM src, size_t start, size_t count)
assert (start + count <= SCM_BYTEVECTOR_LENGTH (src));
do
written += ptob->c_write (port, src, start + written, count - written);
{
size_t ret = ptob->c_write (port, src, start + written, count - written);
if (ret == (size_t) -1)
port_poll (port, POLLOUT, -1);
else
written += ret;
}
while (written < count);
assert (written == count);

4
libguile/ports.h
View file

@ -90,6 +90,10 @@ SCM_API scm_t_port_type *scm_make_port_type
size_t (*write) (SCM port, SCM src, size_t start, size_t count));
SCM_API void scm_set_port_scm_read (scm_t_port_type *ptob, SCM read);
SCM_API void scm_set_port_scm_write (scm_t_port_type *ptob, SCM write);
SCM_API void scm_set_port_read_wait_fd (scm_t_port_type *ptob,
int (*wait_fd) (SCM port));
SCM_API void scm_set_port_write_wait_fd (scm_t_port_type *ptob,
int (*wait_fd) (SCM port));
SCM_API void scm_set_port_print (scm_t_port_type *ptob,
int (*print) (SCM exp,
SCM port,