diff options
Diffstat (limited to 'erts/emulator/beam/erl_port_task.c')
-rw-r--r-- | erts/emulator/beam/erl_port_task.c | 119 |
1 files changed, 56 insertions, 63 deletions
diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c index e6b55c45e4..3dc7c14faf 100644 --- a/erts/emulator/beam/erl_port_task.c +++ b/erts/emulator/beam/erl_port_task.c @@ -1,7 +1,7 @@ /* * %CopyrightBegin% * - * Copyright Ericsson AB 2006-2011. All Rights Reserved. + * Copyright Ericsson AB 2006-2012. 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 @@ -32,6 +32,7 @@ #include "global.h" #include "erl_port_task.h" #include "dist.h" +#include "dtrace-wrapper.h" #if defined(DEBUG) && 0 #define HARD_DEBUG @@ -61,6 +62,20 @@ do { \ (P)->sched.next = NULL; \ } while (0) +#ifdef USE_VM_PROBES +#define DTRACE_DRIVER(PROBE_NAME, PP) \ + if (DTRACE_ENABLED(driver_ready_input)) { \ + DTRACE_CHARBUF(process_str, DTRACE_TERM_BUF_SIZE); \ + DTRACE_CHARBUF(port_str, DTRACE_TERM_BUF_SIZE); \ + \ + dtrace_pid_str(PP->connected, process_str); \ + dtrace_port_str(PP, port_str); \ + DTRACE3(PROBE_NAME, process_str, port_str, PP->name); \ + } +#else +#define DTRACE_DRIVER(PROBE_NAME, PP) do {} while(0) +#endif + erts_smp_atomic_t erts_port_task_outstanding_io_tasks; struct ErtsPortTaskQueue_ { @@ -121,7 +136,7 @@ ERTS_SCHED_PREF_QUICK_ALLOC_IMPL(port_taskq, static ERTS_INLINE ErtsPortTask * handle2task(ErtsPortTaskHandle *pthp) { - return (ErtsPortTask *) erts_smp_atomic_read(pthp); + return (ErtsPortTask *) erts_smp_atomic_read_nob(pthp); } static ERTS_INLINE void @@ -129,7 +144,7 @@ reset_handle(ErtsPortTask *ptp) { if (ptp->handle) { ASSERT(ptp == handle2task(ptp->handle)); - erts_smp_atomic_set(ptp->handle, (erts_aint_t) NULL); + erts_smp_atomic_set_nob(ptp->handle, (erts_aint_t) NULL); } } @@ -138,7 +153,7 @@ set_handle(ErtsPortTask *ptp, ErtsPortTaskHandle *pthp) { ptp->handle = pthp; if (pthp) { - erts_smp_atomic_set(pthp, (erts_aint_t) ptp); + erts_smp_atomic_set_nob(pthp, (erts_aint_t) ptp); ASSERT(ptp == handle2task(ptp->handle)); } } @@ -479,8 +494,8 @@ erts_port_task_abort(Eterm id, ErtsPortTaskHandle *pthp) case ERTS_PORT_TASK_INPUT: case ERTS_PORT_TASK_OUTPUT: case ERTS_PORT_TASK_EVENT: - ASSERT(erts_smp_atomic_read(&erts_port_task_outstanding_io_tasks) > 0); - erts_smp_atomic_dec(&erts_port_task_outstanding_io_tasks); + ASSERT(erts_smp_atomic_read_nob(&erts_port_task_outstanding_io_tasks) > 0); + erts_smp_atomic_dec_relb(&erts_port_task_outstanding_io_tasks); break; default: break; @@ -558,26 +573,26 @@ erts_port_task_schedule(Eterm id, ERTS_PT_CHK_PRES_PORTQ(runq, pp); - if (!pp->sched.taskq) { - pp->sched.taskq = port_taskq_init(port_taskq_alloc(), pp); - enq_port = !pp->sched.exe_taskq; - } - + if (!pp->sched.taskq && !pp->sched.exe_taskq) { #ifdef ERTS_SMP - if (enq_port) { ErtsRunQueue *xrunq = erts_check_emigration_need(runq, ERTS_PORT_PRIO_LEVEL); if (xrunq) { /* Port emigrated ... */ - erts_smp_atomic_set(&pp->run_queue, (erts_aint_t) xrunq); + erts_smp_atomic_set_nob(&pp->run_queue, (erts_aint_t) xrunq); erts_smp_runq_unlock(runq); runq = xrunq; } - } + enq_port = !pp->sched.taskq && !pp->sched.exe_taskq; +#else + enq_port = 1; #endif + } ASSERT(!enq_port || !(runq->flags & ERTS_RUNQ_FLG_SUSPENDED)); - ASSERT(pp->sched.taskq); + if (!pp->sched.taskq) + pp->sched.taskq = port_taskq_init(port_taskq_alloc(), pp); + ASSERT(ptp); ptp->type = type; @@ -594,7 +609,7 @@ erts_port_task_schedule(Eterm id, case ERTS_PORT_TASK_INPUT: case ERTS_PORT_TASK_OUTPUT: case ERTS_PORT_TASK_EVENT: - erts_smp_atomic_inc(&erts_port_task_outstanding_io_tasks); + erts_smp_atomic_inc_relb(&erts_port_task_outstanding_io_tasks); /* Fall through... */ default: enqueue_task(pp->sched.taskq, ptp); @@ -662,7 +677,7 @@ erts_port_task_free_port(Port *pp) pp->status |= ERTS_PORT_SFLG_FREE_SCHEDULED; erts_may_save_closed_port(pp); erts_smp_port_state_unlock(pp); - ERTS_SMP_LC_ASSERT(erts_smp_atomic_read(&pp->refc) > 1); + ERTS_SMP_LC_ASSERT(erts_smp_atomic_read_nob(&pp->refc) > 1); ptp->type = ERTS_PORT_TASK_FREE; ptp->event = (ErlDrvEvent) -1; ptp->event_data = NULL; @@ -684,9 +699,9 @@ erts_port_task_free_port(Port *pp) erts_may_save_closed_port(pp); erts_smp_port_state_unlock(pp); #ifdef ERTS_SMP - erts_smp_atomic_dec(&pp->refc); /* Not alive */ + erts_smp_atomic_dec_nob(&pp->refc); /* Not alive */ #endif - ERTS_SMP_LC_ASSERT(erts_smp_atomic_read(&pp->refc) > 0); /* Lock */ + ERTS_SMP_LC_ASSERT(erts_smp_atomic_read_nob(&pp->refc) > 0); /* Lock */ handle_remaining_tasks(runq, pp); /* May release runq lock */ ASSERT(!pp->sched.exe_taskq && (!ptqp || !ptqp->first)); pp->sched.taskq = NULL; @@ -711,23 +726,6 @@ typedef struct { int *resp; } ErtsPortTaskExeBlockData; -static void -prepare_for_block(void *vd) -{ - ErtsPortTaskExeBlockData *d = (ErtsPortTaskExeBlockData *) vd; - erts_smp_runq_unlock(d->runq); -} - -static void -resume_after_block(void *vd) -{ - ErtsPortTaskExeBlockData *d = (ErtsPortTaskExeBlockData *) vd; - erts_smp_runq_lock(d->runq); - if (d->resp) - *d->resp = (erts_smp_atomic_read(&erts_port_task_outstanding_io_tasks) - != (erts_aint_t) 0); -} - /* * Run all scheduled tasks for the first port in run queue. If * new tasks appear while running reschedule port (free task is @@ -748,15 +746,9 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) int reds = ERTS_PORT_REDS_EXECUTE; erts_aint_t io_tasks_executed = 0; int fpe_was_unmasked; - ErtsPortTaskExeBlockData blk_data = {runq, NULL}; ERTS_SMP_LC_ASSERT(erts_smp_lc_runq_is_locked(runq)); - erts_smp_activity_begin(ERTS_ACTIVITY_IO, - prepare_for_block, - resume_after_block, - (void *) &blk_data); - ERTS_PT_CHK_PORTQ(runq); pp = pop_port(runq); @@ -832,8 +824,8 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) ASSERT(!ptqp->first && (!pp->sched.taskq || !pp->sched.taskq->first)); #ifdef ERTS_SMP - erts_smp_atomic_dec(&pp->refc); /* Not alive */ - ERTS_SMP_LC_ASSERT(erts_smp_atomic_read(&pp->refc) > 0); /* Lock */ + erts_smp_atomic_dec_nob(&pp->refc); /* Not alive */ + ERTS_SMP_LC_ASSERT(erts_smp_atomic_read_nob(&pp->refc) > 0); /* Lock */ #else erts_port_status_bor_set(pp, ERTS_PORT_SFLG_FREE); #endif @@ -846,12 +838,15 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) goto tasks_done; case ERTS_PORT_TASK_TIMEOUT: reds += ERTS_PORT_REDS_TIMEOUT; - if (!(pp->status & ERTS_PORT_SFLGS_DEAD)) + if (!(pp->status & ERTS_PORT_SFLGS_DEAD)) { + DTRACE_DRIVER(driver_timeout, pp); (*pp->drv_ptr->timeout)((ErlDrvData) pp->drv_data); + } break; case ERTS_PORT_TASK_INPUT: reds += ERTS_PORT_REDS_INPUT; ASSERT((pp->status & ERTS_PORT_SFLGS_DEAD) == 0); + DTRACE_DRIVER(driver_ready_input, pp); /* NOTE some windows drivers use ->ready_input for input and output */ (*pp->drv_ptr->ready_input)((ErlDrvData) pp->drv_data, ptp->event); io_tasks_executed++; @@ -859,12 +854,14 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) case ERTS_PORT_TASK_OUTPUT: reds += ERTS_PORT_REDS_OUTPUT; ASSERT((pp->status & ERTS_PORT_SFLGS_DEAD) == 0); + DTRACE_DRIVER(driver_ready_output, pp); (*pp->drv_ptr->ready_output)((ErlDrvData) pp->drv_data, ptp->event); io_tasks_executed++; break; case ERTS_PORT_TASK_EVENT: reds += ERTS_PORT_REDS_EVENT; ASSERT((pp->status & ERTS_PORT_SFLGS_DEAD) == 0); + DTRACE_DRIVER(driver_event, pp); (*pp->drv_ptr->event)((ErlDrvData) pp->drv_data, ptp->event, ptp->event_data); io_tasks_executed++; break; @@ -906,14 +903,16 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) erts_unblock_fpe(fpe_was_unmasked); if (io_tasks_executed) { - ASSERT(erts_smp_atomic_read(&erts_port_task_outstanding_io_tasks) >= io_tasks_executed); - erts_smp_atomic_add(&erts_port_task_outstanding_io_tasks, -1*io_tasks_executed); + ASSERT(erts_smp_atomic_read_nob(&erts_port_task_outstanding_io_tasks) + >= io_tasks_executed); + erts_smp_atomic_add_relb(&erts_port_task_outstanding_io_tasks, + -1*io_tasks_executed); } *curr_port_pp = NULL; #ifdef ERTS_SMP - ASSERT(runq == (ErtsRunQueue *) erts_smp_atomic_read(&pp->run_queue)); + ASSERT(runq == (ErtsRunQueue *) erts_smp_atomic_read_nob(&pp->run_queue)); #endif if (!pp->sched.taskq) { @@ -940,7 +939,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) } else { /* Port emigrated ... */ - erts_smp_atomic_set(&pp->run_queue, (erts_aint_t) xrunq); + erts_smp_atomic_set_nob(&pp->run_queue, (erts_aint_t) xrunq); enqueue_port(xrunq, pp); ASSERT(pp->sched.exe_taskq); pp->sched.exe_taskq = NULL; @@ -951,7 +950,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) port_was_enqueued = 1; } - res = (erts_smp_atomic_read(&erts_port_task_outstanding_io_tasks) + res = (erts_smp_atomic_read_nob(&erts_port_task_outstanding_io_tasks) != (erts_aint_t) 0); ERTS_PT_CHK_PRES_PORTQ(runq, pp); @@ -972,27 +971,22 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) { erts_aint_t refc; erts_smp_mtx_unlock(pp->lock); - refc = erts_smp_atomic_dectest(&pp->refc); + refc = erts_smp_atomic_dec_read_nob(&pp->refc); ASSERT(refc >= 0); if (refc == 0) { erts_smp_runq_unlock(runq); erts_port_cleanup(pp); /* Might aquire runq lock */ erts_smp_runq_lock(runq); - res = (erts_smp_atomic_read(&erts_port_task_outstanding_io_tasks) + res = (erts_smp_atomic_read_nob(&erts_port_task_outstanding_io_tasks) != (erts_aint_t) 0); } } #endif done: - blk_data.resp = &res; - erts_smp_activity_end(ERTS_ACTIVITY_IO, - prepare_for_block, - resume_after_block, - (void *) &blk_data); + runq->scheduler->reductions += reds; ERTS_SMP_LC_ASSERT(erts_smp_lc_runq_is_locked(runq)); - ERTS_PORT_REDUCTIONS_EXECUTED(runq, reds); return res; @@ -1072,8 +1066,6 @@ erts_port_migrate(Port *prt, int *prt_locked, ERTS_SMP_LC_CHK_RUNQ_LOCK(from_rq, *from_locked); ERTS_SMP_LC_CHK_RUNQ_LOCK(to_rq, *to_locked); - ASSERT(!erts_common_run_queue); - if (!*from_locked || !*to_locked) { if (from_rq < to_rq) { if (!*to_locked) { @@ -1107,12 +1099,12 @@ erts_port_migrate(Port *prt, int *prt_locked, /* Refuse to migrate to a suspended run queue */ if (to_rq->flags & ERTS_RUNQ_FLG_SUSPENDED) return ERTS_MIGRATE_FAILED_RUNQ_SUSPENDED; - if (from_rq != (ErtsRunQueue *) erts_smp_atomic_read(&prt->run_queue)) + if (from_rq != (ErtsRunQueue *) erts_smp_atomic_read_nob(&prt->run_queue)) return ERTS_MIGRATE_FAILED_RUNQ_CHANGED; if (!ERTS_PORT_IS_IN_RUNQ(from_rq, prt)) return ERTS_MIGRATE_FAILED_NOT_IN_RUNQ; dequeue_port(from_rq, prt); - erts_smp_atomic_set(&prt->run_queue, (erts_aint_t) to_rq); + erts_smp_atomic_set_nob(&prt->run_queue, (erts_aint_t) to_rq); enqueue_port(to_rq, prt); return ERTS_MIGRATE_SUCCESS; } @@ -1125,7 +1117,8 @@ erts_port_migrate(Port *prt, int *prt_locked, void erts_port_task_init(void) { - erts_smp_atomic_init(&erts_port_task_outstanding_io_tasks, (erts_aint_t) 0); + erts_smp_atomic_init_nob(&erts_port_task_outstanding_io_tasks, + (erts_aint_t) 0); init_port_task_alloc(); init_port_taskq_alloc(); } |