erl_check_io.c

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

#ifdef ERTS_USE_PORT_TASKS
	    erts_port_task_handle_init(&drv_ev_state[fd].driver.event->task);
#endif
	    drv_ev_state[fd].driver.event->port = id;
	    drv_ev_state[fd].driver.event->removed_events = (ErtsPollEvents) 0;
	    drv_ev_state[fd].type = ERTS_EV_TYPE_DRV_EV;
	}
	drv_ev_state[fd].driver.event->data = event_data;
    }
    else {
	if (drv_ev_state[fd].type == ERTS_EV_TYPE_DRV_EV) {
#ifdef ERTS_USE_PORT_TASKS
	    abort_tasks(fd, 0);
#endif
	    erts_free(ERTS_ALC_T_DRV_EV_D_STATE,
		      drv_ev_state[fd].driver.event);
	}
	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, events, drv_ev_state[fd].events);
    drv_ev_state[fd].events = events;
    ASSERT(event_data ? events == event_data->events : events == 0); 
    erts_smp_mtx_unlock(&drv_ev_state_mtx);
    return 0;
 error:
    erts_smp_mtx_unlock(&drv_ev_state_mtx);
    return -1;
#endif
}

static ERTS_INLINE int
chk_stale(Eterm id, int fd, int mode)
{
    if (is_nil(id))
	return 0;
    if (erts_is_port_alive(id))
	return 1; /* Steal */
    stale_drv_select(id, fd, mode);
    return 0;
}

static int
need2steal(int fd, int mode)
{
    int do_steal = 0;
    switch (drv_ev_state[fd].type) {
    case ERTS_EV_TYPE_DRV_SEL:
	if (mode & DO_READ)
	    do_steal |= chk_stale(drv_ev_state[fd].driver.select->inport,
				  fd,
				  DO_READ);
	if (mode & DO_WRITE)
	    do_steal |= chk_stale(drv_ev_state[fd].driver.select->outport,
				  fd,
				  DO_WRITE);
	break;
#if ERTS_CIO_HAVE_DRV_EVENT
    case ERTS_EV_TYPE_DRV_EV:
	do_steal |= chk_stale(drv_ev_state[fd].driver.event->port, fd, 0);
	break;
#endif
    default:
	break;
    }
    return do_steal;
}

static void
print_driver_name(erts_dsprintf_buf_t *dsbufp, Eterm id)
{
    ErtsPortNames *pnp = erts_get_port_names(id);
    if (!pnp->name && !pnp->driver_name)
	erts_dsprintf(dsbufp, "%s ", "<unknown>");
    else {
	if (pnp->name) {
	    if (!pnp->driver_name || strcmp(pnp->driver_name, pnp->name) == 0)
		erts_dsprintf(dsbufp, "%s ", pnp->name);
	    else
		erts_dsprintf(dsbufp, "%s (%s) ", pnp->driver_name, pnp->name);
	}
	else if (pnp->driver_name) {
	    erts_dsprintf(dsbufp, "%s ", pnp->driver_name);
	}
    }
    erts_free_port_names(pnp);
}

static void
steal(erts_dsprintf_buf_t *dsbufp, int fd, int mode)
{
    erts_dsprintf(dsbufp, "stealing control of fd=%d from ", fd);
    switch (drv_ev_state[fd].type) {
    case ERTS_EV_TYPE_DRV_SEL: {
	int deselect_mode = 0;
	Eterm iid = drv_ev_state[fd].driver.select->inport;
	Eterm oid = drv_ev_state[fd].driver.select->outport;
	if ((mode & DO_READ) && (is_not_nil(iid))) {
	    erts_dsprintf(dsbufp, "input driver ");
	    print_driver_name(dsbufp, iid);
	    erts_dsprintf(dsbufp, "%T ", iid);
	    deselect_mode |= DO_READ;
	}
	if ((mode & DO_READ) && is_not_nil(iid)
	    && (mode & DO_WRITE) && is_not_nil(oid))
	    erts_dsprintf(dsbufp, "and ");
	if ((mode & DO_WRITE) && is_not_nil(oid)) {
	    erts_dsprintf(dsbufp, "output driver ");
	    print_driver_name(dsbufp, oid);
	    erts_dsprintf(dsbufp, "%T ", oid);
	    deselect_mode |= DO_WRITE;
	}
	if (deselect_mode)
	    deselect(fd, deselect_mode);
	else {
	    erts_dsprintf(dsbufp, "no one", fd);
	    ASSERT(0);
	}
	erts_dsprintf(dsbufp, "\n");
	break;
    }
#if ERTS_CIO_HAVE_DRV_EVENT
    case ERTS_EV_TYPE_DRV_EV: {
	Eterm eid = drv_ev_state[fd].driver.event->port;
	if (is_nil(eid)) {
	    erts_dsprintf(dsbufp, "no one", fd);
	    ASSERT(0);
	}
	else {
	    erts_dsprintf(dsbufp, "event driver ");
	    print_driver_name(dsbufp, eid);
	    erts_dsprintf(dsbufp, "%T ", eid);
	}
	erts_dsprintf(dsbufp, "\n");
	deselect(fd, 0);
	break;
    }
#endif
    default:
	erts_dsprintf(dsbufp, "no one\n", fd);
	ASSERT(0);
    }
}

