Teach call count tracing to use atomics

Call count previously used a global lock for accessing and writing
its counter in the breakpoint. This is now changed to atomics instead.

The change will let call count tracing and cprof to scale better
when increasing the number of schedulers.

3 files changed, 25 insertions, 22 deletions

diff --git a/erts/emulator/beam/beam_bp.c b/erts/emulator/beam/beam_bp.c
index 79d150c9e2..424e3324f9 100644
--- a/erts/emulator/beam/beam_bp.c
+++ b/erts/emulator/beam/beam_bp.c
@@ -439,9 +439,7 @@ erts_is_count_break(BeamInstr *pc, Sint *count_ret) {
     
     if (bdc) {
 	if (count_ret) {
-	    ErtsSmpBPLock(bdc);
-	    *count_ret = bdc->count;
-	    ErtsSmpBPUnlock(bdc);
+	    *count_ret = (Sint) erts_smp_atomic_read(&bdc->acount);
 	}
 	return !0;
     }
@@ -976,16 +974,23 @@ static int set_function_break(Module *modp, BeamInstr *pc,
 	    ASSERT(is_nil(tracer_pid));
 	    if (break_op == (BeamInstr) BeamOp(op_i_count_breakpoint)) {
 		BpDataCount *bdc = (BpDataCount *) bd;
+		long count = 0;
+		long res   = 0;
 		
-		ErtsSmpBPLock(bdc);
 		if (count_op == erts_break_stop) {
-		    if (bdc->count >= 0) {
-			bdc->count = -bdc->count-1; /* Stop call counter */
+		    count = erts_smp_atomic_read(&bdc->acount);
+		    if (count >= 0) {
+			while(1) {
+			    res = erts_smp_atomic_cmpxchg(&bdc->acount, -count - 1, count);
+			    if ((res == count) || count < 0) break;
+			    count = res;
+			}
 		    }
 		} else {
-		    bdc->count = 0; /* Reset call counter */
+		    /* Reset call counter */
+		    erts_smp_atomic_set(&bdc->acount, 0);
 		}
-		ErtsSmpBPUnlock(bdc);
+
 	    } else if (break_op == (Uint) BeamOp(op_i_time_breakpoint)) {
 		BpDataTime *bdt = (BpDataTime *) bd;
 		Uint i = 0;
@@ -1015,8 +1020,7 @@ static int set_function_break(Module *modp, BeamInstr *pc,
 	ASSERT(! match_spec);
 	ASSERT(is_nil(tracer_pid));
 	if (break_op == (BeamInstr) BeamOp(op_i_count_breakpoint)) {
-	    if (count_op == erts_break_reset
-		|| count_op == erts_break_stop) {
+	    if (count_op == erts_break_reset || count_op == erts_break_stop) {
 		/* Do not insert a new breakpoint */
 		return 1;
 	    }
@@ -1108,8 +1112,7 @@ static int set_function_break(Module *modp, BeamInstr *pc,
 	}
     } else if (break_op == (BeamInstr) BeamOp(op_i_count_breakpoint)) {
 	BpDataCount *bdc = (BpDataCount *) bd;
-
-	bdc->count = 0;
+	erts_smp_atomic_init(&bdc->acount, 0);
     }
     ++(*(BeamInstr*)&code_base[MI_NUM_BREAKPOINTS]);
     return 1;
diff --git a/erts/emulator/beam/beam_bp.h b/erts/emulator/beam/beam_bp.h
index b76e0c01e7..d22dd5de87 100644
--- a/erts/emulator/beam/beam_bp.h
+++ b/erts/emulator/beam/beam_bp.h
@@ -93,7 +93,7 @@ typedef struct bp_data_count {    /* Call count */
     struct bp_data *prev;
     BeamInstr       orig_instr;
     BeamInstr       this_instr;   /* key */
-    Sint            count;
+    erts_smp_atomic_t acount;
 } BpDataCount;
 
 typedef struct {
@@ -152,11 +152,12 @@ ERTS_INLINE Uint bp_sched2ix(void);
 #define bp_sched2ix_proc(p) (0)
 #endif
 
-#define ErtsCountBreak(pc,instr_result)                          \
+#define ErtsCountBreak(p, pc,instr_result)                       \
 do {                                                             \
     BpData **bds = (BpData **) (pc)[-4];                         \
     BpDataCount *bdc = NULL;                                     \
-    Uint ix = bp_sched2ix();                                     \
+    Uint ix = bp_sched2ix_proc( (p) );                           \
+    long count = 0;                                              \
                                                                  \
     ASSERT((pc)[-5] == (BeamInstr) BeamOp(op_i_func_info_IaaI)); \
     ASSERT(bds);                                                 \
@@ -164,17 +165,16 @@ do {                                                             \
     bdc = (BpDataCount *) bdc->next;                             \
     ASSERT(bdc);                                                 \
     bds[ix] = (BpData *) bdc;                                    \
-    ErtsSmpBPLock(bdc);                                          \
-    if (bdc->count >= 0) bdc->count++;                           \
-    ErtsSmpBPUnlock(bdc);                                        \
+    count = erts_smp_atomic_read(&bdc->acount);                  \
+    if (count >= 0)  erts_smp_atomic_inc(&bdc->acount);          \
     *(instr_result) = bdc->orig_instr;                           \
 } while (0)
 
-#define ErtsBreakSkip(pc,instr_result)                           \
+#define ErtsBreakSkip(p, pc,instr_result)                        \
 do {                                                             \
     BpData **bds = (BpData **) (pc)[-4];                         \
     BpData *bd = NULL;                                           \
-    Uint ix = bp_sched2ix();                                     \
+    Uint ix = bp_sched2ix_proc( (p) );                           \
                                                                  \
     ASSERT((pc)[-5] == (BeamInstr) BeamOp(op_i_func_info_IaaI)); \
     ASSERT(bds);                                                 \
diff --git a/erts/emulator/beam/beam_emu.c b/erts/emulator/beam/beam_emu.c
index 6313319b9b..d543006b62 100644
--- a/erts/emulator/beam/beam_emu.c
+++ b/erts/emulator/beam/beam_emu.c
@@ -4404,7 +4404,7 @@ apply_bif_or_nif_epilogue:
  OpCase(i_count_breakpoint): {
      BeamInstr real_I;
      
-     ErtsCountBreak((BeamInstr *) I, &real_I);
+     ErtsCountBreak(c_p, (BeamInstr *) I, &real_I);
      ASSERT(VALID_INSTR(real_I));
      Goto(real_I);
  }
@@ -4485,7 +4485,7 @@ apply_bif_or_nif_epilogue:
      if (! IS_TRACED_FL(c_p, F_TRACE_CALLS)) {
 	 BeamInstr real_I;
 	 
-	 ErtsBreakSkip((BeamInstr *) I, &real_I);
+	 ErtsBreakSkip(c_p, (BeamInstr *) I, &real_I);
 	 Goto(real_I);
      }
  /* Fall through to next case */