erl_check_io.c

来自「OTP是开放电信平台的简称」· C语言 代码 · 共 1,430 行 · 第 1/3 页

/* ``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 via the world wide web 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.
 * 
 * The Initial Developer of the Original Code is Ericsson Utvecklings AB.
 * Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
 * AB. All Rights Reserved.''
 * 
 *     $Id$
 */

/*
 * Description:	Check I/O
 *
 * Author: 	Rickard Green
 */

#ifdef HAVE_CONFIG_H
#  include "config.h"
#endif

#define ERL_CHECK_IO_C__
#define ERTS_WANT_BREAK_HANDLING
#define WANT_NONBLOCKING 
#include "sys.h"
#include "global.h"
#include "erl_check_io.h"

#define ERTS_EV_FLG_IGNORE		(((short) 1) << 0)

#define ERTS_EV_TYPE_NONE		((short) 0)
#define ERTS_EV_TYPE_DRV_SEL		((short) 1)
#define ERTS_EV_TYPE_DRV_EV		((short) 2)

#define ERTS_DRV_EV_STATE_EXTRA_SIZE 128

#if defined(ERTS_KERNEL_POLL_VERSION)
#  define ERTS_CIO_EXPORT(FUNC) FUNC ## _kp
#elif defined(ERTS_NO_KERNEL_POLL_VERSION)
#  define ERTS_CIO_EXPORT(FUNC) FUNC ## _nkp
#else
#  define ERTS_CIO_EXPORT(FUNC) FUNC
#endif

#define ERTS_CIO_HAVE_DRV_EVENT \
  (ERTS_POLL_USE_POLL && !ERTS_POLL_USE_KERNEL_POLL)

#define ERTS_CIO_POLL_CTL	ERTS_POLL_EXPORT(erts_poll_control)
#define ERTS_CIO_POLL_WAIT	ERTS_POLL_EXPORT(erts_poll_wait)
#define ERTS_CIO_POLL_INTR 	ERTS_POLL_EXPORT(erts_poll_interrupt)
#define ERTS_CIO_POLL_INTR_TMD	ERTS_POLL_EXPORT(erts_poll_interrupt_timed)
#define ERTS_CIO_NEW_POLLSET 	ERTS_POLL_EXPORT(erts_poll_create_pollset)
#define ERTS_CIO_FREE_POLLSET	ERTS_POLL_EXPORT(erts_poll_destroy_pollset)
#define ERTS_CIO_POLL_MAX_FDS	ERTS_POLL_EXPORT(erts_poll_max_fds)
#define ERTS_CIO_POLL_INIT	ERTS_POLL_EXPORT(erts_poll_init)
#define ERTS_CIO_POLL_INFO	ERTS_POLL_EXPORT(erts_poll_info)

static ErtsPollSet pollset;

typedef struct {
    union {
	ErtsDrvEventDataState *event;
	ErtsDrvSelectDataState *select;
    } driver;
    ErtsPollEvents events;
    short flags;
    short type;
} ErtsDrvEventState;

struct erts_fd_list {
    struct erts_fd_list *next;
    int fd;
};

static int max_fds = -1;
static erts_smp_mtx_t drv_ev_state_mtx;
static int drv_ev_state_len;
static ErtsDrvEventState *drv_ev_state;

static erts_smp_atomic_t in_poll_wait;

struct erts_fd_list *ignored_list;

static void stale_drv_select(Eterm id, int fd, int mode);
static void select_steal(ErlDrvPort ix, int fd, int mode, int on);
static void select_large_fd_error(ErlDrvPort, int, int, int);
#if ERTS_CIO_HAVE_DRV_EVENT
static void event_steal(ErlDrvPort ix, int fd, ErlDrvEventData event_data);
static void event_large_fd_error(ErlDrvPort, int, ErlDrvEventData);
#endif

static ERTS_INLINE Eterm
drvport2id(ErlDrvPort dp)
{
    Port *pp = erts_drvport2port(dp);
    if (pp)
	return pp->id;
    else {
	ASSERT(0);
	return am_undefined;
    }
}

ERTS_QUALLOC_IMPL(fd_list, struct erts_fd_list, 64, ERTS_ALC_T_FD_LIST)

