diff options
| -rw-r--r-- | bootstrap/lib/stdlib/ebin/ets.beam | bin | 22544 -> 22808 bytes | |||
| -rw-r--r-- | erts/emulator/beam/bif.tab | 2 | ||||
| -rw-r--r-- | erts/emulator/beam/break.c | 2 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_alloc.types | 1 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_db.c | 559 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_db.h | 42 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_db_util.h | 9 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_init.c | 1 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_process.c | 50 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_process.h | 16 | ||||
| -rw-r--r-- | lib/stdlib/src/ets.erl | 18 | ||||
| -rw-r--r-- | lib/stdlib/test/ets_SUITE.erl | 64 |
12 files changed, 672 insertions, 92 deletions
diff --git a/bootstrap/lib/stdlib/ebin/ets.beam b/bootstrap/lib/stdlib/ebin/ets.beam Binary files differindex e66749048d..ee769b5dec 100644 --- a/bootstrap/lib/stdlib/ebin/ets.beam +++ b/bootstrap/lib/stdlib/ebin/ets.beam diff --git a/erts/emulator/beam/bif.tab b/erts/emulator/beam/bif.tab index a6510ace81..08c8c3adbd 100644 --- a/erts/emulator/beam/bif.tab +++ b/erts/emulator/beam/bif.tab @@ -324,7 +324,7 @@ bif erlang:match_spec_test/3 # Bifs in ets module. # -bif ets:all/0 +bif ets:internal_request_all/0 bif ets:new/2 bif ets:delete/1 bif ets:delete/2 diff --git a/erts/emulator/beam/break.c b/erts/emulator/beam/break.c index 4dad5736c5..0b40d70cb7 100644 --- a/erts/emulator/beam/break.c +++ b/erts/emulator/beam/break.c @@ -512,6 +512,8 @@ do_break(void) erts_free_read_env(mode); #endif /* __WIN32__ */ + ASSERT(erts_smp_thr_progress_is_blocking()); + erts_printf("\n" "BREAK: (a)bort (c)ontinue (p)roc info (i)nfo (l)oaded\n" " (v)ersion (k)ill (D)b-tables (d)istribution\n"); diff --git a/erts/emulator/beam/erl_alloc.types b/erts/emulator/beam/erl_alloc.types index 3d5de72ee7..32f84c8593 100644 --- a/erts/emulator/beam/erl_alloc.types +++ b/erts/emulator/beam/erl_alloc.types @@ -227,6 +227,7 @@ type DB_DMC_ERROR ETS ETS db_dmc_error type DB_DMC_ERR_INFO ETS ETS db_dmc_error_info type DB_TERM ETS ETS db_term type DB_PROC_CLEANUP SHORT_LIVED ETS db_proc_cleanup_state +type ETS_ALL_REQ SHORT_LIVED ETS ets_all_request type INSTR_INFO LONG_LIVED SYSTEM instr_info type LOGGER_DSBUF TEMPORARY SYSTEM logger_dsbuf type TMP_DSBUF TEMPORARY SYSTEM tmp_dsbuf diff --git a/erts/emulator/beam/erl_db.c b/erts/emulator/beam/erl_db.c index ad11d72d5f..8c4c76e187 100644 --- a/erts/emulator/beam/erl_db.c +++ b/erts/emulator/beam/erl_db.c @@ -114,7 +114,12 @@ static struct { #define MARK_SLOT_DEAD(i) (meta_main_tab[(i)].u.next_free |= 2) #define GET_ANY_SLOT_TAB(i) ((DbTable*)(meta_main_tab[(i)].u.next_free & ~(1|2))) /* dead or alive */ +static void +send_ets_transfer_message(Process *c_p, Process *proc, + ErtsProcLocks *locks, + DbTable *tb, Eterm heir_data); static void schedule_free_dbtable(DbTable* tb); +static void delete_sched_table(Process *c_p, DbTable *tb); static void table_dec_refc(DbTable *tb, erts_aint_t min_val) { @@ -164,16 +169,32 @@ tid2tab(Eterm tid) return tb; } +static ERTS_INLINE int +is_table_alive(DbTable *tb) +{ + erts_smp_atomic_t *tbref; + DbTable *rtb; + + tbref = erts_smp_binary_to_magic_indirection(tb->common.btid); + rtb = (DbTable *) erts_smp_atomic_read_nob(tbref); + + ASSERT(!rtb || rtb == tb); + + return !!rtb; +} + static ERTS_INLINE void -tid_clear(DbTable *tb) +tid_clear(Process *c_p, DbTable *tb) { DbTable *rtb; Binary *btid = tb->common.btid; erts_smp_atomic_t *tbref = erts_smp_binary_to_magic_indirection(btid); rtb = (DbTable *) erts_smp_atomic_xchg_nob(tbref, (erts_aint_t) NULL); ASSERT(!rtb || tb == rtb); - if (rtb) + if (rtb) { table_dec_refc(tb, 1); + delete_sched_table(c_p, tb); + } } static ERTS_INLINE Eterm @@ -348,6 +369,99 @@ static void schedule_free_dbtable(DbTable* tb) sizeof(DbTable)); } +static ERTS_INLINE void +save_sched_table(Process *c_p, DbTable *tb) +{ + ErtsSchedulerData *esdp = erts_proc_sched_data(c_p); + DbTable *first; + + ASSERT(esdp); + esdp->ets_tables.count++; + erts_smp_refc_inc(&tb->common.refc, 1); + + first = esdp->ets_tables.clist; + if (!first) { + tb->common.all.next = tb->common.all.prev = tb; + esdp->ets_tables.clist = tb; + } + else { + tb->common.all.prev = first->common.all.prev; + tb->common.all.next = first; + tb->common.all.prev->common.all.next = tb; + first->common.all.prev = tb; + } +} + +static ERTS_INLINE void +remove_sched_table(ErtsSchedulerData *esdp, DbTable *tb) +{ + ErtsEtsAllYieldData *eaydp; + ASSERT(esdp); + ASSERT(erts_get_ref_numbers_thr_id(ERTS_MAGIC_BIN_REFN(tb->common.btid)) + == (Uint32) esdp->no); + + ASSERT(esdp->ets_tables.count > 0); + esdp->ets_tables.count--; + + eaydp = ERTS_SCHED_AUX_YIELD_DATA(esdp, ets_all); + if (eaydp->ongoing) { + /* ets:all() op process list from last to first... */ + if (eaydp->tab == tb) { + if (eaydp->tab == esdp->ets_tables.clist) + eaydp->tab = NULL; + else + eaydp->tab = tb->common.all.prev; + } + } + + if (tb->common.all.next == tb) { + ASSERT(tb->common.all.prev == tb); + ASSERT(esdp->ets_tables.clist == tb); + esdp->ets_tables.clist = NULL; + } + else { +#ifdef DEBUG + DbTable *tmp = esdp->ets_tables.clist; + do { + if (tmp == tb) break; + tmp = tmp->common.all.next; + } while (tmp != esdp->ets_tables.clist); + ASSERT(tmp == tb); +#endif + tb->common.all.prev->common.all.next = tb->common.all.next; + tb->common.all.next->common.all.prev = tb->common.all.prev; + + if (esdp->ets_tables.clist == tb) + esdp->ets_tables.clist = tb->common.all.next; + + } + + table_dec_refc(tb, 0); +} + +static void +scheduled_remove_sched_table(void *vtb) +{ + remove_sched_table(erts_get_scheduler_data(), (DbTable *) vtb); +} + +static void +delete_sched_table(Process *c_p, DbTable *tb) +{ + ErtsSchedulerData *esdp = erts_proc_sched_data(c_p); + Uint32 sid; + + ASSERT(esdp); + + ASSERT(tb->common.btid); + sid = erts_get_ref_numbers_thr_id(ERTS_MAGIC_BIN_REFN(tb->common.btid)); + ASSERT(1 <= sid && sid <= erts_no_schedulers); + if (sid == (Uint32) esdp->no) + remove_sched_table(esdp, tb); + else + erts_schedule_misc_aux_work((int) sid, scheduled_remove_sched_table, tb); +} + static ERTS_INLINE void db_init_lock(DbTable* tb, int use_frequent_read_lock, char *rwname, char* fixname) { @@ -1322,6 +1436,7 @@ BIF_RETTYPE ets_rename_2(BIF_ALIST_2) { DbTable* tb; Eterm ret; + Eterm old_name; erts_smp_rwmtx_t *lck1, *lck2; #ifdef HARDDEBUG @@ -1338,12 +1453,10 @@ BIF_RETTYPE ets_rename_2(BIF_ALIST_2) (void) meta_name_tab_bucket(BIF_ARG_2, &lck1); - if (is_small(BIF_ARG_1)) { - Uint slot = unsigned_val(BIF_ARG_1) & meta_main_tab_slot_mask; - lck2 = get_meta_main_tab_lock(slot); - } - else if (is_atom(BIF_ARG_1)) { - (void) meta_name_tab_bucket(BIF_ARG_1, &lck2); + if (is_atom(BIF_ARG_1)) { + old_name = BIF_ARG_1; + named_tab: + (void) meta_name_tab_bucket(old_name, &lck2); if (lck1 == lck2) lck2 = NULL; else if (lck1 > lck2) { @@ -1353,7 +1466,18 @@ BIF_RETTYPE ets_rename_2(BIF_ALIST_2) } } else { - BIF_ERROR(BIF_P, BADARG); + Uint slot; + tb = tid2tab(BIF_ARG_1); + if (!tb) + BIF_ERROR(BIF_P, BADARG); + else { + if (is_atom(tb->common.id)) { + old_name = tb->common.id; + goto named_tab; + } + slot = unsigned_val(tb->common.id) & meta_main_tab_slot_mask; + lck2 = get_meta_main_tab_lock(slot); + } } erts_smp_rwmtx_rwlock(lck1); @@ -1378,7 +1502,10 @@ BIF_RETTYPE ets_rename_2(BIF_ALIST_2) tb->common.id = tb->common.the_name = BIF_ARG_2; done: - ret = tb->common.id; + if (is_atom(tb->common.id)) + ret = tb->common.id; + else + ret = BIF_ARG_1; db_unlock(tb, LCK_WRITE); erts_smp_rwmtx_rwunlock(lck1); if (lck2) @@ -1622,6 +1749,8 @@ BIF_RETTYPE ets_new_2(BIF_ALIST_2) if (!is_named) ret = make_tid(BIF_P, tb); + save_sched_table(BIF_P, tb); + mmtl = get_meta_main_tab_lock(slot); erts_smp_rwmtx_rwlock(mmtl); meta_main_tab[slot].u.tb = tb; @@ -1634,7 +1763,7 @@ BIF_RETTYPE ets_new_2(BIF_ALIST_2) free_slot(slot); erts_smp_rwmtx_rwunlock(mmtl); - tid_clear(tb); + tid_clear(BIF_P, tb); db_lock(tb,LCK_WRITE); free_heir_data(tb); @@ -1825,7 +1954,7 @@ BIF_RETTYPE ets_delete_1(BIF_ALIST_1) UnUseTmpHeap(3,BIF_P); } - tid_clear(tb); + tid_clear(BIF_P, tb); mmtl = get_meta_main_tab_lock(tb->common.slot); #ifdef ERTS_SMP @@ -1913,12 +2042,8 @@ BIF_RETTYPE ets_give_away_3(BIF_ALIST_3) db_meta_unlock(meta_pid_to_tab, LCK_WRITE_REC); db_unlock(tb,LCK_WRITE); - erts_send_message(BIF_P, to_proc, &to_locks, - TUPLE4(buf, am_ETS_TRANSFER, - tb->common.id, - from_pid, - BIF_ARG_3), - 0); + send_ets_transfer_message(BIF_P, to_proc, &to_locks, + tb, BIF_ARG_3); erts_smp_proc_unlock(to_proc, to_locks); UnUseTmpHeap(5,BIF_P); BIF_RET(am_true); @@ -2188,46 +2313,254 @@ BIF_RETTYPE ets_select_delete_2(BIF_ALIST_2) return result; } -/* -** Return a list of tables on this node -*/ -BIF_RETTYPE ets_all_0(BIF_ALIST_0) +/* + * ets:all/0 + * + * ets:all() calls ets:internal_request_all/0 which + * requests information about all tables from + * each scheduler thread. Each scheduler replies + * to the calling process with information about + * existing tables created on that specific scheduler. + */ + +struct ErtsEtsAllReq_ { + erts_smp_atomic32_t refc; + Process *proc; + ErtsOIRefStorage ref; + ErtsEtsAllReqList list[1]; /* one per scheduler */ +}; + +#define ERTS_ETS_ALL_REQ_SIZE \ + (sizeof(ErtsEtsAllReq) \ + + (sizeof(ErtsEtsAllReqList) \ + * (erts_no_schedulers - 1))) + +typedef struct { + ErtsEtsAllReq *ongoing; + ErlHeapFragment *hfrag; + DbTable *tab; + ErtsEtsAllReq *queue; +} ErtsEtsAllData; + +/* Tables handled before yielding */ +#define ERTS_ETS_ALL_TB_YCNT 200 +/* + * Min yield count required before starting + * an operation that will require yield. + */ +#define ERTS_ETS_ALL_TB_YCNT_START 10 + +#ifdef DEBUG +/* Test yielding... */ +#undef ERTS_ETS_ALL_TB_YCNT +#undef ERTS_ETS_ALL_TB_YCNT_START +#define ERTS_ETS_ALL_TB_YCNT 10 +#define ERTS_ETS_ALL_TB_YCNT_START 1 +#endif + +static int +ets_all_reply(ErtsSchedulerData *esdp, ErtsEtsAllReq **reqpp, + ErlHeapFragment **hfragpp, DbTable **tablepp, + int *yield_count_p) { - DbTable* tb; - Eterm previous; - int i; - Eterm* hp; - Eterm* hendp; - int t_tabs_cnt; - int t_top; + ErtsEtsAllReq *reqp = *reqpp; + ErlHeapFragment *hfragp = *hfragpp; + int ycount = *yield_count_p; + DbTable *tb, *first; + Uint sz; + Eterm list, msg, ref, *hp; + ErlOffHeap *ohp; + ErtsMessage *mp; - erts_smp_spin_lock(&meta_main_tab_main_lock); - t_tabs_cnt = meta_main_tab_cnt; - t_top = meta_main_tab_top; - erts_smp_spin_unlock(&meta_main_tab_main_lock); + /* + * - save_sched_table() inserts at end of circular list. + * + * - This function scans from the end so we know that + * the amount of tables to scan wont grow even if we + * yield. + * + * - remove_sched_table() updates the table we yielded + * on if it removes it. + */ + + if (hfragp) { + /* Restart of a yielded operation... */ + ASSERT(hfragp->used_size < hfragp->alloc_size); + ohp = &hfragp->off_heap; + hp = &hfragp->mem[hfragp->used_size]; + list = *hp; + hfragp->used_size = hfragp->alloc_size; + first = esdp->ets_tables.clist; + tb = *tablepp; + } + else { + /* A new operation... */ + ASSERT(!*tablepp); - hp = HAlloc(BIF_P, 2*t_tabs_cnt); - hendp = hp + 2*t_tabs_cnt; + /* Max heap size needed... */ + sz = esdp->ets_tables.count; + sz *= ERTS_MAGIC_REF_THING_SIZE + 2; + sz += 3 + ERTS_REF_THING_SIZE; + hfragp = new_message_buffer(sz); - previous = NIL; - for(i = 0; i < t_top; i++) { - erts_smp_rwmtx_t *mmtl = get_meta_main_tab_lock(i); - erts_smp_rwmtx_rlock(mmtl); - if (IS_SLOT_ALIVE(i)) { - if (hp == hendp) { - /* Racing table creator, grab some more heap space */ - t_tabs_cnt = 10; - hp = HAlloc(BIF_P, 2*t_tabs_cnt); - hendp = hp + 2*t_tabs_cnt; - } - tb = meta_main_tab[i].u.tb; - previous = CONS(hp, tb->common.id, previous); - hp += 2; - } - erts_smp_rwmtx_runlock(mmtl); + hp = &hfragp->mem[0]; + ohp = &hfragp->off_heap; + list = NIL; + first = esdp->ets_tables.clist; + tb = first ? first->common.all.prev : NULL; + } + + if (tb) { + while (1) { + if (is_table_alive(tb)) { + Eterm tid; + if (is_atom(tb->common.id)) + tid = tb->common.id; + else + tid = erts_mk_magic_ref(&hp, ohp, tb->common.btid); + list = CONS(hp, tid, list); + hp += 2; + } + + if (tb == first) + break; + + tb = tb->common.all.prev; + + if (--ycount <= 0) { + sz = hp - &hfragp->mem[0]; + ASSERT(hfragp->alloc_size > sz + 1); + *hp = list; + hfragp->used_size = sz; + *hfragpp = hfragp; + *reqpp = reqp; + *tablepp = tb; + *yield_count_p = 0; + return 1; /* Yield! */ + } + } + } + + ref = erts_oiref_storage_make_ref(&reqp->ref, &hp); + msg = TUPLE2(hp, ref, list); + hp += 3; + + sz = hp - &hfragp->mem[0]; + ASSERT(sz <= hfragp->alloc_size); + + hfragp = erts_resize_message_buffer(hfragp, sz, &msg, 1); + + mp = erts_alloc_message(0, NULL); + mp->data.heap_frag = hfragp; + + erts_queue_message(reqp->proc, 0, mp, msg, am_system); + + erts_proc_dec_refc(reqp->proc); + + if (erts_smp_atomic32_dec_read_nob(&reqp->refc) == 0) + erts_free(ERTS_ALC_T_ETS_ALL_REQ, reqp); + + *reqpp = NULL; + *hfragpp = NULL; + *tablepp = NULL; + *yield_count_p = ycount; + + return 0; +} + +int +erts_handle_yielded_ets_all_request(ErtsSchedulerData *esdp, + ErtsEtsAllYieldData *eaydp) +{ + int ix = (int) esdp->no - 1; + int yc = ERTS_ETS_ALL_TB_YCNT; + + while (1) { + if (!eaydp->ongoing) { + ErtsEtsAllReq *ongoing; + + if (!eaydp->queue) + return 0; /* All work completed! */ + + if (yc < ERTS_ETS_ALL_TB_YCNT_START && yc > esdp->ets_tables.count) + return 1; /* Yield! */ + + eaydp->ongoing = ongoing = eaydp->queue; + if (ongoing->list[ix].next == ongoing) + eaydp->queue = NULL; + else { + ongoing->list[ix].next->list[ix].prev = ongoing->list[ix].prev; + ongoing->list[ix].prev->list[ix].next = ongoing->list[ix].next; + eaydp->queue = ongoing->list[ix].next; + } + ASSERT(!eaydp->hfrag); + ASSERT(!eaydp->tab); + } + + if (ets_all_reply(esdp, &eaydp->ongoing, &eaydp->hfrag, &eaydp->tab, &yc)) + return 1; /* Yield! */ + } +} + +static void +handle_ets_all_request(void *vreq) +{ + ErtsSchedulerData *esdp = erts_get_scheduler_data(); + ErtsEtsAllYieldData *eayp = ERTS_SCHED_AUX_YIELD_DATA(esdp, ets_all); + ErtsEtsAllReq *req = (ErtsEtsAllReq *) vreq; + + if (!eayp->ongoing && !eayp->queue) { + /* No ets:all() operations ongoing... */ + ErlHeapFragment *hf = NULL; + DbTable *tb = NULL; + int yc = ERTS_ETS_ALL_TB_YCNT; + if (ets_all_reply(esdp, &req, &hf, &tb, &yc)) { + /* Yielded... */ + ASSERT(hf); + eayp->ongoing = req; + eayp->hfrag = hf; + eayp->tab = tb; + erts_notify_new_aux_yield_work(esdp); + } + } + else { + /* Ongoing ets:all() operations; queue up this request... */ + int ix = (int) esdp->no - 1; + if (!eayp->queue) { + req->list[ix].next = req; + req->list[ix].prev = req; + eayp->queue = req; + } + else { + req->list[ix].next = eayp->queue; + req->list[ix].prev = eayp->queue->list[ix].prev; + eayp->queue->list[ix].prev = req; + req->list[ix].prev->list[ix].next = req; + } } - HRelease(BIF_P, hendp, hp); - BIF_RET(previous); +} + +BIF_RETTYPE ets_internal_request_all_0(BIF_ALIST_0) +{ + Eterm ref = erts_make_ref(BIF_P); + ErtsEtsAllReq *req = erts_alloc(ERTS_ALC_T_ETS_ALL_REQ, + ERTS_ETS_ALL_REQ_SIZE); + erts_smp_atomic32_init_nob(&req->refc, + (erts_aint32_t) erts_no_schedulers); + erts_oiref_storage_save(&req->ref, ref); + req->proc = BIF_P; + erts_proc_add_refc(BIF_P, (Sint) erts_no_schedulers); + +#ifdef ERTS_SMP + if (erts_no_schedulers > 1) + erts_schedule_multi_misc_aux_work(1, + erts_no_schedulers, + handle_ets_all_request, + (void *) req); +#endif + + handle_ets_all_request((void *) req); + BIF_RET(ref); } @@ -2788,7 +3121,7 @@ BIF_RETTYPE ets_info_1(BIF_ALIST_1) */ if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_INFO, LCK_READ)) == NULL) { - if (is_atom(BIF_ARG_1) || is_small(BIF_ARG_1)) { + if (is_atom(BIF_ARG_1) || is_ref(BIF_ARG_1)) { BIF_RET(am_undefined); } BIF_ERROR(BIF_P, BADARG); @@ -2850,7 +3183,7 @@ BIF_RETTYPE ets_info_2(BIF_ALIST_2) Eterm ret = THE_NON_VALUE; if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_INFO, LCK_READ)) == NULL) { - if (is_atom(BIF_ARG_1) || is_small(BIF_ARG_1)) { + if (is_atom(BIF_ARG_1) || is_ref(BIF_ARG_1)) { BIF_RET(am_undefined); } BIF_ERROR(BIF_P, BADARG); @@ -3140,6 +3473,19 @@ void init_db(ErtsDbSpinCount db_spin_count) ms_delete_all = CONS(hp, ms_delete_all,NIL); } +void +erts_ets_sched_spec_data_init(ErtsSchedulerData *esdp) +{ + ErtsEtsAllYieldData *eaydp = ERTS_SCHED_AUX_YIELD_DATA(esdp, ets_all); + eaydp->ongoing = NULL; + eaydp->hfrag = NULL; + eaydp->tab = NULL; + eaydp->queue = NULL; + esdp->ets_tables.clist = NULL; + esdp->ets_tables.count = 0; +} + + #define ARRAY_CHUNK 100 typedef enum { @@ -3288,17 +3634,46 @@ retry: ASSERT(arityval(*tpv) == 1); heir_data = tpv[1]; } - erts_send_message(p, to_proc, &to_locks, - TUPLE4(buf, - am_ETS_TRANSFER, - tb->common.id, - p->common.id, - heir_data), - 0); + send_ets_transfer_message(p, to_proc, &to_locks, tb, heir_data); erts_smp_proc_unlock(to_proc, to_locks); return !0; } +static void +send_ets_transfer_message(Process *c_p, Process *proc, + ErtsProcLocks *locks, + DbTable *tb, Eterm heir_data) +{ + Uint hsz, hd_sz; + ErtsMessage *mp; + Eterm *hp; + ErlOffHeap *ohp; + Eterm tid, hd_copy, msg, sender; + + hsz = 5; + if (is_not_atom(tb->common.id)) + hsz += ERTS_MAGIC_REF_THING_SIZE; + if (is_immed(heir_data)) + hd_sz = 0; + else { + hd_sz = size_object(heir_data); + hsz += hd_sz; + } + + mp = erts_alloc_message_heap(proc, locks, hsz, &hp, &ohp); + if (is_atom(tb->common.id)) + tid = tb->common.id; + else + tid = erts_mk_magic_ref(&hp, ohp, tb->common.btid); + if (!hd_sz) + hd_copy = heir_data; + else + hd_copy = copy_struct(heir_data, hd_sz, &hp, ohp); + sender = c_p->common.id; + msg = TUPLE4(hp, am_ETS_TRANSFER, tid, sender, hd_copy); + erts_queue_message(proc, *locks, mp, msg, sender); +} + /* * erts_db_process_exiting() is called when a process terminates. * It returns 0 when completely done, and !0 when it wants to @@ -3384,7 +3759,7 @@ erts_db_process_exiting(Process *c_p, ErtsProcLocks c_p_locks) #endif first_call = (tb->common.status & DB_DELETE) == 0; if (first_call) { - tid_clear(tb); + tid_clear(c_p, tb); /* Clear all access bits. */ tb->common.status &= ~(DB_PROTECTED | DB_PUBLIC @@ -3975,21 +4350,30 @@ static void print_table(fmtfn_t to, void *to_arg, int show, DbTable* tb) erts_print(to, to_arg, "Read Concurrency: %T\n", table_info(NULL, tb, am_read_concurrency)); } +typedef struct { + fmtfn_t to; + void *to_arg; + int show; +} ErtsPrintDbInfo; + +static void +db_info_print(DbTable *tb, void *vpdbip) +{ + ErtsPrintDbInfo *pdbip = (ErtsPrintDbInfo *) vpdbip; + erts_print(pdbip->to, pdbip->to_arg, "=ets:%T\n", tb->common.owner); + erts_print(pdbip->to, pdbip->to_arg, "Slot: %bpu\n", (Uint) tb); + print_table(pdbip->to, pdbip->to_arg, pdbip->show, tb); +} + void db_info(fmtfn_t to, void *to_arg, int show) /* Called by break handler */ { - int i; - for (i=0; i < db_max_tabs; i++) - if (IS_SLOT_ALIVE(i)) { - erts_print(to, to_arg, "=ets:%T\n", meta_main_tab[i].u.tb->common.owner); - erts_print(to, to_arg, "Slot: %d\n", i); - print_table(to, to_arg, show, meta_main_tab[i].u.tb); - } -#ifdef DEBUG - erts_print(to, to_arg, "=internal_ets: Process to table index\n"); - print_table(to, to_arg, show, meta_pid_to_tab); - erts_print(to, to_arg, "=internal_ets: Process to fixation index\n"); - print_table(to, to_arg, show, meta_pid_to_fixed_tab); -#endif + ErtsPrintDbInfo pdbi; + + pdbi.to = to; + pdbi.to_arg = to_arg; + pdbi.show = show; + + erts_db_foreach_table(db_info_print, &pdbi); } Uint @@ -4004,15 +4388,22 @@ erts_get_ets_misc_mem_size(void) void erts_db_foreach_table(void (*func)(DbTable *, void *), void *arg) { - int i, j; - j = 0; - for(i = 0; (i < db_max_tabs && j < meta_main_tab_cnt); i++) { - if (IS_SLOT_ALIVE(i)) { - j++; - (*func)(meta_main_tab[i].u.tb, arg); - } + int ix; + + ASSERT(erts_smp_thr_progress_is_blocking()); + + for (ix = 0; ix < erts_no_schedulers; ix++) { + ErtsSchedulerData *esdp = ERTS_SCHEDULER_IX(ix); + DbTable *first = esdp->ets_tables.clist; + if (first) { + DbTable *tb = first; + do { + if (is_table_alive(tb)) + (*func)(tb, arg); + tb = tb->common.all.next; + } while (tb != first); + } } - ASSERT(j == meta_main_tab_cnt); } /* SMP Note: May only be used when system is locked */ diff --git a/erts/emulator/beam/erl_db.h b/erts/emulator/beam/erl_db.h index 852440ff31..1c53401590 100644 --- a/erts/emulator/beam/erl_db.h +++ b/erts/emulator/beam/erl_db.h @@ -24,8 +24,37 @@ * */ -#ifndef __DB_H__ -#define __DB_H__ +#ifndef ERTS_DB_SCHED_SPEC_TYPES__ +#define ERTS_DB_SCHED_SPEC_TYPES__ + +union db_table; +typedef union db_table DbTable; + +typedef struct ErtsEtsAllReq_ ErtsEtsAllReq; + +typedef struct { + ErtsEtsAllReq *next; + ErtsEtsAllReq *prev; +} ErtsEtsAllReqList; + +typedef struct { + ErtsEtsAllReq *ongoing; + ErlHeapFragment *hfrag; + DbTable *tab; + ErtsEtsAllReq *queue; +} ErtsEtsAllYieldData; + +typedef struct { + Uint count; + DbTable *clist; +} ErtsEtsTables; + +#endif /* ERTS_DB_SCHED_SPEC_TYPES__ */ + +#ifndef ERTS_ONLY_SCHED_SPEC_ETS_DATA + +#ifndef ERL_DB_H__ +#define ERL_DB_H__ #include "sys.h" #undef ERL_THR_PROGRESS_TSD_TYPE_ONLY @@ -46,6 +75,12 @@ typedef struct { ErtsThrPrgrLaterOp data; } DbTableRelease; +struct ErtsSchedulerData_; +int erts_handle_yielded_ets_all_request(struct ErtsSchedulerData_ *esdp, + ErtsEtsAllYieldData *eadp); + +void erts_ets_sched_spec_data_init(struct ErtsSchedulerData_ *esdp); + /* * So, the structure for a database table, NB this is only * interesting in db.c. @@ -95,7 +130,7 @@ Eterm erts_ets_restore_meta_state(Process* p, Eterm target_state); Uint erts_db_get_max_tabs(void); -#endif +#endif /* ERL_DB_H__ */ #if defined(ERTS_WANT_DB_INTERNAL__) && !defined(ERTS_HAVE_DB_INTERNAL__) #define ERTS_HAVE_DB_INTERNAL__ @@ -271,3 +306,4 @@ erts_db_free_nt(ErtsAlcType_t type, void *ptr, Uint size) #endif /* #if defined(ERTS_WANT_DB_INTERNAL__) && !defined(ERTS_HAVE_DB_INTERNAL__) */ +#endif /* !ERTS_ONLY_SCHED_SPEC_ETS_DATA */ diff --git a/erts/emulator/beam/erl_db_util.h b/erts/emulator/beam/erl_db_util.h index 60dedd5713..3c13035ec5 100644 --- a/erts/emulator/beam/erl_db_util.h +++ b/erts/emulator/beam/erl_db_util.h @@ -75,9 +75,6 @@ typedef struct db_term { */ } DbTerm; -union db_table; -typedef union db_table DbTable; - #define DB_MUST_RESIZE 1 #define DB_NEW_OBJECT 2 #define DB_INC_TRY_GROW 4 @@ -203,6 +200,11 @@ typedef struct db_fixation { struct db_fixation *next; } DbFixation; +typedef struct { + DbTable *next; + DbTable *prev; +} DbTableList; + /* * This structure contains data for all different types of database * tables. Note that these fields must match the same fields @@ -214,6 +216,7 @@ typedef struct db_fixation { typedef struct db_table_common { erts_smp_refc_t refc; /* reference count of table struct */ erts_smp_refc_t fix_count;/* fixation counter */ + DbTableList all; #ifdef ERTS_SMP erts_smp_rwmtx_t rwlock; /* rw lock on table */ erts_smp_mtx_t fixlock; /* Protects fixations,megasec,sec,microsec */ diff --git a/erts/emulator/beam/erl_init.c b/erts/emulator/beam/erl_init.c index b2c307e826..61477af316 100644 --- a/erts/emulator/beam/erl_init.c +++ b/erts/emulator/beam/erl_init.c @@ -2307,6 +2307,7 @@ erl_start(int argc, char **argv) #endif set_main_stack_size(); erts_sched_init_time_sup(esdp); + erts_ets_sched_spec_data_init(esdp); process_main(esdp->x_reg_array, esdp->f_reg_array); } #endif diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c index 52d9f9ddf7..b816ba59fb 100644 --- a/erts/emulator/beam/erl_process.c +++ b/erts/emulator/beam/erl_process.c @@ -602,6 +602,7 @@ dbg_chk_aux_work_val(erts_aint32_t value) valid |= ERTS_SSI_AUX_WORK_REAP_PORTS; #endif valid |= ERTS_SSI_AUX_WORK_DEBUG_WAIT_COMPLETED; + valid |= ERTS_SSI_AUX_WORK_YIELD; if (~valid & value) erts_exit(ERTS_ABORT_EXIT, @@ -690,6 +691,8 @@ erts_pre_init_process(void) = "SET_TMO"; erts_aux_work_flag_descr[ERTS_SSI_AUX_WORK_MSEG_CACHE_CHECK_IX] = "MSEG_CACHE_CHECK"; + erts_aux_work_flag_descr[ERTS_SSI_AUX_WORK_YIELD_IX] + = "YIELD"; erts_aux_work_flag_descr[ERTS_SSI_AUX_WORK_REAP_PORTS_IX] = "REAP_PORTS"; erts_aux_work_flag_descr[ERTS_SSI_AUX_WORK_DEBUG_WAIT_COMPLETED_IX] @@ -2557,6 +2560,48 @@ handle_reap_ports(ErtsAuxWorkData *awdp, erts_aint32_t aux_work, int waiting) return aux_work & ~ERTS_SSI_AUX_WORK_REAP_PORTS; } +void +erts_notify_new_aux_yield_work(ErtsSchedulerData *esdp) +{ + ASSERT(esdp == erts_get_scheduler_data()); + /* Always called by the scheduler itself... */ + set_aux_work_flags_wakeup_nob(esdp->ssi, ERTS_SSI_AUX_WORK_YIELD); +} + +static ERTS_INLINE erts_aint32_t +handle_yield(ErtsAuxWorkData *awdp, erts_aint32_t aux_work, int waiting) +{ + int yield = 0; + /* + * Yield operations are always requested by the scheduler itself. + * + * The following handlers should *not* set the ERTS_SSI_AUX_WORK_YIELD + * flag in order to indicate more work. They should instead return + * information so this "main handler" can manipulate the flag... + * + * The following handlers should be able to handle being called + * even though no work is to be done... + */ + + /* Various yielding operations... */ + + yield |= erts_handle_yielded_ets_all_request(awdp->esdp, + &awdp->yield.ets_all); + + /* + * Other yielding operations... + * + */ + + if (!yield) { + unset_aux_work_flags(awdp->ssi, ERTS_SSI_AUX_WORK_YIELD); + return aux_work & ~ERTS_SSI_AUX_WORK_YIELD; + } + + return aux_work; +} + + #if HAVE_ERTS_MSEG static ERTS_INLINE erts_aint32_t @@ -2697,6 +2742,9 @@ handle_aux_work(ErtsAuxWorkData *awdp, erts_aint32_t orig_aux_work, int waiting) handle_mseg_cache_check); #endif + HANDLE_AUX_WORK(ERTS_SSI_AUX_WORK_YIELD, + handle_yield); + HANDLE_AUX_WORK(ERTS_SSI_AUX_WORK_REAP_PORTS, handle_reap_ports); @@ -8727,6 +8775,8 @@ sched_thread_func(void *vesdp) ERTS_VERIFY_UNUSED_TEMP_ALLOC(NULL); #endif + erts_ets_sched_spec_data_init(esdp); + process_main(esdp->x_reg_array, esdp->f_reg_array); /* No schedulers should *ever* terminate */ diff --git a/erts/emulator/beam/erl_process.h b/erts/emulator/beam/erl_process.h index ce0989883c..58f1e15c23 100644 --- a/erts/emulator/beam/erl_process.h +++ b/erts/emulator/beam/erl_process.h @@ -63,6 +63,9 @@ typedef struct process Process; #define ERTS_ONLY_INCLUDE_TRACE_FLAGS #include "erl_trace.h" #undef ERTS_ONLY_INCLUDE_TRACE_FLAGS +#define ERTS_ONLY_SCHED_SPEC_ETS_DATA +#include "erl_db.h" +#undef ERTS_ONLY_SCHED_SPEC_ETS_DATA #ifdef HIPE #include "hipe_process.h" @@ -312,6 +315,7 @@ typedef enum { ERTS_SSI_AUX_WORK_PENDING_EXITERS_IX, ERTS_SSI_AUX_WORK_SET_TMO_IX, ERTS_SSI_AUX_WORK_MSEG_CACHE_CHECK_IX, + ERTS_SSI_AUX_WORK_YIELD_IX, ERTS_SSI_AUX_WORK_REAP_PORTS_IX, ERTS_SSI_AUX_WORK_DEBUG_WAIT_COMPLETED_IX, /* SHOULD be last flag index */ @@ -348,6 +352,8 @@ typedef enum { (((erts_aint32_t) 1) << ERTS_SSI_AUX_WORK_SET_TMO_IX) #define ERTS_SSI_AUX_WORK_MSEG_CACHE_CHECK \ (((erts_aint32_t) 1) << ERTS_SSI_AUX_WORK_MSEG_CACHE_CHECK_IX) +#define ERTS_SSI_AUX_WORK_YIELD \ + (((erts_aint32_t) 1) << ERTS_SSI_AUX_WORK_YIELD_IX) #define ERTS_SSI_AUX_WORK_REAP_PORTS \ (((erts_aint32_t) 1) << ERTS_SSI_AUX_WORK_REAP_PORTS_IX) #define ERTS_SSI_AUX_WORK_DEBUG_WAIT_COMPLETED \ @@ -613,6 +619,10 @@ typedef struct { } delayed_wakeup; #endif struct { + ErtsEtsAllYieldData ets_all; + /* Other yielding operations... */ + } yield; + struct { struct { erts_aint32_t flags; void (*callback)(void *); @@ -621,6 +631,10 @@ typedef struct { } debug; } ErtsAuxWorkData; +#define ERTS_SCHED_AUX_YIELD_DATA(ESDP, NAME) \ + (&(ESDP)->aux_work_data.yield.NAME) +void erts_notify_new_aux_yield_work(ErtsSchedulerData *esdp); + #ifdef ERTS_DIRTY_SCHEDULERS typedef enum { ERTS_DIRTY_CPU_SCHEDULER, @@ -688,7 +702,7 @@ struct ErtsSchedulerData_ { ErtsSchedWallTime sched_wall_time; ErtsGCInfo gc_info; ErtsPortTaskHandle nosuspend_port_task_handle; - + ErtsEtsTables ets_tables; #ifdef ERTS_DO_VERIFY_UNUSED_TEMP_ALLOC erts_alloc_verify_func_t verify_unused_temp_alloc; Allctr_t *verify_unused_temp_alloc_data; diff --git a/lib/stdlib/src/ets.erl b/lib/stdlib/src/ets.erl index d6fd1e3ea1..90e19e6b9f 100644 --- a/lib/stdlib/src/ets.erl +++ b/lib/stdlib/src/ets.erl @@ -75,10 +75,28 @@ take/2, update_counter/3, update_counter/4, update_element/3]). +%% internal exports +-export([internal_request_all/0]). + -spec all() -> [Tab] when Tab :: tab(). all() -> + receive_all(ets:internal_request_all(), + erlang:system_info(schedulers), + []). + +receive_all(_Ref, 0, All) -> + All; +receive_all(Ref, N, All) -> + receive + {Ref, SchedAll} -> + receive_all(Ref, N-1, SchedAll ++ All) + end. + +-spec internal_request_all() -> reference(). + +internal_request_all() -> erlang:nif_error(undef). -spec delete(Tab) -> true when diff --git a/lib/stdlib/test/ets_SUITE.erl b/lib/stdlib/test/ets_SUITE.erl index 8581440d58..9e2f9608e8 100644 --- a/lib/stdlib/test/ets_SUITE.erl +++ b/lib/stdlib/test/ets_SUITE.erl @@ -76,6 +76,7 @@ -export([otp_10182/1]). -export([ets_all/1]). -export([memory_check_summary/1]). +-export([massive_ets_all/1]). -export([take/1]). -export([init_per_testcase/2, end_per_testcase/2]). @@ -134,6 +135,7 @@ all() -> otp_9932, otp_9423, ets_all, + massive_ets_all, take, memory_check_summary]. % MUST BE LAST @@ -5545,6 +5547,68 @@ ets_all_run() -> false = lists:member(Table, ets:all()), ets_all_run(). +create_tables(N) -> + create_tables(N, []). + +create_tables(0, Ts) -> + Ts; +create_tables(N, Ts) -> + create_tables(N-1, [ets:new(tjo, [])|Ts]). + +massive_ets_all(Config) when is_list(Config) -> + Me = self(), + InitTables = lists:sort(ets:all()), + io:format("InitTables=~p~n", [InitTables]), + PMs0 = lists:map(fun (Sid) -> + my_spawn_opt(fun () -> + Ts = create_tables(250), + Me ! {self(), up, Ts}, + receive {Me, die} -> ok end + end, + [link, monitor, {scheduler, Sid}]) + end, + lists:seq(1, erlang:system_info(schedulers_online))), + AllRes = lists:sort(lists:foldl(fun ({P, _M}, Ts) -> + receive + {P, up, PTs} -> + PTs ++ Ts + end + end, + InitTables, + PMs0)), + AllRes = lists:sort(ets:all()), + PMs1 = lists:map(fun (_) -> + my_spawn_opt(fun () -> + AllRes = lists:sort(ets:all()) + end, + [link, monitor]) + end, lists:seq(1, 50)), + lists:foreach(fun ({P, M}) -> + receive + {'DOWN', M, process, P, _} -> + ok + end + end, PMs1), + PMs2 = lists:map(fun (_) -> + my_spawn_opt(fun () -> + _ = ets:all() + end, + [link, monitor]) + end, lists:seq(1, 50)), + lists:foreach(fun ({P, _M}) -> + P ! {Me, die} + end, PMs0), + lists:foreach(fun ({P, M}) -> + receive + {'DOWN', M, process, P, _} -> + ok + end + end, PMs0 ++ PMs2), + EndTables = lists:sort(ets:all()), + io:format("EndTables=~p~n", [EndTables]), + InitTables = EndTables, + ok. + take(Config) when is_list(Config) -> %% Simple test for set tables. |