Use reduction limit in order to determine when to do wakeup

3 files changed, 55 insertions, 16 deletions

diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c
index 0f1a0d441a..31c92724cf 100644
--- a/erts/emulator/beam/erl_port_task.c
+++ b/erts/emulator/beam/erl_port_task.c
@@ -984,8 +984,9 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
 #endif
 
  done:
-    ERTS_SMP_LC_ASSERT(erts_smp_lc_runq_is_locked(runq));
+    runq->scheduler->reductions += reds;
 
+    ERTS_SMP_LC_ASSERT(erts_smp_lc_runq_is_locked(runq));
     ERTS_PORT_REDUCTIONS_EXECUTED(runq, reds);
 
     return res;
diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c
index 0fa2def5af..c5127bc29d 100644
--- a/erts/emulator/beam/erl_process.c
+++ b/erts/emulator/beam/erl_process.c
@@ -43,6 +43,9 @@
 #include "erl_async.h"
 #include "dtrace-wrapper.h"
 
+#define ERTS_DELAYED_WAKEUP_INFINITY (~(Uint64) 0)
+#define ERTS_DELAYED_WAKEUP_REDUCTIONS ((Uint64) CONTEXT_REDS/2)
+
 #define ERTS_RUNQ_CHECK_BALANCE_REDS_PER_SCHED (2000*CONTEXT_REDS)
 #define ERTS_RUNQ_CALL_CHECK_BALANCE_REDS \
   (ERTS_RUNQ_CHECK_BALANCE_REDS_PER_SCHED/2)
@@ -932,9 +935,15 @@ haw_thr_prgr_current_check_progress(ErtsAuxWorkData *awdp)
 }
 
 static ERTS_INLINE erts_aint32_t
