Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Implement naive clamping of the chunk cache size for mining worker #660

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Implement naive clamping of the chunk cache size for mining worker #660

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
516ad72
Implement naive clamping of the chunk cache size for mining worker
shizzard Dec 2, 2024
44d55ed
Revert the metric name
shizzard Dec 2, 2024
675d7a9
Fix diff_pair typespec
shizzard Dec 2, 2024
2389f3f
Add a comment describing the temporary solution
shizzard Dec 2, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 32 additions & 32 deletions apps/arweave/src/ar_mining_worker.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,24 +13,24 @@
-include_lib("eunit/include/eunit.hrl").

-record(state, {
name = not_set,
partition_number = not_set,
diff_pair = not_set,
packing_difficulty = 0,
task_queue = gb_sets:new(),
active_sessions = sets:new(),
%% The sub_chunk_cache stores either the first or second sub-chunk for a given nonce. This
%% is because we process both the first and second recall ranges in parallel and don't know
%% which data will be available first. For spora_2_6 packing (aka difficulty 0), sub-chunks
%% and chunks are the same size (256KiB), for composite packing each sub-chunk is 8KiB.
sub_chunk_cache = #{},
sub_chunk_cache_size = #{},
sub_chunk_cache_limit = 0,
vdf_queue_limit = 0,
latest_vdf_step_number = 0,
is_pool_client = false,
h1_hashes = #{},
h2_hashes = #{}
name = not_set :: atom(),
partition_number = not_set :: non_neg_integer() | not_set,
diff_pair = not_set :: {non_neg_integer(), non_neg_integer()} | not_set,
packing_difficulty = 0 :: non_neg_integer(),
task_queue = gb_sets:new() :: gb_sets:gb_set(),
active_sessions = sets:new() :: sets:set(),
%% The sub_chunk_cache stores either the first or second sub-chunk for a given nonce. This
%% is because we process both the first and second recall ranges in parallel and don't know
%% which data will be available first. For spora_2_6 packing (aka difficulty 0), sub-chunks
%% and chunks are the same size (256KiB), for composite packing each sub-chunk is 8KiB.
sub_chunk_cache = #{} :: maps:map(),
sub_chunk_cache_size = #{} :: maps:map(),
sub_chunk_cache_limit = 0 :: non_neg_integer(),
vdf_queue_limit = 0 :: non_neg_integer(),
latest_vdf_step_number = 0 :: non_neg_integer(),
is_pool_client = false,
h1_hashes = #{} :: maps:map(),
h2_hashes = #{} :: maps:map()
}).

-define(TASK_CHECK_FREQUENCY_MS, 200).
Expand Down Expand Up @@ -63,7 +63,7 @@ add_task(Worker, TaskType, Candidate, ExtraArgs) ->

