| % Licensed under the Apache License, Version 2.0 (the "License"); you may not |
| % use this file except in compliance with the License. You may obtain a copy of |
| % the License at |
| % |
| % http://www.apache.org/licenses/LICENSE-2.0 |
| % |
| % Unless required by applicable law or agreed to in writing, software |
| % distributed under the License is distributed on an "AS IS" BASIS, WITHOUT |
| % WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the |
| % License for the specific language governing permissions and limitations under |
| % the License. |
| |
| -module(couch_stats_resource_tracker). |
| |
| -behaviour(gen_server). |
| |
| -export([ |
| start_link/0, |
| init/1, |
| handle_call/3, |
| handle_cast/2, |
| handle_info/2, |
| code_change/3, |
| terminate/2 |
| ]). |
| |
| -export([ |
| pause_eviction/0, |
| enable_eviction/0, |
| eviction_status/0 |
| ]). |
| |
| -export([ |
| inc/1, inc/2, |
| maybe_inc/2, |
| get_pid_ref/0, |
| accumulate_delta/1 |
| ]). |
| |
| -export([ |
| close_pid_ref/0, close_pid_ref/1, |
| create_context/3, |
| create_coordinator_context/1, create_coordinator_context/2, |
| destroy_context/0, destroy_context/1, |
| is_enabled/0, |
| get_resource/0, |
| get_resource/1, |
| set_context_dbname/1, |
| set_context_handler_fun/1, |
| set_context_username/1, |
| track/1, tracker/1, |
| stop_tracker/0, stop_tracker/1, stop_tracker/2, |
| should_track/1 |
| ]). |
| |
| -export([ |
| active/0, |
| active_coordinators/0, |
| active_workers/0, |
| find_unmonitored/0 |
| ]). |
| |
| -export([ |
| count_by/1, |
| group_by/2, |
| group_by/3, |
| group_by/4, |
| sorted/1, |
| sorted_by/1, |
| sorted_by/2, |
| sorted_by/3, |
| |
| find_by_pid/1, |
| |
| unsafe_foldl/3, |
| |
| term_to_flat_json/1 |
| ]). |
| |
| -export([ |
| make_delta/0 |
| ]). |
| |
| %% Singular increment operations |
| -export([ |
| db_opened/0, |
| doc_read/0, |
| row_read/0, |
| btree_fold/0, |
| ioq_called/0, |
| js_evaled/0, |
| js_filtered/0, |
| js_filtered_error/0, |
| js_filtered_doc/0, |
| mango_match_evaled/0, |
| get_kv_node/0, |
| get_kp_node/0 |
| ]). |
| |
| %% Plural increment operations |
| -export([ |
| js_filtered_docs/1, |
| io_bytes_read/1, |
| io_bytes_written/1 |
| ]). |
| |
| -export([ |
| field/2, |
| curry_field/1 |
| ]). |
| |
| -include_lib("couch/include/couch_db.hrl"). |
| |
| %% Use these for record upgrades over the wire and in ETS tables |
| %% TODO: alternatively, just delete these. Currently using a map |
| %% for shipping deltas over the wire, avoiding much of the |
| %% problem here. We'll likely still need to handle upgrades to |
| %% map format over time, so let's decide a course of action here. |
| -define(RCTX_V1, rctx_v1). |
| -define(RCTX, ?RCTX_V1). |
| |
| -define(MANGO_EVAL_MATCH, mango_eval_match). |
| -define(DB_OPEN_DOC, docs_read). |
| -define(DB_OPEN, db_open). |
| -define(COUCH_SERVER_OPEN, db_open). |
| -define(COUCH_BT_FOLDS, btree_folds). |
| -define(COUCH_BT_GET_KP_NODE, get_kp_node). |
| -define(COUCH_BT_GET_KV_NODE, get_kv_node). |
| -define(COUCH_JS_FILTER, js_filter). |
| -define(COUCH_JS_FILTER_ERROR, js_filter_error). |
| -define(COUCH_JS_FILTERED_DOCS, js_filtered_docs). |
| -define(ROWS_READ, rows_read). |
| |
| %% TODO: overlap between this and couch btree fold invocations |
| %% TODO: need some way to distinguish fols on views vs find vs all_docs |
| -define(FRPC_CHANGES_ROW, changes_processed). |
| -define(FRPC_CHANGES_RETURNED, changes_returned). |
| %%-define(FRPC_CHANGES_ROW, ?ROWS_READ). |
| |
| %% Module pdict markers |
| -define(DELTA_TA, csrt_delta_ta). |
| -define(DELTA_TZ, csrt_delta_tz). %% T Zed instead of T0 |
| -define(PID_REF, csrt_pid_ref). %% track local ID |
| -define(TRACKER_PID, csrt_tracker). %% tracker pid |
| |
| |
| -record(st, { |
| eviction = enabled, %% or paused |
| ev_queue = [], %% eviction queue for when eviction is paused |
| eviction_delay = 10 * 1000, %% How many ms dead processes are visible |
| scan_interval = 2048, %% How regularly to perfom scans |
| tracking = #{} %% track active processes for eventual eviction |
| }). |
| |
| |
| %% TODO: switch to: |
| %% -record(?RCTX, { |
| -record(rctx, { |
| %% Metadata |
| started_at = tnow(), |
| updated_at = tnow(), |
| exited_at, %% TODO: do we need a final exit time and additional update times afterwards? |
| pid_ref, |
| mon_ref, |
| mfa, |
| nonce, |
| from, |
| type = unknown, %% unknown/background/system/rpc/coordinator/fabric_rpc/etc_rpc/etc |
| state = alive, |
| dbname, |
| username, |
| path, |
| |
| %% Stats counters |
| db_open = 0, |
| docs_read = 0, |
| rows_read = 0, |
| btree_folds = 0, |
| changes_processed = 0, |
| changes_returned = 0, |
| ioq_calls = 0, |
| io_bytes_read = 0, |
| io_bytes_written = 0, |
| js_evals = 0, |
| js_filter = 0, |
| js_filter_error = 0, |
| js_filtered_docs = 0, |
| mango_eval_match = 0, |
| %% TODO: switch record definitions to be macro based, eg: |
| %% ?COUCH_BT_GET_KP_NODE = 0, |
| get_kv_node = 0, |
| get_kp_node = 0 |
| }). |
| |
| %% monotonic time now in millisecionds |
| tnow() -> |
| erlang:monotonic_time(millisecond). |
| |
| is_enabled() -> |
| config:get_boolean(?MODULE_STRING, "enabled", true). |
| |
| should_scan() -> |
| config:get_boolean(?MODULE_STRING, "scans_enabled", false). |
| |
| db_opened() -> inc(db_opened). |
| doc_read() -> inc(docs_read). |
| row_read() -> inc(rows_read). |
| btree_fold() -> inc(?COUCH_BT_FOLDS). |
| %% TODO: do we need ioq_called and this access pattern? |
| ioq_called() -> is_enabled() andalso inc(ioq_calls). |
| js_evaled() -> inc(js_evals). |
| js_filtered() -> inc(js_filter). |
| js_filtered_error() -> inc(js_filter_error). |
| js_filtered_doc() -> inc(js_filtered_docs). |
| mango_match_evaled() -> inc(mango_eval_match). |
| get_kv_node() -> inc(get_kv_node). |
| get_kp_node() -> inc(get_kp_node). |
| |
| js_filtered_docs(N) -> inc(js_filtered_docs, N). |
| io_bytes_read(N) -> inc(io_bytes_read, N). |
| io_bytes_written(N) -> inc(io_bytes_written, N). |
| |
| inc(?DB_OPEN) -> |
| inc(?DB_OPEN, 1); |
| inc(docs_read) -> |
| inc(docs_read, 1); |
| inc(?ROWS_READ) -> |
| inc(?ROWS_READ, 1); |
| inc(?FRPC_CHANGES_RETURNED) -> |
| inc(?FRPC_CHANGES_RETURNED, 1); |
| inc(?COUCH_BT_FOLDS) -> |
| inc(?COUCH_BT_FOLDS, 1); |
| inc(ioq_calls) -> |
| inc(ioq_calls, 1); |
| inc(io_bytes_read) -> |
| inc(io_bytes_read, 1); |
| inc(io_bytes_written) -> |
| inc(io_bytes_written, 1); |
| inc(js_evals) -> |
| inc(js_evals, 1); |
| inc(?COUCH_JS_FILTER) -> |
| inc(?COUCH_JS_FILTER, 1); |
| inc(?COUCH_JS_FILTER_ERROR) -> |
| inc(?COUCH_JS_FILTER_ERROR, 1); |
| inc(?COUCH_JS_FILTERED_DOCS) -> |
| inc(?COUCH_JS_FILTERED_DOCS, 1); |
| inc(?MANGO_EVAL_MATCH) -> |
| inc(?MANGO_EVAL_MATCH, 1); |
| inc(?COUCH_BT_GET_KP_NODE) -> |
| inc(?COUCH_BT_GET_KP_NODE, 1); |
| inc(?COUCH_BT_GET_KV_NODE) -> |
| inc(?COUCH_BT_GET_KV_NODE, 1); |
| inc(_) -> |
| 0. |
| |
| |
| inc(?DB_OPEN, N) -> |
| update_counter(#rctx.?DB_OPEN, N); |
| inc(?ROWS_READ, N) -> |
| update_counter(#rctx.?ROWS_READ, N); |
| inc(?FRPC_CHANGES_RETURNED, N) -> |
| update_counter(#rctx.?FRPC_CHANGES_RETURNED, N); |
| inc(ioq_calls, N) -> |
| update_counter(#rctx.ioq_calls, N); |
| inc(io_bytes_read, N) -> |
| update_counter(#rctx.io_bytes_read, N); |
| inc(io_bytes_written, N) -> |
| update_counter(#rctx.io_bytes_written, N); |
| inc(js_evals, N) -> |
| update_counter(#rctx.js_evals, N); |
| inc(?COUCH_JS_FILTER, N) -> |
| update_counter(#rctx.?COUCH_JS_FILTER, N); |
| inc(?COUCH_JS_FILTER_ERROR, N) -> |
| update_counter(#rctx.?COUCH_JS_FILTER_ERROR, N); |
| inc(?COUCH_JS_FILTERED_DOCS, N) -> |
| update_counter(#rctx.?COUCH_JS_FILTERED_DOCS, N); |
| inc(?MANGO_EVAL_MATCH, N) -> |
| update_counter(#rctx.?MANGO_EVAL_MATCH, N); |
| inc(?DB_OPEN_DOC, N) -> |
| update_counter(#rctx.?DB_OPEN_DOC, N); |
| inc(?FRPC_CHANGES_ROW, N) -> |
| update_counter(#rctx.?ROWS_READ, N); %% TODO: rework double use of rows_read |
| inc(?COUCH_BT_GET_KP_NODE, N) -> |
| update_counter(#rctx.?COUCH_BT_GET_KP_NODE, N); |
| inc(?COUCH_BT_GET_KV_NODE, N) -> |
| update_counter(#rctx.?COUCH_BT_GET_KV_NODE, N); |
| inc(_, _) -> |
| %% inc needs to allow unknown types to pass for accumulate_update to handle |
| %% updates from nodes with newer data formats |
| 0. |
| |
| maybe_inc([mango, evaluate_selector], Val) -> |
| inc(?MANGO_EVAL_MATCH, Val); |
| maybe_inc([couchdb, database_reads], Val) -> |
| inc(?DB_OPEN_DOC, Val); |
| maybe_inc([fabric_rpc, changes, processed], Val) -> |
| inc(?FRPC_CHANGES_ROW, Val); |
| maybe_inc([fabric_rpc, changes, returned], Val) -> |
| inc(?FRPC_CHANGES_RETURNED, Val); |
| maybe_inc([fabric_rpc, view, rows_read], Val) -> |
| inc(?ROWS_READ, Val); |
| maybe_inc([couchdb, couch_server, open], Val) -> |
| inc(?DB_OPEN, Val); |
| maybe_inc([couchdb, btree, folds], Val) -> |
| inc(?COUCH_BT_FOLDS, Val); |
| maybe_inc([couchdb, btree, kp_node], Val) -> |
| inc(?COUCH_BT_GET_KP_NODE, Val); |
| maybe_inc([couchdb, btree, kv_node], Val) -> |
| inc(?COUCH_BT_GET_KV_NODE, Val); |
| maybe_inc([couchdb, query_server, js_filter_error], Val) -> |
| inc(?COUCH_JS_FILTER_ERROR, Val); |
| maybe_inc([couchdb, query_server, js_filter], Val) -> |
| inc(?COUCH_JS_FILTER, Val); |
| maybe_inc([couchdb, query_server, js_filtered_docs], Val) -> |
| inc(?COUCH_JS_FILTERED_DOCS, Val); |
| maybe_inc(_Metric, _Val) -> |
| %%io:format("SKIPPING MAYBE_INC METRIC[~p]: ~p~n", [Val, Metric]), |
| 0. |
| |
| |
| %% TODO: update stats_descriptions.cfg for relevant apps |
| should_track([fabric_rpc, all_docs, spawned]) -> |
| is_enabled(); |
| should_track([fabric_rpc, changes, spawned]) -> |
| is_enabled(); |
| should_track([fabric_rpc, changes, processed]) -> |
| is_enabled(); |
| should_track([fabric_rpc, changes, returned]) -> |
| is_enabled(); |
| should_track([fabric_rpc, map_view, spawned]) -> |
| is_enabled(); |
| should_track([fabric_rpc, reduce_view, spawned]) -> |
| is_enabled(); |
| should_track([fabric_rpc, get_all_security, spawned]) -> |
| is_enabled(); |
| should_track([fabric_rpc, open_doc, spawned]) -> |
| is_enabled(); |
| should_track([fabric_rpc, update_docs, spawned]) -> |
| is_enabled(); |
| should_track([fabric_rpc, open_shard, spawned]) -> |
| is_enabled(); |
| should_track([mango_cursor, view, all_docs]) -> |
| is_enabled(); |
| should_track([mango_cursor, view, idx]) -> |
| is_enabled(); |
| should_track(_Metric) -> |
| %%io:format("SKIPPING METRIC: ~p~n", [Metric]), |
| false. |
| |
| accumulate_delta(Delta) when is_map(Delta) -> |
| %% TODO: switch to creating a batch of updates to invoke a single |
| %% update_counter rather than sequentially invoking it for each field |
| is_enabled() andalso maps:foreach(fun inc/2, Delta); |
| accumulate_delta(undefined) -> |
| ok; |
| accumulate_delta(Other) -> |
| io:format("CSRT:ACC_DELTA UNKNOWN DELTA: ~p~n", [Other]). |
| |
| |
| update_counter(Field, Count) -> |
| is_enabled() andalso update_counter(get_pid_ref(), Field, Count). |
| |
| |
| update_counter(undefined, _Field, _Count) -> |
| ok; |
| update_counter({_Pid,_Ref}=PidRef, Field, Count) -> |
| %% TODO: mem3 crashes without catch, why do we lose the stats table? |
| is_enabled() andalso catch ets:update_counter(?MODULE, PidRef, {Field, Count}, #rctx{pid_ref=PidRef}). |
| |
| |
| active() -> active_int(all). |
| active_coordinators() -> active_int(coordinators). |
| active_workers() -> active_int(workers). |
| |
| |
| active_int(coordinators) -> |
| select_by_type(coordinators); |
| active_int(workers) -> |
| select_by_type(workers); |
| active_int(all) -> |
| lists:map(fun to_flat_json/1, ets:tab2list(?MODULE)). |
| |
| |
| select_by_type(coordinators) -> |
| ets:select(couch_stats_resource_tracker, |
| [{#rctx{type = {coordinator,'_','_'}, _ = '_'}, [], ['$_']}]); |
| select_by_type(workers) -> |
| ets:select(couch_stats_resource_tracker, |
| [{#rctx{type = {worker,'_','_'}, _ = '_'}, [], ['$_']}]); |
| select_by_type(all) -> |
| lists:map(fun to_flat_json/1, ets:tab2list(?MODULE)). |
| |
| |
| field(#rctx{pid_ref=Val}, pid_ref) -> Val; |
| %% NOTE: Pros and cons to doing these convert functions here |
| %% Ideally, this would be done later so as to prefer the core data structures |
| %% as long as possible, but we currently need the output of this function to |
| %% be jiffy:encode'able. The tricky bit is dynamically encoding the group_by |
| %% structure provided by the caller of *_by aggregator functions below. |
| %% For now, we just always return jiffy:encode'able data types. |
| field(#rctx{mfa=Val}, mfa) -> convert_mfa(Val); |
| field(#rctx{nonce=Val}, nonce) -> Val; |
| field(#rctx{from=Val}, from) -> Val; |
| field(#rctx{type=Val}, type) -> convert_type(Val); |
| field(#rctx{state=Val}, state) -> Val; |
| field(#rctx{dbname=Val}, dbname) -> Val; |
| field(#rctx{username=Val}, username) -> Val; |
| field(#rctx{path=Val}, path) -> Val; |
| field(#rctx{db_open=Val}, db_open) -> Val; |
| field(#rctx{docs_read=Val}, docs_read) -> Val; |
| field(#rctx{rows_read=Val}, rows_read) -> Val; |
| field(#rctx{btree_folds=Val}, btree_folds) -> Val; |
| field(#rctx{changes_processed=Val}, changes_processed) -> Val; |
| field(#rctx{changes_returned=Val}, changes_returned) -> Val; |
| field(#rctx{ioq_calls=Val}, ioq_calls) -> Val; |
| field(#rctx{io_bytes_read=Val}, io_bytes_read) -> Val; |
| field(#rctx{io_bytes_written=Val}, io_bytes_written) -> Val; |
| field(#rctx{js_evals=Val}, js_evals) -> Val; |
| field(#rctx{js_filter=Val}, js_filter) -> Val; |
| field(#rctx{js_filter_error=Val}, js_filter_error) -> Val; |
| field(#rctx{js_filtered_docs=Val}, js_filtered_docs) -> Val; |
| field(#rctx{mango_eval_match=Val}, mango_eval_match) -> Val; |
| field(#rctx{get_kv_node=Val}, get_kv_node) -> Val; |
| field(#rctx{get_kp_node=Val}, get_kp_node) -> Val. |
| |
| |
| curry_field(Field) -> |
| fun(Ele) -> field(Ele, Field) end. |
| |
| |
| count_by(KeyFun) -> |
| group_by(KeyFun, fun(_) -> 1 end). |
| |
| |
| group_by(KeyFun, ValFun) -> |
| group_by(KeyFun, ValFun, fun erlang:'+'/2). |
| |
| |
| group_by(KeyFun, ValFun, AggFun) -> |
| Fold = case conf_get("fold_fun") of |
| "unsafe" -> |
| fun unsafe_foldl/3; |
| _ -> |
| fun ets:foldl/3 |
| end, |
| group_by(KeyFun, ValFun, AggFun, Fold). |
| |
| |
| %% eg: group_by(mfa, docs_read). |
| %% eg: group_by(fun(#rctx{mfa=MFA,docs_read=DR}) -> {MFA, DR} end, ioq_calls). |
| %% eg: ^^ or: group_by([mfa, docs_read], ioq_calls). |
| %% eg: group_by([username, dbname, mfa], docs_read). |
| %% eg: group_by([username, dbname, mfa], ioq_calls). |
| %% eg: group_by([username, dbname, mfa], js_filters). |
| group_by(KeyL, ValFun, AggFun, Fold) when is_list(KeyL) -> |
| KeyFun = fun(Ele) -> list_to_tuple([field(Ele, Key) || Key <- KeyL]) end, |
| group_by(KeyFun, ValFun, AggFun, Fold); |
| group_by(Key, ValFun, AggFun, Fold) when is_atom(Key) -> |
| group_by(curry_field(Key), ValFun, AggFun, Fold); |
| group_by(KeyFun, Val, AggFun, Fold) when is_atom(Val) -> |
| group_by(KeyFun, curry_field(Val), AggFun, Fold); |
| group_by(KeyFun, ValFun, AggFun, Fold) -> |
| FoldFun = fun(Ele, Acc) -> |
| Key = KeyFun(Ele), |
| Val = ValFun(Ele), |
| CurrVal = maps:get(Key, Acc, 0), |
| NewVal = AggFun(CurrVal, Val), |
| %% TODO: should we skip here? how to make this optional? |
| case NewVal > 0 of |
| true -> |
| maps:put(Key, NewVal, Acc); |
| false -> |
| Acc |
| end |
| end, |
| case conf_get("fold_fun") of |
| "unsafe" -> |
| %% When fold_fun is unsafe we must pause deletion of keys as |
| %% otherwise `ets:next/2` will fail when the supplied key has |
| %% already been deleted |
| pause_eviction(), |
| try |
| Fold(FoldFun, #{}, ?MODULE) |
| after |
| enable_eviction() |
| end; |
| _ -> |
| Fold(FoldFun, #{}, ?MODULE) |
| end. |
| |
| |
| %% Sorts largest first |
| sorted(Map) when is_map(Map) -> |
| lists:sort(fun({_K1, A}, {_K2, B}) -> B < A end, maps:to_list(Map)). |
| |
| shortened(L) -> |
| lists:sublist(L, 10). |
| |
| |
| %% eg: sorted_by([username, dbname, mfa], ioq_calls) |
| %% eg: sorted_by([dbname, mfa], doc_reads) |
| sorted_by(KeyFun) -> shortened(sorted(count_by(KeyFun))). |
| sorted_by(KeyFun, ValFun) -> shortened(sorted(group_by(KeyFun, ValFun))). |
| sorted_by(KeyFun, ValFun, AggFun) -> shortened(sorted(group_by(KeyFun, ValFun, AggFun))). |
| |
| |
| term_to_flat_json({shutdown, Reason0}) when is_atom(Reason0) -> |
| Reason = atom_to_binary(Reason0), |
| <<"shutdown: ", Reason/binary>>; |
| term_to_flat_json({type, Atom}) when is_atom(Atom) -> |
| atom_to_binary(Atom); |
| term_to_flat_json({type, {coordinator, Verb0, Path0}}=_Type) -> |
| Verb = atom_to_binary(Verb0), |
| Path = list_to_binary(Path0), |
| <<"coordinator:", Verb/binary, ":", Path/binary>>; |
| term_to_flat_json({type, {worker, M0, F0}}=_Type) -> |
| M = atom_to_binary(M0), |
| F = atom_to_binary(F0), |
| <<"worker:", M/binary, ":", F/binary>>; |
| term_to_flat_json(Tuple) when is_tuple(Tuple) -> |
| erlang:tuple_to_list(Tuple); |
| term_to_flat_json(Pid) when is_pid(Pid) -> |
| ?l2b(pid_to_list(Pid)); |
| term_to_flat_json(Ref) when is_reference(Ref) -> |
| ?l2b(ref_to_list(Ref)); |
| term_to_flat_json(Atom) when is_atom(Atom) -> |
| atom_to_binary(Atom); |
| term_to_flat_json(undefined) -> |
| null; |
| term_to_flat_json(null) -> |
| null; |
| term_to_flat_json(T) -> |
| T. |
| |
| to_flat_json(#rctx{}=Rctx) -> |
| #rctx{ |
| updated_at = TP, |
| started_at = TInit, |
| pid_ref = {Pid0, Ref0}, |
| mfa = MFA0, |
| nonce = Nonce0, |
| from = From0, |
| dbname = DbName, |
| username = UserName, |
| db_open = DbOpens, |
| docs_read = DocsRead, |
| rows_read = RowsRead, |
| js_filter = JSFilters, |
| js_filter_error = JSFilterErrors, |
| js_filtered_docs = JSFilteredDocss, |
| state = State0, |
| type = Type, |
| get_kp_node = KpNodes, |
| get_kv_node = KvNodes, |
| btree_folds = ChangesProcessed, |
| changes_returned = ChangesReturned, |
| ioq_calls = IoqCalls |
| } = Rctx, |
| Pid = term_to_flat_json(Pid0), |
| Ref = term_to_flat_json(Ref0), |
| PidRef = <<Pid/binary, ":", Ref/binary>>, |
| MFA = case MFA0 of |
| {M0, F0, A0} -> |
| M = atom_to_binary(M0), |
| F = atom_to_binary(F0), |
| A = integer_to_binary(A0), |
| <<M/binary, ":", F/binary, "/", A/binary>>; |
| MFA0 when is_list(MFA0) -> |
| MFA0; |
| undefined -> |
| null; |
| OtherMFA -> |
| throw({error, {unexpected, OtherMFA}}) |
| end, |
| From = case From0 of |
| {Parent0, ParentRef0} -> |
| Parent = term_to_flat_json(Parent0), |
| ParentRef = term_to_flat_json(ParentRef0), |
| <<Parent/binary, ":", ParentRef/binary>>; |
| undefined -> |
| null |
| end, |
| State = case State0 of |
| alive -> |
| alive; |
| {down, Reason0} -> |
| Reason = term_to_flat_json(Reason0), |
| <<"down: ", Reason/binary>> |
| end, |
| Nonce = case Nonce0 of |
| undefined -> |
| null; |
| Nonce0 -> |
| list_to_binary(Nonce0) |
| end, |
| #{ |
| updated_at => TP, |
| started_at => TInit, |
| pid_ref => PidRef, |
| mfa => MFA, |
| nonce => Nonce, |
| from => From, |
| dbname => DbName, |
| username => UserName, |
| db_open => DbOpens, |
| docs_read => DocsRead, |
| js_filter => JSFilters, |
| js_filter_error => JSFilterErrors, |
| js_filtered_docs => JSFilteredDocss, |
| rows_read => RowsRead, |
| state => State, |
| type => term_to_flat_json({type, Type}), |
| kp_nodes => KpNodes, |
| kv_nodes => KvNodes, |
| btree_folds => ChangesProcessed, |
| changes_returned => ChangesReturned, |
| ioq_calls => IoqCalls |
| }. |
| |
| |
| convert_mfa(MFA) when is_list(MFA) -> |
| list_to_binary(MFA); |
| convert_mfa({M0, F0, A0}) -> |
| M = atom_to_binary(M0), |
| F = atom_to_binary(F0), |
| A = integer_to_binary(A0), |
| <<M/binary, ":", F/binary, "/", A/binary>>; |
| convert_mfa(undefined) -> |
| null. |
| |
| convert_type(Atom) when is_atom(Atom) -> |
| atom_to_binary(Atom); |
| convert_type({coordinator, Verb0, Atom0}) when is_atom(Atom0) -> |
| Verb = atom_to_binary(Verb0), |
| Atom = atom_to_binary(Atom0), |
| <<"coordinator:", Verb/binary, ":", Atom/binary>>; |
| convert_type({coordinator, Verb0, Path0}) -> |
| Verb = atom_to_binary(Verb0), |
| Path = list_to_binary(Path0), |
| <<"coordinator:", Verb/binary, ":", Path/binary>>; |
| convert_type({worker, M0, F0}) -> |
| M = atom_to_binary(M0), |
| F = atom_to_binary(F0), |
| <<"worker:", M/binary, ":", F/binary>>. |
| |
| get_pid_ref() -> |
| get(?PID_REF). |
| |
| |
| create_pid_ref() -> |
| case get(?PID_REF) of |
| undefined -> |
| ok; |
| PidRef0 -> |
| %% TODO: what to do when it already exists? |
| throw({epidexist, PidRef0}), |
| close_pid_ref(PidRef0) |
| end, |
| PidRef = {self(), make_ref()}, |
| set_pid_ref(PidRef), |
| PidRef. |
| |
| |
| close_pid_ref() -> |
| close_pid_ref(get_pid_ref()). |
| |
| |
| close_pid_ref(undefined) -> |
| undefined; |
| close_pid_ref(_PidRef) -> |
| erase(?PID_REF). |
| |
| destroy_context() -> |
| destroy_context(get_pid_ref()). |
| |
| |
| destroy_context(undefined) -> |
| ok; |
| destroy_context({_, _} = PidRef) -> |
| stop_tracker(PidRef), |
| close_pid_ref(PidRef), |
| gen_server:cast(?MODULE, {destroy, PidRef}), |
| ok. |
| |
| |
| create_resource(#rctx{} = Rctx) -> |
| %% true = ets:insert(?MODULE, Rctx). |
| catch ets:insert(?MODULE, Rctx). |
| |
| |
| %% add type to disnguish coordinator vs rpc_worker |
| create_context(From, {M,F,_A} = MFA, Nonce) -> |
| case is_enabled() of |
| false -> |
| undefined; |
| true -> |
| PidRef = create_pid_ref(), |
| %% TODO: extract user_ctx and db/shard from |
| Rctx = #rctx{ |
| pid_ref = PidRef, |
| from = From, |
| mfa = MFA, |
| type = {worker, M, F}, |
| nonce = Nonce |
| }, |
| track(Rctx), |
| erlang:put(?DELTA_TZ, Rctx), |
| create_resource(Rctx), |
| PidRef |
| end. |
| |
| create_coordinator_context(#httpd{path_parts=Parts} = Req) -> |
| is_enabled() andalso create_coordinator_context(Req, io_lib:format("~p", [Parts])). |
| |
| create_coordinator_context(#httpd{} = Req, Path) -> |
| case is_enabled() of |
| false -> |
| ok; |
| true -> |
| #httpd{ |
| method = Verb, |
| nonce = Nonce |
| } = Req, |
| PidRef = create_pid_ref(), |
| Rctx = #rctx{ |
| pid_ref = PidRef, |
| %%type = {coordinator, Verb, Path}, |
| type = {coordinator, Verb, init}, |
| nonce = Nonce, |
| path = list_to_binary([$/ | Path]) |
| }, |
| track(Rctx), |
| erlang:put(?DELTA_TZ, Rctx), |
| create_resource(Rctx), |
| PidRef |
| end. |
| |
| set_context_dbname(DbName) -> |
| set_context_dbname(DbName, get_pid_ref()). |
| |
| set_context_dbname(_, undefined) -> |
| ok; |
| set_context_dbname(DbName, PidRef) -> |
| case is_enabled() of |
| false -> |
| ok; |
| true -> |
| catch case ets:update_element(?MODULE, PidRef, [{#rctx.dbname, DbName}]) of |
| false -> |
| Stk = try throw(42) catch _:_:Stk0 -> Stk0 end, |
| io:format("UPDATING DBNAME[~p] FAILURE WITH CONTEXT: ~p AND STACK:~n~pFOO:: ~p~n~n", [DbName, get_resource(), Stk, process_info(self(), current_stacktrace)]), |
| timer:sleep(1000), |
| erlang:halt(kaboomz); |
| true -> |
| true |
| end |
| end. |
| |
| set_context_handler_fun(Fun) when is_function(Fun) -> |
| set_context_handler_fun(Fun, get_pid_ref()). |
| |
| set_context_handler_fun(_, undefined) -> |
| ok; |
| set_context_handler_fun(Fun, PidRef) when is_function(Fun) -> |
| case is_enabled() of |
| false -> |
| ok; |
| true -> |
| FunName = erlang:fun_to_list(Fun), |
| #rctx{type={coordinator, Verb, _}} = get_resource(), |
| Update = [{#rctx.type, {coordinator, Verb, FunName}}], |
| catch case ets:update_element(?MODULE, PidRef, Update) of |
| false -> |
| Stk = try throw(42) catch _:_:Stk0 -> Stk0 end, |
| io:format("UPDATING HANDLER FUN[~p] FAILURE WITH CONTEXT: ~p AND STACK:~n~pFOO:: ~p~n~n", [FunName, get_resource(), Stk, process_info(self(), current_stacktrace)]), |
| timer:sleep(1000), |
| erlang:halt(kaboomz); |
| true -> |
| true |
| end |
| end. |
| |
| set_context_username(null) -> |
| ok; |
| set_context_username(UserName) -> |
| set_context_username(UserName, get_pid_ref()). |
| |
| set_context_username(_, undefined) -> |
| ok; |
| set_context_username(UserName, PidRef) -> |
| case is_enabled() of |
| false -> |
| ok; |
| true -> |
| catch case ets:update_element(?MODULE, PidRef, [{#rctx.username, UserName}]) of |
| false -> |
| Stk = try throw(42) catch _:_:Stk0 -> Stk0 end, |
| io:format("UPDATING USERNAME[~p] FAILURE WITH CONTEXT: ~p AND STACK:~n~pFOO:: ~p~n~n", [UserName, get_resource(), Stk, process_info(self(), current_stacktrace)]), |
| timer:sleep(1000), |
| erlang:halt(kaboomz); |
| true -> |
| true |
| end |
| end. |
| |
| track(#rctx{}=Rctx) -> |
| case conf_get("spawn_monitor", "true") of |
| "true" -> |
| track(Rctx, spawn_monitor); |
| _ -> |
| track(Rctx, gen_server) |
| end. |
| |
| track(#rctx{pid_ref=PidRef}, spawn_monitor) -> |
| case get_tracker() of |
| undefined -> |
| Pid = spawn(?MODULE, tracker, [PidRef]), |
| put_tracker(Pid); |
| Pid when is_pid(Pid) -> |
| Pid |
| end; |
| track(#rctx{pid_ref=PidRef}, gen_server) -> |
| %% TODO: should this block or not? If no, what cleans up zombies? |
| %% gen_server:call(?MODULE, {track, PR}). |
| gen_server:cast(?MODULE, {track, PidRef}). |
| |
| get_tracker() -> |
| get(?TRACKER_PID). |
| |
| put_tracker(Pid) when is_pid(Pid) -> |
| put(?TRACKER_PID, Pid). |
| |
| tracker({Pid, _Ref}=PidRef) -> |
| MonRef = erlang:monitor(process, Pid), |
| should_scan() andalso catch ets:update_element(?MODULE, PidRef, [{#rctx.mon_ref, MonRef}]), |
| receive |
| stop -> |
| %% TODO: do we need cleanup here? |
| %%cleanup_tracker(PidRef, <<"shutdown:stopped">>), |
| demonitor(MonRef), |
| ok; |
| {'DOWN', MonRef, _Type, _0DPid, Reason0} -> |
| Reason = case Reason0 of |
| {shutdown, Shutdown0} -> |
| Shutdown = atom_to_binary(Shutdown0), |
| <<"shutdown: ", Shutdown/binary>>; |
| Reason0 -> |
| Reason0 |
| end, |
| cleanup_tracker(PidRef, Reason) |
| end. |
| |
| cleanup_tracker(PidRef, Reason) -> |
| case is_logging_enabled() andalso get_resource(PidRef) of |
| #rctx{} = Rctx -> |
| should_log(Rctx) andalso log_process_lifetime_report(PidRef, Rctx); |
| _ -> |
| ok |
| end, |
| %% update stats |
| should_scan() andalso catch ets:update_element(?MODULE, PidRef, |
| [{#rctx.state, {down, Reason}}, {#rctx.updated_at, tnow()}]), |
| catch evict(PidRef). |
| |
| |
| stop_tracker() -> |
| stop_tracker(get_tracker(), get_pid_ref()). |
| |
| stop_tracker({_Pid, _Ref}=PidRef) -> |
| stop_tracker(get_tracker(), PidRef); |
| stop_tracker(Pid) when is_pid(Pid) -> |
| stop_tracker(Pid, get_pid_ref()). |
| |
| stop_tracker(undefined, _) -> |
| ok; |
| stop_tracker(Pid, PidRef) -> |
| should_scan() andalso catch ets:update_element(?MODULE, PidRef, [{#rctx.mon_ref, undefined}]), |
| Pid ! stop. |
| |
| |
| make_delta() -> |
| TA = case get(?DELTA_TA) of |
| undefined -> |
| %% Need to handle this better, can't just make a new T0 at T' as |
| %% the timestamps will be identical causing a divide by zero error. |
| %% |
| %% Realistically need to ensure that all invocations of database |
| %% operations sets T0 appropriately. Perhaps it's possible to do |
| %% this is the couch_db:open chain, and then similarly, in |
| %% couch_server, and uhhhh... couch_file, and... |
| %% |
| %% I think we need some type of approach for establishing a T0 that |
| %% doesn't result in outrageous deltas. For now zero out the |
| %% microseconds field, or subtract a second on the off chance that |
| %% microseconds is zero. I'm not uptodate on the latest Erlang time |
| %% libraries and don't remember how to easily get an |
| %% `os:timestamp()` out of now() - 100ms or some such. |
| %% |
| %% I think it's unavoidable that we'll have some codepaths that do |
| %% not properly instantiate the T0 at spawn resulting in needing to |
| %% do some time of "time warp" or ignoring the timing collection |
| %% entirely. Perhaps if we hoisted out the stats collection into |
| %% the primary flow of the database and funnel that through all the |
| %% function clauses we could then utilize Dialyzer to statically |
| %% analyze and assert all code paths that invoke database |
| %% operations have properly instantinated a T0 at the appropriate |
| %% start time such that we don't have to "fudge" deltas with a |
| %% missing start point, but we're a long ways from that happening |
| %% so I feel it necessary to address the NULL start time. |
| |
| %% Track how often we fail to initiate T0 correctly |
| %% Perhaps somewhat naughty we're incrementing stats from within |
| %% couch_stats itself? Might need to handle this differently |
| %% TODO: determine appropriate course of action here |
| %% io:format("~n**********MISSING STARTING DELTA************~n~n", []), |
| couch_stats:increment_counter( |
| [couchdb, csrt, delta_missing_t0]), |
| %%[couch_stats_resource_tracker, delta_missing_t0]), |
| |
| case erlang:get(?DELTA_TZ) of |
| undefined -> |
| TA0 = make_delta_base(), |
| %% TODO: handline missing deltas, otherwise divide by zero |
| set_delta_a(TA0), |
| TA0; |
| TA0 -> |
| TA0 |
| end; |
| #rctx{} = TA0 -> |
| TA0 |
| end, |
| TB = get_resource(), |
| Delta = make_delta(TA, TB), |
| set_delta_a(TB), |
| Delta. |
| |
| |
| make_delta(#rctx{}=TA, #rctx{}=TB) -> |
| Delta = #{ |
| docs_read => TB#rctx.docs_read - TA#rctx.docs_read, |
| js_filter => TB#rctx.js_filter - TA#rctx.js_filter, |
| js_filter_error => TB#rctx.js_filter_error - TA#rctx.js_filter_error, |
| js_filtered_docs => TB#rctx.js_filtered_docs - TA#rctx.js_filtered_docs, |
| rows_read => TB#rctx.rows_read - TA#rctx.rows_read, |
| changes_returned => TB#rctx.changes_returned - TA#rctx.changes_returned, |
| btree_folds => TB#rctx.btree_folds - TA#rctx.btree_folds, |
| get_kp_node => TB#rctx.get_kp_node - TA#rctx.get_kp_node, |
| get_kv_node => TB#rctx.get_kv_node - TA#rctx.get_kv_node, |
| db_open => TB#rctx.db_open - TA#rctx.db_open, |
| ioq_calls => TB#rctx.ioq_calls - TA#rctx.ioq_calls, |
| dt => TB#rctx.updated_at - TA#rctx.updated_at |
| }, |
| %% TODO: reevaluate this decision |
| %% Only return non zero (and also positive) delta fields |
| maps:filter(fun(_K,V) -> V > 0 end, Delta); |
| make_delta(_, #rctx{}) -> |
| #{error => missing_beg_rctx}; |
| make_delta(#rctx{}, _) -> |
| #{error => missing_fin_rctx}. |
| |
| make_delta_base() -> |
| make_delta_base(get_pid_ref()). |
| |
| %% TODO: what to do when PidRef=undefined? |
| make_delta_base(PidRef) -> |
| %% TODO: extract user_ctx and db/shard from request |
| Now = tnow(), |
| #rctx{ |
| pid_ref = PidRef, |
| %% TODO: confirm this subtraction works |
| started_at = Now - 100, %% give us 100ms rewind time for missing T0 |
| updated_at = Now |
| }. |
| |
| set_delta_a(TA) -> |
| erlang:put(?DELTA_TA, TA). |
| |
| set_pid_ref(PidRef) -> |
| erlang:put(?PID_REF, PidRef), |
| PidRef. |
| |
| get_resource() -> |
| get_resource(get_pid_ref()). |
| |
| get_resource(undefined) -> |
| undefined; |
| get_resource(PidRef) -> |
| catch get_resource_int(PidRef). |
| |
| |
| get_resource_int(undefined) -> |
| undefined; |
| get_resource_int(PidRef) -> |
| case ets:lookup(?MODULE, PidRef) of |
| [#rctx{}=TP] -> |
| TP; |
| [] -> |
| undefined |
| end. |
| |
| |
| find_unmonitored() -> |
| %% TODO: only need PidRef here, replace with a select that does that... |
| [PR || #rctx{pid_ref=PR} <- ets:match_object(?MODULE, #rctx{mon_ref=undefined, _ = '_'})]. |
| |
| |
| find_by_pid(Pid) -> |
| [R || #rctx{} = R <- ets:match_object(?MODULE, #rctx{pid_ref={Pid, '_'}, _ = '_'})]. |
| |
| |
| pause_eviction() -> |
| gen_server:call(?MODULE, pause_eviction). |
| |
| |
| enable_eviction() -> |
| gen_server:call(?MODULE, enable_eviction). |
| |
| |
| eviction_status() -> |
| gen_server:call(?MODULE, enable_eviction). |
| |
| |
| start_link() -> |
| gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). |
| |
| init([]) -> |
| ets:new(?MODULE, [ |
| named_table, |
| public, |
| {decentralized_counters, true}, %% TODO: test impact of this |
| {write_concurrency, true}, |
| {read_concurrency, true}, |
| {keypos, #rctx.pid_ref} |
| ]), |
| St = #st{}, |
| case is_enabled() andalso should_scan() of |
| false -> |
| ok; |
| true -> |
| _TimerRef = erlang:send_after(St#st.scan_interval, self(), scan) |
| end, |
| {ok, St}. |
| |
| handle_call(eviction_status, _from, #st{eviction=Eviction, ev_queue=EVQ} = St) -> |
| {reply, {ok, Eviction, length(EVQ)}, St}; |
| handle_call(pause_eviction, _from, #st{eviction=Eviction} = St) -> |
| {reply, {ok, Eviction}, St#st{eviction=paused}}; |
| handle_call(enable_eviction, _from, #st{eviction=Eviction, ev_queue=EVQ} = St) -> |
| evict(EVQ), |
| {reply, {ok, Eviction}, St#st{eviction=enabled, ev_queue=[]}}; |
| handle_call(fetch, _from, #st{} = St) -> |
| {reply, {ok, St}, St}; |
| handle_call({track, _}=Msg, _From, St) -> |
| {noreply, St1} = handle_cast(Msg, St), |
| {reply, ok, St1}; |
| handle_call(Msg, _From, St) -> |
| {stop, {unknown_call, Msg}, St}. |
| |
| handle_cast({destroy, {_,_}=PidRef}, #st{tracking=AT0} = St0) -> |
| AT = destroy_context_int(PidRef, AT0, <<"shutdown:completed">>), |
| {noreply, St0#st{tracking=AT}}; |
| handle_cast({track, {_,_}=PidRef}, #st{tracking=AT0} = St0) -> |
| AT = maybe_track(PidRef, AT0), |
| {noreply, St0#st{tracking=AT}}; |
| handle_cast(Msg, St) -> |
| {stop, {unknown_cast, Msg}, St}. |
| |
| handle_info(scan, #st{tracking=AT0} = St0) -> |
| Unmonitored = find_unmonitored(), |
| AT = maybe_track(Unmonitored, AT0), |
| _TimerRef = erlang:send_after(St0#st.scan_interval, self(), scan), |
| {noreply, St0#st{tracking=AT}}; |
| handle_info({'DOWN', MonRef, _Type, _DPid, Reason0}, #st{tracking=AT0} = St) -> |
| Reason = case Reason0 of |
| {shutdown, Shutdown0} -> |
| Shutdown = atom_to_binary(Shutdown0), |
| <<"shutdown: ", Shutdown/binary>>; |
| Reason0 -> |
| Reason0 |
| end, |
| %% TODO: moving to destroy_context_int lost assertion against RPid =:= DPid |
| %% NOTE: we can't flush with demonitor, so we might get a DOWN message for |
| %% an already destroyed context. |
| %% if |
| %% RPid =:= DPid -> ok; |
| %% true -> erlang:halt(...) |
| %% end, |
| AT = destroy_context_int(MonRef, AT0, Reason), |
| {noreply, St#st{tracking=AT}}; |
| handle_info({evict, {_Pid, _Ref}=PidRef}, #st{eviction=paused, ev_queue=EVQ}=St) -> |
| {noreply, St#st{ev_queue=[PidRef|EVQ]}}; |
| handle_info({evict, {_Pid, _Ref}=PidRef}, #st{}=St) -> |
| evict(PidRef), |
| {noreply, St}; |
| handle_info(Msg, St) -> |
| {stop, {unknown_info, Msg}, St}. |
| |
| terminate(_Reason, _St) -> |
| ok. |
| |
| code_change(_OldVsn, St, _Extra) -> |
| {ok, St}. |
| |
| |
| destroy_context_int(MonRef, AT0, Reason) when is_reference(MonRef) -> |
| drop_monitor(MonRef), |
| case maps:get(MonRef, AT0, undefined) of |
| undefined -> |
| %% Due to not flushing in demonitor we can now get an extra DOWN msg |
| AT0; |
| {_Pid, _Ref} = PidRef -> |
| AT = maps:remove(MonRef, maps:remove(PidRef, AT0)), |
| %% TODO: Assert Pid matches Object from MonRef DOWN msg |
| destroy_context_int(PidRef, AT, Reason) |
| end; |
| destroy_context_int({_Pid, _Ref}=PidRef, AT0, Reason) -> |
| AT = case maps:get(PidRef, AT0, undefined) of |
| undefined -> |
| AT0; |
| MonRef -> |
| drop_monitor(MonRef), |
| maps:remove(MonRef, maps:remove(PidRef, AT0)) |
| end, |
| should_scan() andalso ets:update_element(?MODULE, PidRef, |
| [{#rctx.state, {down, Reason}}, {#rctx.updated_at, tnow()}]), |
| log_process_lifetime_report(PidRef), |
| %% Delay eviction to allow human visibility on short lived pids |
| %%erlang:send_after(St0#st.eviction_delay, self(), {evict, PidRef}), |
| erlang:send_after(0, self(), {evict, PidRef}), |
| AT. |
| |
| |
| drop_monitor(MonRef) -> |
| %% can't use [flush] here or we melt erts_internal:flush_monitor_messages/3 |
| demonitor(MonRef). |
| |
| |
| evict([]) -> |
| true; |
| evict([PidRef|Rest]) -> |
| evict(PidRef), |
| evict(Rest); |
| evict(PidRef) -> |
| ets:delete(?MODULE, PidRef). |
| |
| |
| maybe_track([], AT) -> |
| AT; |
| maybe_track(PidRef, AT) when is_tuple(PidRef) -> |
| maybe_track([PidRef], AT); |
| maybe_track([{Pid,_Ref} = PidRef | PidRefs], AT) -> |
| AT1 = case maps:is_key(PidRef, AT) of |
| true -> %% noop, we're already tracking this PidRef |
| AT; |
| false -> %% setup new monitor and double bookkeep refs |
| Mon = erlang:monitor(process, Pid), |
| %% TODO: decide whether we want the true match to crash this process on failure |
| %%true = ets:update_element(?MODULE, PidRef, [{#rctx.mon_ref, Mon}]), |
| should_scan() andalso ets:update_element(?MODULE, PidRef, [{#rctx.mon_ref, Mon}]), |
| maps:put(Mon, PidRef, maps:put(PidRef, Mon, AT)) |
| end, |
| maybe_track(PidRefs, AT1). |
| |
| |
| log_process_lifetime_report(PidRef) -> |
| %% TODO: catch error out of here, report crashes on depth>1 json |
| %%io:format("CSRT RCTX: ~p~n", [to_flat_json(Rctx)]), |
| %% TODO: clean this up |
| case is_enabled() andalso is_logging_enabled() of |
| true -> |
| Rctx = get_resource_int(PidRef), |
| should_log(Rctx) andalso log_process_lifetime_report(PidRef, Rctx); |
| false -> |
| ok |
| end. |
| |
| log_process_lifetime_report(_PidRef, Rctx) -> |
| couch_log:report("csrt-pid-usage-lifetime", to_flat_json(Rctx)). |
| |
| is_logging_enabled() -> |
| logging_enabled() =/= false. |
| |
| logging_enabled() -> |
| case conf_get("log_pid_usage_report", "coordinator") of |
| "coordinator" -> |
| coordinator; |
| "true" -> |
| true; |
| _ -> |
| false |
| end. |
| |
| |
| should_log(undefined) -> |
| false; |
| should_log(#rctx{}=Rctx) -> |
| should_log(Rctx, logging_enabled()). |
| |
| |
| should_log(undefined, _) -> |
| false; |
| should_log(#rctx{}, true) -> |
| true; |
| should_log(#rctx{}, false) -> |
| false; |
| should_log(#rctx{type = {coordinator, _, _}}, coordinator) -> |
| true; |
| should_log(#rctx{type = {worker, fabric_rpc, FName}}, _) -> |
| case conf_get("log_fabric_rpc") of |
| "true" -> |
| true; |
| undefined -> |
| false; |
| Name -> |
| Name =:= atom_to_list(FName) |
| end; |
| should_log(#rctx{}, _) -> |
| false. |
| |
| |
| conf_get(Key) -> |
| conf_get(Key, undefined). |
| |
| |
| conf_get(Key, Default) -> |
| config:get(?MODULE_STRING, Key, Default). |
| |
| |
| %% Reimplementation of: https://github.com/erlang/otp/blob/b2ee4fc9a0b81a139dad2033e9b2bfc178146886/lib/stdlib/src/ets.erl#L633-L658 |
| %% with wrapping of ets:safe_fixtable/2 removed |
| unsafe_foldl(F, Accu, T) -> |
| First = ets:first(T), |
| do_foldl(F, Accu, First, T). |
| |
| do_foldl(F, Accu0, Key, T) -> |
| case Key of |
| '$end_of_table' -> |
| Accu0; |
| _ -> |
| do_foldl(F, |
| lists:foldl(F, Accu0, ets:lookup(T, Key)), |
| ets:next(T, Key), T) |
| end. |