-handle_delayed_aux_work_wakeup(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
+handle_delayed_aux_work_wakeup(ErtsAuxWorkData *awdp, erts_aint32_t aux_work, int waiting)
 {
     int jix, max_jix;
+
+    ASSERT(awdp->delayed_wakeup.next != ERTS_DELAYED_WAKEUP_INFINITY);
+
+    if (!waiting && awdp->delayed_wakeup.next > awdp->esdp->reductions)
+	return aux_work;
+
     unset_aux_work_flags(awdp->ssi, ERTS_SSI_AUX_WORK_DELAYED_AW_WAKEUP);
 
     ERTS_THR_MEMORY_BARRIER;
@@ -950,11 +959,14 @@ handle_delayed_aux_work_wakeup(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
 	set_aux_work_flags_wakeup_nob(ERTS_SCHED_SLEEP_INFO_IX(sched-1),
 				      aux_work);
     }
+    awdp->delayed_wakeup.next = ERTS_DELAYED_WAKEUP_INFINITY;
     return aux_work & ~ERTS_SSI_AUX_WORK_DELAYED_AW_WAKEUP;
 }
 
 static ERTS_INLINE void
-schedule_aux_work_wakeup(ErtsAuxWorkData *awdp, int sched, erts_aint32_t aux_work)
+schedule_aux_work_wakeup(ErtsAuxWorkData *awdp,
+			 int sched,
+			 erts_aint32_t aux_work)
 {
     int jix = awdp->delayed_wakeup.sched2jix[sched];
     if (jix >= 0) {
@@ -967,7 +979,20 @@ schedule_aux_work_wakeup(ErtsAuxWorkData *awdp, int sched, erts_aint32_t aux_wor
 	awdp->delayed_wakeup.job[jix].sched = sched;
 	awdp->delayed_wakeup.job[jix].aux_work = aux_work;
     }
-    set_aux_work_flags_wakeup_nob(awdp->ssi, ERTS_SSI_AUX_WORK_DELAYED_AW_WAKEUP);
+
+    if (awdp->delayed_wakeup.next != ERTS_DELAYED_WAKEUP_INFINITY) {
+	ASSERT(erts_atomic32_read_nob(&awdp->ssi->aux_work)
+	       & ERTS_SSI_AUX_WORK_DELAYED_AW_WAKEUP);
+    }
+    else {
+	awdp->delayed_wakeup.next = (awdp->esdp->reductions
+				     + ERTS_DELAYED_WAKEUP_REDUCTIONS);
+
+	ASSERT(!(erts_atomic32_read_nob(&awdp->ssi->aux_work)
+		 & ERTS_SSI_AUX_WORK_DELAYED_AW_WAKEUP));
+	set_aux_work_flags_wakeup_nob(awdp->ssi,
+				      ERTS_SSI_AUX_WORK_DELAYED_AW_WAKEUP);
+    }
 }
 
 #endif
@@ -1046,7 +1071,8 @@ misc_aux_work_clean(ErtsThrQ_t *q,
 
 static ERTS_INLINE erts_aint32_t
 handle_misc_aux_work(ErtsAuxWorkData *awdp,
-		     erts_aint32_t aux_work)
+		     erts_aint32_t aux_work,
+		     int waiting)
 {
     ErtsThrQ_t *q = &misc_aux_work_queues[awdp->sched_id].q;
 
@@ -1066,7 +1092,8 @@ handle_misc_aux_work(ErtsAuxWorkData *awdp,
 
 static ERTS_INLINE erts_aint32_t
 handle_misc_aux_work_thr_prgr(ErtsAuxWorkData *awdp,
-			      erts_aint32_t aux_work)
+			      erts_aint32_t aux_work,
+			      int waiting)
 {
     if (!erts_thr_progress_has_reached_this(haw_thr_prgr_current(awdp),
 					    awdp->misc.thr_prgr))
@@ -1145,7 +1172,8 @@ erts_notify_check_async_ready_queue(void *vno)
 
 static ERTS_INLINE erts_aint32_t
 handle_async_ready(ErtsAuxWorkData *awdp,
-		   erts_aint32_t aux_work)
+		   erts_aint32_t aux_work,
+		   int waiting)
 {
     ErtsSchedulerSleepInfo *ssi = awdp->ssi;
     unset_aux_work_flags(ssi, ERTS_SSI_AUX_WORK_ASYNC_READY);
@@ -1167,7 +1195,8 @@ handle_async_ready(ErtsAuxWorkData *awdp,
 
 static ERTS_INLINE erts_aint32_t
 handle_async_ready_clean(ErtsAuxWorkData *awdp,
-			 erts_aint32_t aux_work)
+			 erts_aint32_t aux_work,
+			 int waiting)
 {
     void *thr_prgr_p;
 
@@ -1203,7 +1232,7 @@ handle_async_ready_clean(ErtsAuxWorkData *awdp,
 #endif
 
 static ERTS_INLINE erts_aint32_t
-handle_fix_alloc(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
+handle_fix_alloc(ErtsAuxWorkData *awdp, erts_aint32_t aux_work, int waiting)
 {
     ErtsSchedulerSleepInfo *ssi = awdp->ssi;
     erts_aint32_t res;
@@ -1237,7 +1266,7 @@ erts_alloc_notify_delayed_dealloc(int ix)
 }
 
 static ERTS_INLINE erts_aint32_t
-handle_delayed_dealloc(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
+handle_delayed_dealloc(ErtsAuxWorkData *awdp, erts_aint32_t aux_work, int waiting)
 {
     ErtsSchedulerSleepInfo *ssi = awdp->ssi;
     int need_thr_progress = 0;
@@ -1275,7 +1304,7 @@ handle_delayed_dealloc(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
 }
 
 static ERTS_INLINE erts_aint32_t
-handle_delayed_dealloc_thr_prgr(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
+handle_delayed_dealloc_thr_prgr(ErtsAuxWorkData *awdp, erts_aint32_t aux_work, int waiting)
 {
     ErtsSchedulerSleepInfo *ssi;
     int need_thr_progress;
@@ -1403,7 +1432,7 @@ erts_smp_notify_check_children_needed(void)
 }
 
 static ERTS_INLINE erts_aint32_t
-handle_check_children(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
+handle_check_children(ErtsAuxWorkData *awdp, erts_aint32_t aux_work, int waiting)
 {
     unset_aux_work_flags(awdp->ssi, ERTS_SSI_AUX_WORK_CHECK_CHILDREN);
     erts_check_children();
@@ -1426,7 +1455,7 @@ erts_smp_atomic32_t erts_halt_progress;
 int erts_halt_code;
 
 static ERTS_INLINE erts_aint32_t
-handle_reap_ports(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
+handle_reap_ports(ErtsAuxWorkData *awdp, erts_aint32_t aux_work, int waiting)
 {
     unset_aux_work_flags(awdp->ssi, ERTS_SSI_AUX_WORK_REAP_PORTS);
     awdp->esdp->run_queue->halt_in_progress = 1;
@@ -1474,7 +1503,7 @@ handle_reap_ports(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
 #if HAVE_ERTS_MSEG
 
 static ERTS_INLINE erts_aint32_t
-handle_mseg_cache_check(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
+handle_mseg_cache_check(ErtsAuxWorkData *awdp, erts_aint32_t aux_work, int waiting)
 {
     unset_aux_work_flags(awdp->ssi, ERTS_SSI_AUX_WORK_MSEG_CACHE_CHECK);
     erts_mseg_cache_check();
@@ -1484,7 +1513,7 @@ handle_mseg_cache_check(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
 #endif
 
 static ERTS_INLINE erts_aint32_t
-handle_setup_aux_work_timer(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
+handle_setup_aux_work_timer(ErtsAuxWorkData *awdp, erts_aint32_t aux_work, int waiting)
 {
     unset_aux_work_flags(awdp->ssi, ERTS_SSI_AUX_WORK_SET_TMO);
     setup_aux_work_timer();
@@ -1498,7 +1527,7 @@ handle_aux_work(ErtsAuxWorkData *awdp, erts_aint32_t orig_aux_work, int waiting)
 #define HANDLE_AUX_WORK(FLG, HNDLR) \
     ignore |= FLG; \
     if (aux_work & FLG) { \
-	aux_work = HNDLR(awdp, aux_work); \
+	aux_work = HNDLR(awdp, aux_work, waiting);   \
 	ERTS_DBG_CHK_AUX_WORK_VAL(aux_work); \
 	if (!(aux_work & ~ignore)) { \
 	    ERTS_DBG_CHK_AUX_WORK_VAL(aux_work); \
@@ -3917,6 +3946,7 @@ init_aux_work_data(ErtsAuxWorkData *awdp, ErtsSchedulerData *esdp, char *dawwp)
     awdp->async_ready.queue = NULL;
 #endif
 #ifdef ERTS_SMP
+    awdp->delayed_wakeup.next = ERTS_DELAYED_WAKEUP_INFINITY;
     if (!dawwp) {
 	awdp->delayed_wakeup.job = NULL;
 	awdp->delayed_wakeup.sched2jix = NULL;
@@ -4121,6 +4151,9 @@ erts_init_scheduling(int no_schedulers, int no_schedulers_online)
 #ifdef ERTS_SMP
 	daww_ptr += daww_sz;
 #endif
+
+	esdp->reductions = 0;
+
 	init_sched_wall_time(&esdp->sched_wall_time);
     }
 
@@ -6664,6 +6697,9 @@ Process *schedule(Process *p, int calls)
 		   || p->rcount == 0);
 	}
 #endif
+
+	esdp->reductions += reds;
+
 	erts_smp_runq_lock(rq);
 
 	ERTS_PROC_REDUCTIONS_EXECUTED(rq, p->prio, reds, actual_reds);
diff --git a/erts/emulator/beam/erl_process.h b/erts/emulator/beam/erl_process.h
index 9e7a5a5c74..7c481a91dd 100644
--- a/erts/emulator/beam/erl_process.h
+++ b/erts/emulator/beam/erl_process.h
@@ -439,6 +439,7 @@ typedef struct {
 #endif
 #ifdef ERTS_SMP
     struct {
+	Uint64 next;
 	int *sched2jix;
 	int jix;
 	ErtsDelayedAuxWorkWakeupJob *job;
@@ -481,6 +482,7 @@ struct ErtsSchedulerData_ {
 
     ErtsSchedAllocData alloc_data;
 
+    Uint64 reductions;
     ErtsSchedWallTime sched_wall_time;
 
 #ifdef ERTS_DO_VERIFY_UNUSED_TEMP_ALLOC