static void
print_select_op(erts_dsprintf_buf_t *dsbufp,
		ErlDrvPort ix, int fd, int mode, int on)
{
    Port *pp = erts_drvport2port(ix);
    erts_dsprintf(dsbufp,
		  "driver_select(%p, %d, %s%s%s, %d) "
		  "by ",
		  ix,
		  fd,
		  mode & DO_READ ? "DO_READ" : "",
		  (mode & (DO_READ|DO_WRITE)) == (DO_READ|DO_WRITE) ? "|" : "",
		  mode & DO_WRITE ? "DO_WRITE" : "",
		  on);
    print_driver_name(dsbufp, pp->id);
    erts_dsprintf(dsbufp, "driver %T ", pp ? pp->id : NIL);
}

static void
select_steal(ErlDrvPort ix, int fd, int mode, int on)
{
    if (need2steal(fd, mode)) {
	erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
	print_select_op(dsbufp, ix, fd, mode, on);
	steal(dsbufp, fd, mode);
	erts_send_error_to_logger_nogl(dsbufp);
    }
}

static void
large_fd_error_common(erts_dsprintf_buf_t *dsbufp)
{
    erts_dsprintf(dsbufp,
		  "fd=%d is larger than the largest allowed fd=%d\n",
		  max_fds - 1);
}

static void
select_large_fd_error(ErlDrvPort ix, int fd, int mode, int on)
{
    erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
    print_select_op(dsbufp, ix, fd, mode, on);
    erts_dsprintf(dsbufp, "failed: ");
    large_fd_error_common(dsbufp);
    erts_send_error_to_logger_nogl(dsbufp);
}

#if ERTS_CIO_HAVE_DRV_EVENT

static void
print_event_op(erts_dsprintf_buf_t *dsbufp,
	       ErlDrvPort ix, int fd, ErlDrvEventData event_data)
{
    Port *pp = erts_drvport2port(ix);
    erts_dsprintf(dsbufp, "driver_event(%p, %d, ", ix, fd);
    if (!event_data)
	erts_dsprintf(dsbufp, "NULL");
    else
	erts_dsprintf(dsbufp, "{0x%x, 0x%x}",
		      (unsigned int) event_data->events,
		      (unsigned int) event_data->revents);
    erts_dsprintf(dsbufp, ") by ");
    print_driver_name(dsbufp, pp->id);
    erts_dsprintf(dsbufp, "driver %T ", pp ? pp->id : NIL);
}

static void
event_steal(ErlDrvPort ix, int fd, ErlDrvEventData event_data)
{
    if (need2steal(fd, DO_READ|DO_WRITE)) {
	erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
	print_event_op(dsbufp, ix, fd, event_data);
	steal(dsbufp, fd, DO_READ|DO_WRITE);
	erts_send_error_to_logger_nogl(dsbufp);
    }
}

static void
event_large_fd_error(ErlDrvPort ix, int fd, ErlDrvEventData event_data)
{
    erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
    print_event_op(dsbufp, ix, fd, event_data);
    erts_dsprintf(dsbufp, "failed: ");
    large_fd_error_common(dsbufp);
    erts_send_error_to_logger_nogl(dsbufp);
}

#endif

static ERTS_INLINE void
bump_timers(void)
{
#if defined(ERTS_SMP_USE_IO_THREAD) && !defined(ERTS_TIMER_THREAD)
    long dt = do_time_read_and_reset();
    if (dt)
	bump_timer(dt);
#endif
}