static ERTS_INLINE void
check_ignore(int fd, ErtsPollEvents new_evs, ErtsPollEvents old_evs)
{

    if (!new_evs
	&& old_evs
	&& !(drv_ev_state[fd].flags & ERTS_EV_FLG_IGNORE)
	&& erts_smp_atomic_read(&in_poll_wait)) {
	struct erts_fd_list *fdlp = fd_list_alloc();
	fdlp->fd = fd;
	fdlp->next = ignored_list;
	ignored_list = fdlp;
	drv_ev_state[fd].flags |= ERTS_EV_FLG_IGNORE;
    }

}

static ERTS_INLINE void
reset_ignores(void)
{
    struct erts_fd_list *fdlp = ignored_list;

    while (fdlp) {
	struct erts_fd_list *ffdlp = fdlp;
	drv_ev_state[fdlp->fd].flags &= ~ERTS_EV_FLG_IGNORE;
	fdlp = fdlp->next;
	fd_list_free(ffdlp);
    }
    ignored_list = NULL;
}

static void
grow_drv_ev_state(int min_ix)
{
    int i;
    int new_len = min_ix + 1 + ERTS_DRV_EV_STATE_EXTRA_SIZE;
    if (new_len > max_fds)
	new_len = max_fds;
    drv_ev_state = (drv_ev_state_len
		     ? erts_realloc(ERTS_ALC_T_DRV_EV_STATE,
				    drv_ev_state,
				    sizeof(ErtsDrvEventState)*new_len)
		     : erts_alloc(ERTS_ALC_T_DRV_EV_STATE,
				  sizeof(ErtsDrvEventState)*new_len));
    for (i = drv_ev_state_len; i < new_len; i++) {
	drv_ev_state[i].driver.select = NULL;
	drv_ev_state[i].events = 0;
	drv_ev_state[i].flags = 0;
	drv_ev_state[i].type = ERTS_EV_TYPE_NONE;
    }
    drv_ev_state_len = new_len;
}



#ifdef ERTS_USE_PORT_TASKS

static ERTS_INLINE void
abort_task(Eterm id, ErtsPortTaskHandle *pthp, short type)
{
    if (is_nil(id)) {
	ASSERT(type == ERTS_EV_TYPE_NONE
	       || !erts_port_task_is_scheduled(pthp));
    }
    else if (erts_port_task_is_scheduled(pthp)) {
	erts_port_task_abort(pthp);
	ASSERT(erts_is_port_alive(id));
    }
}

static ERTS_INLINE void
abort_tasks(int fd, int mode)
{
    switch (mode) {
    case 0: check_type:
	switch (drv_ev_state[fd].type) {
#if ERTS_CIO_HAVE_DRV_EVENT
	case ERTS_EV_TYPE_DRV_EV:
	    abort_task(drv_ev_state[fd].driver.event->port,
		       &drv_ev_state[fd].driver.event->task,
		       ERTS_EV_TYPE_DRV_EV);
	    return;
#endif
	case ERTS_EV_TYPE_NONE:
	    return;
	default:
	    ASSERT(drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_SEL);
	    /* Fall through */
	}
    case DO_READ|DO_WRITE:
    case DO_WRITE:
	ASSERT(drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_SEL);
	abort_task(drv_ev_state[fd].driver.select->outport,
		   &drv_ev_state[fd].driver.select->outtask,
		   drv_ev_state[fd].type);
	if (mode == DO_WRITE)
	    break;
    case DO_READ:
	ASSERT(drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_SEL);
	abort_task(drv_ev_state[fd].driver.select->inport,
		   &drv_ev_state[fd].driver.select->intask,
		   drv_ev_state[fd].type);
	break;
    default:
	goto check_type;
    }
}
#endif /* ERTS_USE_PORT_TASKS */