add_delayed_task(Worker, TaskType, Candidate) ->
%% Delay task by random amount between ?TASK_CHECK_FREQUENCY_MS and 2*?TASK_CHECK_FREQUENCY_MS
%% The reason for the randomization to avoid a glut tasks to all get added at the same time -
%% The reason for the randomization to avoid a glut tasks to all get added at the same time -
%% in particular when the chunk cache fills up it's possible for all queued compute_h0 tasks
%% to be delayed at about the same time.
Delay = rand:uniform(?TASK_CHECK_FREQUENCY_MS) + ?TASK_CHECK_FREQUENCY_MS,
Expand Down Expand Up @@ -146,7 +146,7 @@ handle_cast({chunks_read, {WhichChunk, Candidate, RangeStart, ChunkOffsets}}, St
{worker, State#state.name},
{active_sessions,
ar_mining_server:encode_sessions(State#state.active_sessions)},
{candidate_session,
{candidate_session,
ar_nonce_limiter:encode_session_key(Candidate#mining_candidate.session_key)},
{partition_number, Candidate#mining_candidate.partition_number},
{step_number, Candidate#mining_candidate.step_number}]),
Expand All @@ -163,14 +163,14 @@ handle_cast({add_task, {TaskType, Candidate, _ExtraArgs} = Task}, State) ->
{task, TaskType},
{active_sessions,
ar_mining_server:encode_sessions(State#state.active_sessions)},
{candidate_session,
{candidate_session,
ar_nonce_limiter:encode_session_key(Candidate#mining_candidate.session_key)},
{partition_number, Candidate#mining_candidate.partition_number},
{step_number, Candidate#mining_candidate.step_number},
{nonce, Candidate#mining_candidate.nonce}]),
{noreply, State}
end;

handle_cast(handle_task, #state{ task_queue = Q } = State) ->
case gb_sets:is_empty(Q) of
true ->
Expand Down Expand Up @@ -305,7 +305,7 @@ process_chunks(WhichChunk, Candidate, RangeStart, Nonce, NoncesPerChunk, NonceMa
ChunkOffsets, SubChunkSize, Count + 1, State2).

process_all_sub_chunks(_WhichChunk, <<>>, _Candidate, _Nonce, State) ->
State;
State;
process_all_sub_chunks(WhichChunk, Chunk, Candidate, Nonce, State) ->
{SubChunk, Rest} = extract_sub_chunk(Chunk, Candidate),
Candidate2 = Candidate#mining_candidate{ nonce = Nonce },
Expand Down Expand Up @@ -344,7 +344,7 @@ process_sub_chunk(chunk2, Candidate, SubChunk, State) ->
{{chunk1, H1}, State2} ->
ar_mining_hash:compute_h2(self(), Candidate2#mining_candidate{ h1 = H1 }),
%% Decrement 1 for chunk2:
%% we're computing h2 for a peer so chunk1 was not previously read or cached
%% we're computing h2 for a peer so chunk1 was not previously read or cached
%% on this node
update_sub_chunk_cache_size(-1, SessionKey, State2);
{do_not_cache, State2} ->
Expand Down Expand Up @@ -433,7 +433,7 @@ handle_task({computed_h0, Candidate, _ExtraArgs}, State) ->
Range2Exists = ar_mining_io:read_recall_range(
chunk2, self(), Candidate3, RecallRange2Start),
case Range2Exists of
true ->
true ->
State;
false ->
%% Release just the Range2 cache space we reserved with
Expand Down Expand Up @@ -628,7 +628,7 @@ maybe_warn_about_lag(Q, Name) ->
%% Since we sample the queue asynchronously, we expect there to regularly
%% be a queue of length 1 (i.e. a task may have just been added to the
%% queue when we run this check).
%%
%%
%% To further reduce log spam, we'll only warn if the queue is greater
%% than 2. We really only care if a queue is consistently long or if
%% it's getting longer. Temporary blips are fine. We may incrase
Expand Down Expand Up @@ -662,7 +662,7 @@ count_h0_tasks(Q) ->
end.

maybe_warn_about_stale_chunks(State) ->
TotalChunkKeys =
TotalChunkKeys =
maps:fold(
fun(_SesssionKey, SessionCache, Acc) ->
Acc + maps:size(SessionCache)
Expand Down Expand Up @@ -752,11 +752,11 @@ update_sub_chunk_cache_size(0, _SessionKey, State) ->
State;
update_sub_chunk_cache_size(Delta, SessionKey, State) ->
CacheSize = maps:get(SessionKey, State#state.sub_chunk_cache_size, 0),
prometheus_gauge:inc(mining_server_chunk_cache_size,
[State#state.partition_number], Delta),
State#state{
sub_chunk_cache_size = maps:put(SessionKey, CacheSize + Delta,
State#state.sub_chunk_cache_size) }.
% The clamp_lower function ensures that the cache size does not go below 0.
% This is temporary until we have a more permanent solution for the chunk cache.
NewCacheSize = ar_util:clamp_lower(CacheSize + Delta, 0),
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The key change: do not let the sub_chunk_cache_size go below zero, as this does not make any sense.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Aren't we fighting the symptom? Perhaps, let's leave a comment this is a temporary measure until we figure out why we subtract more than we add, if that is the case.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Perhaps, yes, you're right; proper fix will require the deep (sub)chunk cache refactoring, which is out of scope for this task.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@shizzard what about adding a log_warning message whenever CacheSize+Delta is below zero? We do want to fix the OOMs as quick as possible and clamping may help there, but I'm a little worried we may also lose this valuable signal that something is wrong. Which may make tracking down the true root cause harder.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I slacked with some more information that might help with reproducing the negative cache size (and so might help get at the root cause)

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what about adding a log_warning message

Thats a very good idea, will do.

prometheus_gauge:inc(mining_server_chunk_cache_size, [State#state.partition_number], NewCacheSize - CacheSize),
State#state{ sub_chunk_cache_size = maps:put(SessionKey, NewCacheSize, State#state.sub_chunk_cache_size) }.

try_to_reserve_cache_space(SessionKey, State) ->
#state{ packing_difficulty = PackingDifficulty } = State,
Expand Down
22 changes: 21 additions & 1 deletion apps/arweave/src/ar_util.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
integer_to_binary/1, binary_to_integer/1, pick_random/1, pick_random/2,
encode/1, decode/1, safe_encode/1, safe_decode/1, timestamp_to_seconds/1,
parse_peer/1, peer_to_str/1, parse_port/1, safe_parse_peer/1, format_peer/1,
unique/1, count/2,
unique/1, count/2, clamp_lower/2, clamp_upper/2, clamp/3,
genesis_wallets/0, pmap/2, pfilter/2,
do_until/3, block_index_entry_from_block/1,
bytes_to_mb_string/1, cast_after/3, encode_list_indices/1, parse_list_indices/1,
Expand Down Expand Up @@ -381,3 +381,23 @@ assert_file_exists_and_readable(FilePath) ->
io:format("~nThe filepath ~p doesn't exist or isn't readable.~n~n", [FilePath]),
erlang:halt(1)
end.

%% @doc Clamp a value to a lower bound.
%% If the value is less than the lower bound, return the lower bound.
clamp_lower(Value, LowerBound) when Value < LowerBound ->
LowerBound;
clamp_lower(Value, _LowerBound) ->
Value.

%% @doc Clamp a value to an upper bound.
%% If the value is greater than the upper bound, return the upper bound.
clamp_upper(Value, UpperBound) when Value > UpperBound ->
UpperBound;
clamp_upper(Value, _UpperBound) ->
Value.

%% @doc Clamp a value to a lower and upper bound.
%% If the value is less than the lower bound, return the lower bound.
%% If the value is greater than the upper bound, return the upper bound.
clamp(Value, LowerBound, UpperBound) ->
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

FYI: ar_util:between/3 also does this: https://github.com/ArweaveTeam/arweave/pull/660/files#diff-1f26621f4ea9ef47e071afce0f4ce9fc3e6bacc76089ac9037c00e0c07a5ce9eR46

Can you add a comment to ar_util:between that people should start using ar_util:clamp instead?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, didn't see the existing one, okay.

clamp_lower(clamp_upper(Value, UpperBound), LowerBound).
Loading