static ERTS_INLINE void
iready(Eterm id, int fd)
{
#ifdef ERTS_USE_PORT_TASKS
    if (erts_port_task_schedule(id,
				&drv_ev_state[fd].driver.select->intask,
				ERTS_PORT_TASK_INPUT,
				(ErlDrvEvent) fd,
				NULL) != 0) {
	stale_drv_select(id, fd, DO_READ);
    }
#else
    Port *pp = erts_drvportid2port(id);
    if (!pp)
	stale_drv_select(id, fd, DO_READ);
    else {
	erts_smp_mtx_unlock(&drv_ev_state_mtx);
	erts_port_ready_input(pp, (ErlDrvEvent) fd);
	erts_smp_mtx_lock(&drv_ev_state_mtx);
    }
#endif
}

static ERTS_INLINE void
oready(Eterm id, int fd)
{
#ifdef ERTS_USE_PORT_TASKS
    if (erts_port_task_schedule(id,
				&drv_ev_state[fd].driver.select->outtask,
				ERTS_PORT_TASK_OUTPUT,
				(ErlDrvEvent) fd,
				NULL) != 0) {
	stale_drv_select(id, fd, DO_WRITE);
    }
#else
    Port *pp = erts_drvportid2port(id);
    if (!pp)
	stale_drv_select(id, fd, DO_WRITE);
    else {
	erts_smp_mtx_unlock(&drv_ev_state_mtx);
	erts_port_ready_output(pp, (ErlDrvEvent) fd);
	erts_smp_mtx_lock(&drv_ev_state_mtx);
    }
#endif
}

#if ERTS_CIO_HAVE_DRV_EVENT
static ERTS_INLINE void
eready(Eterm id, int fd, ErlDrvEventData event_data)
{
#ifdef ERTS_USE_PORT_TASKS
    if (erts_port_task_schedule(id,
				&drv_ev_state[fd].driver.event->task,
				ERTS_PORT_TASK_EVENT,
				(ErlDrvEvent) fd,
				event_data) != 0) {
	stale_drv_select(id, fd, 0);
    }
#else
    Port *pp = erts_drvportid2port(id);
    if (!pp)
	stale_drv_select(id, fd, 0);
    else {
	erts_smp_mtx_unlock(&drv_ev_state_mtx);
	erts_port_ready_event(pp, (ErlDrvEvent) fd, event_data);
	erts_smp_mtx_lock(&drv_ev_state_mtx);
    }
#endif
}
#endif

static void bad_fd_in_pollset(int, Eterm, Eterm, ErtsPollEvents);

void
ERTS_CIO_EXPORT(erts_check_io_interrupt)(int set)
{
    ERTS_CIO_POLL_INTR(pollset, set);
}

void
ERTS_CIO_EXPORT(erts_check_io_interrupt_timed)(int set, long msec)
{
    ERTS_CIO_POLL_INTR_TMD(pollset, set, msec);
}