static void
deselect(int fd, int mode)
{
    ErtsPollEvents old_events = drv_ev_state[fd].events;
    ErtsPollEvents rm_events;
    ERTS_SMP_LC_ASSERT(erts_smp_lc_mtx_is_locked(&drv_ev_state_mtx));
    ASSERT(drv_ev_state[fd].events);

#ifdef ERTS_USE_PORT_TASKS
    abort_tasks(fd, mode);
#endif /* ERTS_USE_PORT_TASKS */

    if (!mode)
	rm_events = drv_ev_state[fd].events;
    else {
	rm_events = 0;
	ASSERT(drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_SEL);
	if (mode & DO_READ) {
	    drv_ev_state[fd].driver.select->inport = NIL;
	    rm_events |= ERTS_POLL_EV_IN;
	}
	if (mode & DO_WRITE) {
	    drv_ev_state[fd].driver.select->outport = NIL;
	    rm_events |= ERTS_POLL_EV_OUT;
	}
    }

    drv_ev_state[fd].events = ERTS_CIO_POLL_CTL(pollset, fd, rm_events, 0);

    if (!drv_ev_state[fd].events) {
	switch (drv_ev_state[fd].type) {
	case ERTS_EV_TYPE_DRV_SEL:
#ifdef ERTS_USE_PORT_TASKS
	    ASSERT(!erts_port_task_is_scheduled(&drv_ev_state[fd].driver.select->intask));
	    ASSERT(!erts_port_task_is_scheduled(&drv_ev_state[fd].driver.select->outtask));
#endif
	    erts_free(ERTS_ALC_T_DRV_SEL_D_STATE,
		      drv_ev_state[fd].driver.select);
	    break;
#if ERTS_CIO_HAVE_DRV_EVENT
	case ERTS_EV_TYPE_DRV_EV:
#ifdef ERTS_USE_PORT_TASKS
	    ASSERT(!erts_port_task_is_scheduled(&drv_ev_state[fd].driver.event->task));
#endif
	    erts_free(ERTS_ALC_T_DRV_EV_D_STATE,
		      drv_ev_state[fd].driver.event);
	    break;
#endif
	case ERTS_EV_TYPE_NONE:
	    break;
	default:
	    ASSERT(0);
	    break;
	}
	    
	drv_ev_state[fd].driver.select = NULL;
	drv_ev_state[fd].flags = 0;
	drv_ev_state[fd].type = ERTS_EV_TYPE_NONE;
    }
    check_ignore(fd, drv_ev_state[fd].events, old_events);
}

int
ERTS_CIO_EXPORT(driver_select)(ErlDrvPort ix,
			       ErlDrvEvent e,
			       int mode,
			       int on)
{
    Eterm id = drvport2id(ix);
    int fd = (int) e;
    ErtsPollEvents ctl_events = (ErtsPollEvents) 0;
    ErtsPollEvents new_events, old_events;

    ERTS_SMP_LC_ASSERT(erts_drvport2port(ix)
		       && erts_lc_is_port_locked(erts_drvport2port(ix)));

    if (fd < 0)
	return -1;

    if (fd >= max_fds) {
	select_large_fd_error(ix, fd, mode, on);
	return -1;
    }

    erts_smp_mtx_lock(&drv_ev_state_mtx);

    if (fd >= drv_ev_state_len)
	grow_drv_ev_state(fd);

#if ERTS_CIO_HAVE_DRV_EVENT
    if (drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_EV)
	select_steal(ix, fd, mode, on);;
#endif

    if (mode & DO_READ) {
	if (drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_SEL) {
	    Eterm owner = drv_ev_state[fd].driver.select->inport;
	    if (owner != id && is_not_nil(owner))
		select_steal(ix, fd, mode, on);
	}
	ctl_events |= ERTS_POLL_EV_IN;
    }
    if (mode & DO_WRITE) {
	if (drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_SEL) {
	    Eterm owner = drv_ev_state[fd].driver.select->outport;
	    if (owner != id && is_not_nil(owner))
		select_steal(ix, fd, mode, on);
	}
	ctl_events |= ERTS_POLL_EV_OUT;
    }	

    ASSERT(drv_ev_state[fd].events
	   ? (drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_SEL)
	   : (drv_ev_state[fd].type == ERTS_EV_TYPE_NONE));

    new_events = ERTS_CIO_POLL_CTL(pollset, fd, ctl_events, on);
    if (new_events & (ERTS_POLL_EV_ERR|ERTS_POLL_EV_NVAL))
	goto error;

    old_events = drv_ev_state[fd].events;

    ASSERT(on
	   ? (new_events == (drv_ev_state[fd].events | ctl_events))
	   : (new_events == (drv_ev_state[fd].events & ~ctl_events)));

    ASSERT(drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_SEL
	   || drv_ev_state[fd].type == ERTS_EV_TYPE_NONE);

    drv_ev_state[fd].events = new_events;
    if (ctl_events) {
	if (on) {
	    if (drv_ev_state[fd].type == ERTS_EV_TYPE_NONE) {
		ErtsDrvSelectDataState *dsdsp
		    = erts_alloc(ERTS_ALC_T_DRV_SEL_D_STATE,
				 sizeof(ErtsDrvSelectDataState));
		dsdsp->inport = NIL;
		dsdsp->outport = NIL;
#ifdef ERTS_USE_PORT_TASKS
		erts_port_task_handle_init(&dsdsp->intask);
		erts_port_task_handle_init(&dsdsp->outtask);
#endif
		drv_ev_state[fd].driver.select = dsdsp;
		drv_ev_state[fd].type = ERTS_EV_TYPE_DRV_SEL;
	    }
	    ASSERT(drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_SEL);
	    if (ctl_events & ERTS_POLL_EV_IN)
		drv_ev_state[fd].driver.select->inport = id;
	    if (ctl_events & ERTS_POLL_EV_OUT)
		drv_ev_state[fd].driver.select->outport = id;
	}
	else if (drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_SEL) {
	    if (ctl_events & ERTS_POLL_EV_IN) {
#ifdef ERTS_USE_PORT_TASKS
		abort_tasks(fd, DO_READ);
#endif
		drv_ev_state[fd].driver.select->inport = NIL;
	    }
	    if (ctl_events & ERTS_POLL_EV_OUT) {
#ifdef ERTS_USE_PORT_TASKS
		abort_tasks(fd, DO_WRITE);
#endif
		drv_ev_state[fd].driver.select->outport = NIL;
	    }
	    if (new_events == 0) {
#ifdef ERTS_USE_PORT_TASKS
		ASSERT(!erts_port_task_is_scheduled(&drv_ev_state[fd].driver.select->intask));
		ASSERT(!erts_port_task_is_scheduled(&drv_ev_state[fd].driver.select->outtask));
#endif
		erts_free(ERTS_ALC_T_DRV_SEL_D_STATE,
			  drv_ev_state[fd].driver.select);
		drv_ev_state[fd].driver.select = NULL;
		drv_ev_state[fd].flags = 0;
		drv_ev_state[fd].type = ERTS_EV_TYPE_NONE;
	    }
	}
    }
    check_ignore(fd, new_events, old_events);
    erts_smp_mtx_unlock(&drv_ev_state_mtx);

    return 0;

 error:
    erts_smp_mtx_unlock(&drv_ev_state_mtx);
    return -1;
}

