aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--erts/emulator/beam/atom.names1
-rw-r--r--erts/emulator/beam/dist.c5
-rw-r--r--erts/emulator/beam/erl_alloc.types1
-rw-r--r--erts/emulator/beam/erl_bif_info.c23
-rw-r--r--erts/emulator/beam/erl_port.h3
-rw-r--r--erts/emulator/beam/erl_process.c3
-rw-r--r--erts/emulator/beam/erl_process.h5
-rw-r--r--erts/emulator/beam/io.c154
-rw-r--r--erts/preloaded/ebin/erts_internal.beambin5380 -> 5964 bytes
-rw-r--r--erts/preloaded/src/erts_internal.erl19
10 files changed, 177 insertions, 37 deletions
diff --git a/erts/emulator/beam/atom.names b/erts/emulator/beam/atom.names
index 74b42c647e..0cf21b4635 100644
--- a/erts/emulator/beam/atom.names
+++ b/erts/emulator/beam/atom.names
@@ -265,6 +265,7 @@ atom get_seq_token
atom get_tcw
atom getenv
atom gather_gc_info_result
+atom gather_io_bytes
atom gather_sched_wall_time_result
atom getting_linked
atom getting_unlinked
diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index ae46174a14..1354cee267 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -2088,6 +2088,7 @@ erts_dist_command(Port *prt, int reds_limit)
DistEntry *dep = prt->dist_entry;
Uint (*send)(Port *prt, ErtsDistOutputBuf *obuf);
erts_aint32_t sched_flags;
+ ErtsSchedulerData *esdp = erts_get_scheduler_data();
ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
@@ -2142,12 +2143,12 @@ erts_dist_command(Port *prt, int reds_limit)
ErtsDistOutputBuf *fob;
size = (*send)(prt, foq.first);
+ esdp->io.out += (Uint64) size;
#ifdef ERTS_RAW_DIST_MSG_DBG
erts_fprintf(stderr, ">> ");
bw(foq.first->extp, size);
#endif
reds += ERTS_PORT_REDS_DIST_CMD_DATA(size);
- erts_smp_atomic_add_nob(&erts_bytes_out, size);
fob = foq.first;
obufsize += size_obuf(fob);
foq.first = foq.first->next;
@@ -2227,12 +2228,12 @@ erts_dist_command(Port *prt, int reds_limit)
ASSERT(&oq.first->data[0] <= oq.first->extp
&& oq.first->extp < oq.first->ext_endp);
size = (*send)(prt, oq.first);
+ esdp->io.out += (Uint64) size;
#ifdef ERTS_RAW_DIST_MSG_DBG
erts_fprintf(stderr, ">> ");
bw(oq.first->extp, size);
#endif
reds += ERTS_PORT_REDS_DIST_CMD_DATA(size);
- erts_smp_atomic_add_nob(&erts_bytes_out, size);
fob = oq.first;
obufsize += size_obuf(fob);
oq.first = oq.first->next;
diff --git a/erts/emulator/beam/erl_alloc.types b/erts/emulator/beam/erl_alloc.types
index 2721e13250..92c397ffd3 100644
--- a/erts/emulator/beam/erl_alloc.types
+++ b/erts/emulator/beam/erl_alloc.types
@@ -274,6 +274,7 @@ type BUSY_CALLER SHORT_LIVED SYSTEM busy_caller
type PROC_SYS_TSK SHORT_LIVED PROCESSES proc_sys_task
type PROC_SYS_TSK_QS SHORT_LIVED PROCESSES proc_sys_task_queues
type NEW_TIME_OFFSET SHORT_LIVED SYSTEM new_time_offset
+type IOB_REQ SHORT_LIVED SYSTEM io_bytes_request
+if threads_no_smp
# Need thread safe allocs, but std_alloc and fix_alloc are not;
diff --git a/erts/emulator/beam/erl_bif_info.c b/erts/emulator/beam/erl_bif_info.c
index 8a675302b2..7eb31fb80e 100644
--- a/erts/emulator/beam/erl_bif_info.c
+++ b/erts/emulator/beam/erl_bif_info.c
@@ -60,6 +60,7 @@
static Export* alloc_info_trap = NULL;
static Export* alloc_sizes_trap = NULL;
+static Export* gather_io_bytes_trap = NULL;
static Export *gather_sched_wall_time_res_trap;
static Export *gather_gc_info_res_trap;
@@ -3220,7 +3221,6 @@ BIF_RETTYPE process_display_2(BIF_ALIST_2)
BIF_RET(am_true);
}
-
/* this is a general call which return some possibly useful information */
BIF_RETTYPE statistics_1(BIF_ALIST_1)
@@ -3293,23 +3293,8 @@ BIF_RETTYPE statistics_1(BIF_ALIST_1)
res = TUPLE2(hp, b1, b2);
BIF_RET(res);
} else if (BIF_ARG_1 == am_io) {
- Eterm r1, r2;
- Eterm in, out;
- Uint hsz = 9;
- Uint bytes_in = (Uint) erts_smp_atomic_read_nob(&erts_bytes_in);
- Uint bytes_out = (Uint) erts_smp_atomic_read_nob(&erts_bytes_out);
-
- (void) erts_bld_uint(NULL, &hsz, bytes_in);
- (void) erts_bld_uint(NULL, &hsz, bytes_out);
- hp = HAlloc(BIF_P, hsz);
- in = erts_bld_uint(&hp, NULL, bytes_in);
- out = erts_bld_uint(&hp, NULL, bytes_out);
-
- r1 = TUPLE2(hp, am_input, in);
- hp += 3;
- r2 = TUPLE2(hp, am_output, out);
- hp += 3;
- BIF_RET(TUPLE2(hp, r1, r2));
+ Eterm ref = erts_request_io_bytes(BIF_P);
+ BIF_TRAP2(gather_io_bytes_trap, BIF_P, ref, make_small(erts_no_schedulers));
}
else if (ERTS_IS_ATOM_STR("run_queues", BIF_ARG_1)) {
Eterm res, *hp, **hpp;
@@ -4370,6 +4355,8 @@ erts_bif_info_init(void)
= erts_export_put(am_erlang, am_gather_sched_wall_time_result, 1);
gather_gc_info_res_trap
= erts_export_put(am_erlang, am_gather_gc_info_result, 1);
+ gather_io_bytes_trap
+ = erts_export_put(am_erts_internal, am_gather_io_bytes, 2);
process_info_init();
os_info_init();
}
diff --git a/erts/emulator/beam/erl_port.h b/erts/emulator/beam/erl_port.h
index 3920fae2d9..66cbb67a0d 100644
--- a/erts/emulator/beam/erl_port.h
+++ b/erts/emulator/beam/erl_port.h
@@ -266,8 +266,7 @@ erts_prtsd_set(Port *prt, int ix, void *data)
#endif
-extern erts_smp_atomic_t erts_bytes_out; /* no bytes written out */
-extern erts_smp_atomic_t erts_bytes_in; /* no bytes sent into the system */
+Eterm erts_request_io_bytes(Process *c_p);
/* port status flags */
diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c
index 43e84a1be1..88c1b5c121 100644
--- a/erts/emulator/beam/erl_process.c
+++ b/erts/emulator/beam/erl_process.c
@@ -5542,6 +5542,9 @@ init_scheduler_data(ErtsSchedulerData* esdp, int num,
esdp->thr_id = (Uint32) num;
erts_sched_bif_unique_init(esdp);
+ esdp->io.out = (Uint64) 0;
+ esdp->io.in = (Uint64) 0;
+
if (daww_ptr) {
init_aux_work_data(&esdp->aux_work_data, esdp, *daww_ptr);
#ifdef ERTS_SMP
diff --git a/erts/emulator/beam/erl_process.h b/erts/emulator/beam/erl_process.h
index 1da1239609..27b6fcef50 100644
--- a/erts/emulator/beam/erl_process.h
+++ b/erts/emulator/beam/erl_process.h
@@ -659,6 +659,11 @@ struct ErtsSchedulerData_ {
ErtsSchedAllocData alloc_data;
+ struct {
+ Uint64 out;
+ Uint64 in;
+ } io;
+
Uint64 reductions;
ErtsSchedWallTime sched_wall_time;
ErtsGCInfo gc_info;
diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c
index 42f0bc88f2..75d80b1a58 100644
--- a/erts/emulator/beam/io.c
+++ b/erts/emulator/beam/io.c
@@ -65,8 +65,6 @@ static erts_smp_tsd_key_t driver_list_last_error_key; /* Save last DDLL error o
per thread basis (for BC interfaces) */
ErtsPTab erts_port erts_align_attribute(ERTS_CACHE_LINE_SIZE); /* The port table */
-erts_smp_atomic_t erts_bytes_out; /* No bytes sent out of the system */
-erts_smp_atomic_t erts_bytes_in; /* No bytes gotten into the system */
const ErlDrvTermData driver_term_nil = (ErlDrvTermData)NIL;
@@ -80,6 +78,9 @@ int erts_port_synchronous_ops = 0;
int erts_port_schedule_all_ops = 0;
int erts_port_parallelism = 0;
+static erts_atomic64_t bytes_in;
+static erts_atomic64_t bytes_out;
+
static void deliver_result(Eterm sender, Eterm pid, Eterm res);
static int init_driver(erts_driver_t *, ErlDrvEntry *, DE_Handle *);
static void terminate_port(Port *p);
@@ -1675,6 +1676,7 @@ call_driver_outputv(int bang_op,
if (bang_op && from != ERTS_PORT_GET_CONNECTED(prt))
send_badsig(prt);
else {
+ ErtsSchedulerData *esdp = erts_get_scheduler_data();
ErlDrvSizeT size = evp->size;
ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)
@@ -1692,7 +1694,10 @@ call_driver_outputv(int bang_op,
prt->caller = NIL;
prt->bytes_out += size;
- erts_smp_atomic_add_nob(&erts_bytes_out, size);
+ if (esdp)
+ esdp->io.out += (Uint64) size;
+ else
+ erts_atomic64_add_nob(&bytes_out, (erts_aint64_t) size);
}
}
@@ -1772,7 +1777,7 @@ call_driver_output(int bang_op,
if (bang_op && from != ERTS_PORT_GET_CONNECTED(prt))
send_badsig(prt);
else {
-
+ ErtsSchedulerData *esdp = erts_get_scheduler_data();
ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)
|| ERTS_IS_CRASH_DUMPING);
@@ -1788,7 +1793,10 @@ call_driver_output(int bang_op,
prt->caller = NIL;
prt->bytes_out += size;
- erts_smp_atomic_add_nob(&erts_bytes_out, size);
+ if (esdp)
+ esdp->io.out += (Uint64) size;
+ else
+ erts_atomic64_add_nob(&bytes_out, (erts_aint64_t) size);
}
}
@@ -2735,6 +2743,9 @@ void erts_init_io(int port_tab_size,
drv_list_rwmtx_opts.type = ERTS_SMP_RWMTX_TYPE_EXTREMELY_FREQUENT_READ;
drv_list_rwmtx_opts.lived = ERTS_SMP_RWMTX_LONG_LIVED;
+ erts_atomic64_init_nob(&bytes_in, 0);
+ erts_atomic64_init_nob(&bytes_out, 0);
+
common_element_size = ERTS_ALC_DATA_ALIGN_SIZE(sizeof(Port));
common_element_size += ERTS_ALC_DATA_ALIGN_SIZE(sizeof(ErtsPortTaskBusyPortQ));
common_element_size += 10; /* name */
@@ -2776,9 +2787,6 @@ void erts_init_io(int port_tab_size,
legacy_port_tab,
1);
- erts_smp_atomic_init_nob(&erts_bytes_out, 0);
- erts_smp_atomic_init_nob(&erts_bytes_in, 0);
-
sys_init_io();
erts_smp_tsd_set(driver_list_lock_status_key, (void *) 1);
@@ -4568,6 +4576,102 @@ erts_port_info(Process* c_p,
}
typedef struct {
+ Uint sched_id;
+ Eterm pid;
+ Uint32 refn[ERTS_REF_NUMBERS];
+ erts_smp_atomic32_t refc;
+} ErtsIOBytesReq;
+
+static void
+reply_io_bytes(void *vreq)
+{
+ ErtsIOBytesReq *req = (ErtsIOBytesReq *) vreq;
+ Process *rp;
+
+ rp = erts_proc_lookup(req->pid);
+ if (rp) {
+ ErlOffHeap *ohp = NULL;
+ ErlHeapFragment *bp = NULL;
+ ErtsProcLocks rp_locks;
+ Eterm ref, msg, ein, eout, *hp;
+ Uint64 in, out;
+ Uint hsz;
+ ErtsSchedulerData *esdp = erts_get_scheduler_data();
+ Uint sched_id = esdp->no;
+ in = esdp->io.in;
+ out = esdp->io.out;
+ if (req->sched_id != sched_id)
+ rp_locks = 0;
+ else {
+ in += (Uint64) erts_atomic64_read_nob(&bytes_in);
+ out += (Uint64) erts_atomic64_read_nob(&bytes_out);
+ rp_locks = ERTS_PROC_LOCK_MAIN;
+ }
+
+ hsz = 5 /* 4-tuple */ + REF_THING_SIZE;
+
+ erts_bld_uint64(NULL, &hsz, in);
+ erts_bld_uint64(NULL, &hsz, out);
+
+ hp = erts_alloc_message_heap(hsz, &bp, &ohp, rp, &rp_locks);
+
+ ref = make_internal_ref(hp);
+ write_ref_thing(hp, req->refn[0], req->refn[1], req->refn[2]);
+ hp += REF_THING_SIZE;
+
+ ein = erts_bld_uint64(&hp, NULL, in);
+ eout = erts_bld_uint64(&hp, NULL, out);
+
+ msg = TUPLE4(hp, ref, make_small(sched_id), ein, eout);
+ erts_queue_message(rp, &rp_locks, bp, msg, NIL);
+
+ if (req->sched_id == sched_id)
+ rp_locks &= ~ERTS_PROC_LOCK_MAIN;
+ if (rp_locks)
+ erts_smp_proc_unlock(rp, rp_locks);
+ }
+
+ if (erts_smp_atomic32_dec_read_nob(&req->refc) == 0)
+ erts_free(ERTS_ALC_T_IOB_REQ, req);
+}
+
+Eterm
+erts_request_io_bytes(Process *c_p)
+{
+ Uint *hp;
+ Eterm ref;
+ Uint32 *refn;
+ ErtsSchedulerData *esdp = ERTS_PROC_GET_SCHDATA(c_p);
+ ErtsIOBytesReq *req = erts_alloc(ERTS_ALC_T_IOB_REQ,
+ sizeof(ErtsIOBytesReq));
+
+ hp = HAlloc(c_p, REF_THING_SIZE);
+ ref = erts_sched_make_ref_in_buffer(esdp, hp);
+ refn = internal_ref_numbers(ref);
+
+ req->sched_id = esdp->no;
+ req->pid = c_p->common.id;
+ req->refn[0] = refn[0];
+ req->refn[1] = refn[1];
+ req->refn[2] = refn[2];
+ erts_smp_atomic32_init_nob(&req->refc,
+ (erts_aint32_t) erts_no_schedulers);
+
+#ifdef ERTS_SMP
+ if (erts_no_schedulers > 1)
+ erts_schedule_multi_misc_aux_work(1,
+ erts_no_schedulers,
+ reply_io_bytes,
+ (void *) req);
+#endif
+
+ reply_io_bytes((void *) req);
+
+ return ref;
+}
+
+
+typedef struct {
int to;
void *arg;
} prt_one_lnk_data;
@@ -5105,6 +5209,7 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len)
struct b2t_states__ b2t;
int scheduler;
int is_heap_need_limited = 1;
+ ErtsSchedulerData *esdp = erts_get_scheduler_data();
ERTS_UNDEF(mess,NIL);
ERTS_UNDEF(scheduler,1);
@@ -5412,7 +5517,10 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len)
Uint size = ptr[1];
Uint offset = ptr[2];
- erts_smp_atomic_add_nob(&erts_bytes_in, (erts_aint_t) size);
+ if (esdp)
+ esdp->io.in += (Uint64) size;
+ else
+ erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) size);
if (size <= ERL_ONHEAP_BIN_LIMIT) {
ErlHeapBin* hbp = (ErlHeapBin *) erts_produce_heap(&factory,
@@ -5446,7 +5554,10 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len)
byte *bufp = (byte *) ptr[0];
Uint size = (Uint) ptr[1];
- erts_smp_atomic_add_nob(&erts_bytes_in, (erts_aint_t) size);
+ if (esdp)
+ esdp->io.in += (Uint64) size;
+ else
+ erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) size);
if (size <= ERL_ONHEAP_BIN_LIMIT) {
ErlHeapBin* hbp = (ErlHeapBin *) erts_produce_heap(&factory,
@@ -5483,7 +5594,10 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len)
}
case ERL_DRV_STRING: /* char*, length */
- erts_smp_atomic_add_nob(&erts_bytes_in, (erts_aint_t) ptr[1]);
+ if (esdp)
+ esdp->io.in += (Uint64) ptr[1];
+ else
+ erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) ptr[1]);
erts_reserve_heap(&factory, 2*ptr[1]);
mess = buf_to_intlist(&factory.hp, (char*)ptr[0], ptr[1], NIL);
ptr += 2;
@@ -5759,6 +5873,7 @@ int driver_output_binary(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen,
{
erts_aint32_t state;
Port* prt = erts_drvport2port_state(ix, &state);
+ ErtsSchedulerData *esdp = erts_get_scheduler_data();
ERTS_SMP_CHK_NO_PROC_LOCKS;
@@ -5769,7 +5884,10 @@ int driver_output_binary(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen,
return 0;
prt->bytes_in += (hlen + len);
- erts_smp_atomic_add_nob(&erts_bytes_in, (erts_aint_t) (hlen + len));
+ if (esdp)
+ esdp->io.in += (Uint64) (hlen + len);
+ else
+ erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) (hlen + len));
if (state & ERTS_PORT_SFLG_DISTRIBUTION) {
return erts_net_message(prt,
prt->dist_entry,
@@ -5794,6 +5912,7 @@ int driver_output2(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen,
{
erts_aint32_t state;
Port* prt = erts_drvport2port_state(ix, &state);
+ ErtsSchedulerData *esdp = erts_get_scheduler_data();
ERTS_SMP_CHK_NO_PROC_LOCKS;
@@ -5805,7 +5924,10 @@ int driver_output2(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen,
return 0;
prt->bytes_in += (hlen + len);
- erts_smp_atomic_add_nob(&erts_bytes_in, (erts_aint_t) (hlen + len));
+ if (esdp)
+ esdp->io.in += (Uint64) (hlen + len);
+ else
+ erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) (hlen + len));
if (state & ERTS_PORT_SFLG_DISTRIBUTION) {
if (len == 0)
return erts_net_message(prt,
@@ -5845,6 +5967,7 @@ int driver_outputv(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen,
ErlDrvBinary** binv;
Port* prt;
erts_aint32_t state;
+ ErtsSchedulerData *esdp = erts_get_scheduler_data();
ERTS_SMP_CHK_NO_PROC_LOCKS;
@@ -5883,7 +6006,10 @@ int driver_outputv(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen,
/* XXX handle distribution !!! */
prt->bytes_in += (hlen + size);
- erts_smp_atomic_add_nob(&erts_bytes_in, (erts_aint_t) (hlen + size));
+ if (esdp)
+ esdp->io.in += (Uint64) (hlen + size);
+ else
+ erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) (hlen + size));
deliver_vec_message(prt, ERTS_PORT_GET_CONNECTED(prt), hbuf, hlen,
binv, iov, n, size);
return 0;
diff --git a/erts/preloaded/ebin/erts_internal.beam b/erts/preloaded/ebin/erts_internal.beam
index 0e0811af3f..dc8c711e1a 100644
--- a/erts/preloaded/ebin/erts_internal.beam
+++ b/erts/preloaded/ebin/erts_internal.beam
Binary files differ
diff --git a/erts/preloaded/src/erts_internal.erl b/erts/preloaded/src/erts_internal.erl
index 65a1f1ed3a..cf8edefd7d 100644
--- a/erts/preloaded/src/erts_internal.erl
+++ b/erts/preloaded/src/erts_internal.erl
@@ -40,7 +40,7 @@
-export([flush_monitor_messages/3]).
--export([await_result/1]).
+-export([await_result/1, gather_io_bytes/2]).
-export([time_unit/0]).
@@ -67,6 +67,23 @@ await_result(Ref) when is_reference(Ref) ->
end.
%%
+%% statistics(io) end up in gather_io_bytes/2
+%%
+
+gather_io_bytes(Ref, No) when is_reference(Ref),
+ is_integer(No),
+ No > 0 ->
+ gather_io_bytes(Ref, No, 0, 0).
+
+gather_io_bytes(_Ref, 0, InAcc, OutAcc) ->
+ {{input, InAcc}, {output, OutAcc}};
+gather_io_bytes(Ref, No, InAcc, OutAcc) ->
+ receive
+ {Ref, _SchedId, In, Out} ->
+ gather_io_bytes(Ref, No-1, InAcc + In, OutAcc + Out)
+ end.
+
+%%
%% Statically linked port NIFs
%%
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-20 12:08:50 +0000