void
ERTS_CIO_EXPORT(erts_check_io)(int do_wait)
{
    ErtsPollResFd pollres[256];
    int pollres_len;
    SysTimeval wait_time;
    int poll_ret, i;

 restart:

#ifdef ERTS_SMP_USE_IO_THREAD
    ERTS_CIO_POLL_INTR(pollset, 0);
#endif

    /* Figure out timeout value */
    if (do_wait) {
	erts_time_remaining(&wait_time);
    } else {			/* poll only */
	wait_time.tv_sec = 0;
	wait_time.tv_usec = 0;
    }

#ifdef ERTS_ENABLE_LOCK_CHECK
    erts_lc_check_exact(NULL, 0); /* No locks should be locked */
#endif
#ifndef ERTS_SMP_USE_IO_THREAD
    erts_smp_activity_begin(ERTS_ACTIVITY_WAIT, NULL, NULL, NULL);
#else
    erts_smp_activity_change(ERTS_ACTIVITY_IO,
			     ERTS_ACTIVITY_WAIT,
			     NULL,
			     NULL,
			     NULL);
#endif
    pollres_len = sizeof(pollres)/sizeof(ErtsPollResFd);
    erts_smp_atomic_set(&in_poll_wait, 1);
    poll_ret = ERTS_CIO_POLL_WAIT(pollset, pollres, &pollres_len, &wait_time);

#ifdef ERTS_ENABLE_LOCK_CHECK
    erts_lc_check_exact(NULL, 0); /* No locks should be locked */
#endif
#ifndef ERTS_SMP_USE_IO_THREAD
    erts_smp_activity_end(ERTS_ACTIVITY_WAIT, NULL, NULL, NULL);
#else
    erts_smp_activity_change(ERTS_ACTIVITY_WAIT,
			     ERTS_ACTIVITY_IO,
			     NULL,
			     NULL,
			     NULL);
#endif

    erts_deliver_time(); /* sync the machine's idea of time */

    bump_timers();

    if (ERTS_BREAK_REQUESTED)
	erts_do_break_handling();

    if (poll_ret != 0) {
	erts_smp_atomic_set(&in_poll_wait, 0);

	if (poll_ret == EAGAIN)
	    goto restart;

	if (poll_ret != ETIMEDOUT
	    && poll_ret != EINTR
	    && poll_ret != ERRNO_BLOCK) {
	    erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
	    erts_dsprintf(dsbufp, "erts_poll_wait() failed: %s (%d)\n",
			  erl_errno_id(poll_ret), poll_ret);
	    erts_send_error_to_logger_nogl(dsbufp);
	}

	return;
    }

    erts_smp_mtx_lock(&drv_ev_state_mtx);
    erts_smp_atomic_set(&in_poll_wait, 0);

    for (i = 0; i < pollres_len; i++) {
	int fd = pollres[i].fd;

	if (drv_ev_state[fd].flags & ERTS_EV_FLG_IGNORE)
	    continue;

	switch (drv_ev_state[fd].type) {
	case ERTS_EV_TYPE_DRV_SEL: { /* Requested via driver_select()... */
	    ErtsPollEvents revents;
	    ErtsPollEvents revent_mask;

	    revent_mask = ~(ERTS_POLL_EV_IN|ERTS_POLL_EV_OUT);
	    revent_mask |= drv_ev_state[fd].events;
	    revents = pollres[i].events & revent_mask;

	    if (revents & ERTS_POLL_EV_ERR) {
		/*
		 * Let the driver handle the error condition. Only input,
		 * only output, or nothing might have been selected.
		 * We *do not* want to call a callback that corresponds
		 * to an event not selected. revents might give us a clue
		 * on which one to call.
		 */ 
		if ((revents & ERTS_POLL_EV_IN)
		    || (!(revents & ERTS_POLL_EV_OUT)
			&& drv_ev_state[fd].events & ERTS_POLL_EV_IN))
		    iready(drv_ev_state[fd].driver.select->inport, fd);
		else if (drv_ev_state[fd].events & ERTS_POLL_EV_OUT)
		    oready(drv_ev_state[fd].driver.select->outport, fd);
	    }
	    else if (revents & (ERTS_POLL_EV_IN|ERTS_POLL_EV_OUT)) {
		if (revents & ERTS_POLL_EV_OUT)
		    oready(drv_ev_state[fd].driver.select->outport, fd);
		/* Someone might have deselected input since revents
		   was read (true also on the non-smp emulator since
		   oready() may have been called); therefore, update
		   revents... */
		revents &= ~(~drv_ev_state[fd].events & ERTS_POLL_EV_IN);
		if (revents & ERTS_POLL_EV_IN)
		    iready(drv_ev_state[fd].driver.select->inport, fd);
	    }
	    else if (revents & ERTS_POLL_EV_NVAL) {
		bad_fd_in_pollset(fd,
				  drv_ev_state[fd].driver.select->inport,
				  drv_ev_state[fd].driver.select->outport,
				  drv_ev_state[fd].events);
	    }
	    break;
	}

#if ERTS_CIO_HAVE_DRV_EVENT
	case ERTS_EV_TYPE_DRV_EV: { /* Requested via driver_event()... */
	    ErlDrvEventData event_data;
	    ErtsPollEvents revents;
	    ASSERT(drv_ev_state[fd].driver.event);
	    ASSERT(drv_ev_state[fd].driver.event->data);
	    event_data = drv_ev_state[fd].driver.event->data;
	    revents = pollres[i].events;
	    revents &= ~drv_ev_state[fd].driver.event->removed_events;

	    if (revents) {
		event_data->events = drv_ev_state[fd].events;
		event_data->revents = revents;

		eready(drv_ev_state[fd].driver.event->port, fd, event_data);
	    }
	    break;
	}
#endif