int
ERTS_CIO_EXPORT(driver_event)(ErlDrvPort ix,
			      ErlDrvEvent e,
			      ErlDrvEventData event_data)
{
#if !ERTS_CIO_HAVE_DRV_EVENT
    return -1;
#else
    int fd = (int) e;
    ErtsPollEvents events;
    ErtsPollEvents add_events;
    ErtsPollEvents remove_events;
    Eterm id = drvport2id(ix);

    ERTS_SMP_LC_ASSERT(erts_drvport2port(ix)
		       && erts_lc_is_port_locked(erts_drvport2port(ix)));

    if (fd < 0)
	return -1;

    if (fd >= max_fds) {
	event_large_fd_error(ix, fd, event_data);
	return -1;
    }

    erts_smp_mtx_lock(&drv_ev_state_mtx);

    if (fd >= drv_ev_state_len)
	grow_drv_ev_state(fd);

    if (drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_SEL
	|| (drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_EV
	    && drv_ev_state[fd].driver.event->port != id)) {
	event_steal(ix, fd, event_data);
    }

    ASSERT(drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_EV
	   || drv_ev_state[fd].type == ERTS_EV_TYPE_NONE);

    events = drv_ev_state[fd].events;

    if (!event_data) {
	remove_events = events;
	add_events = 0;
    }
    else {
	remove_events = ~event_data->events & events;
	add_events = ~events & event_data->events;

    }

    if (add_events) {
	events = ERTS_CIO_POLL_CTL(pollset, fd, add_events, 1);
	if (events & (ERTS_POLL_EV_ERR|ERTS_POLL_EV_NVAL))
	    goto error;
    }
    if (remove_events) {
	events = ERTS_CIO_POLL_CTL(pollset, fd, remove_events, 0);
	if (events & (ERTS_POLL_EV_ERR|ERTS_POLL_EV_NVAL))
	    goto error;
    }
    if (event_data) {
	if (drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_EV) {
	    drv_ev_state[fd].driver.event->removed_events &= ~add_events;
	    drv_ev_state[fd].driver.event->removed_events |= remove_events;
	}
	else {
	    drv_ev_state[fd].driver.event
		= erts_alloc(ERTS_ALC_T_DRV_EV_D_STATE,
			     sizeof(ErtsDrvEventDataState));