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 configurable requesting of packed chunks from peers #633

Merged
merged 2 commits into from
Oct 22, 2024
Merged

Implement configurable requesting of packed chunks from peers #633

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
34e923f
Implement configurable requesting packed chunks from peers
shizzard Oct 20, 2024
b5f4eed
Add flag to the state to avoid polling config
shizzard Oct 22, 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
1 change: 1 addition & 0 deletions apps/arweave/include/ar_config.hrl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,7 @@
tx_propagation_parallelization, % DEPRECATED.
sync_jobs = ?DEFAULT_SYNC_JOBS,
header_sync_jobs = ?DEFAULT_HEADER_SYNC_JOBS,
data_sync_request_packed_chunks = false,
disk_pool_jobs = ?DEFAULT_DISK_POOL_JOBS,
load_key = not_set,
disk_space,
Expand Down
4 changes: 4 additions & 0 deletions apps/arweave/src/ar.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,8 @@ show_help() ->
" and downloads it from peers.",
[?DEFAULT_HEADER_SYNC_JOBS]
)},
{"data_sync_request_packed_chunks",
"Enables requesting the packed chunks from peers."},
{"disk_pool_jobs (num)",
io_lib:format(
"The number of disk pool jobs to run. Default: ~B."
Expand Down Expand Up @@ -520,6 +522,8 @@ parse_cli_args(["sync_jobs", Num | Rest], C) ->
parse_cli_args(Rest, C#config{ sync_jobs = list_to_integer(Num) });
parse_cli_args(["header_sync_jobs", Num | Rest], C) ->
parse_cli_args(Rest, C#config{ header_sync_jobs = list_to_integer(Num) });
parse_cli_args(["data_sync_request_packed_chunks" | Rest], C) ->
parse_cli_args(Rest, C#config{ data_sync_request_packed_chunks = true });
parse_cli_args(["tx_validators", Num | Rest], C) ->
parse_cli_args(Rest, C#config{ tx_validators = list_to_integer(Num) });
parse_cli_args(["post_tx_timeout", Num | Rest], C) ->
Expand Down
7 changes: 6 additions & 1 deletion apps/arweave/src/ar_config.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -656,6 +656,11 @@ parse_options([{<<"rocksdb_wal_sync_interval">>, IntervalS} | Rest], Config)
parse_options([{<<"rocksdb_wal_sync_interval">>, IntervalS} | _], _) ->
{error, {bad_type, rocksdb_wal_sync_interval, number}, IntervalS};

parse_options([{<<"data_sync_request_packed_chunks">>, Bool} | Rest], Config) when is_boolean(Bool) ->
parse_options(Rest, Config#config{ data_sync_request_packed_chunks = Bool });
parse_options([{<<"data_sync_request_packed_chunks">>, InvalidValue} | Rest], Config) ->
{error, {bad_type, data_sync_request_packed_chunks, boolean}, InvalidValue};

parse_options([Opt | _], _) ->
{error, unknown, Opt};
parse_options([], Config) ->
Expand Down Expand Up @@ -937,4 +942,4 @@ validate_storage_modules(Config) ->
io:format("~nThe node cannot mine multiple packing difficulties "
"for the same mining address.~n~n"),
false
end.
end.
26 changes: 17 additions & 9 deletions apps/arweave/src/ar_data_sync_worker.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,8 @@
-include_lib("arweave/include/ar_data_sync.hrl").

-record(state, {
name = undefined
name = undefined,
request_packed_chunks = false
}).

%% # of messages to cast to ar_data_sync at once. Each message carries at least 1 chunk worth
Expand All @@ -36,7 +37,11 @@ start_link(Name) ->
%%%===================================================================

init(Name) ->
{ok, #state{ name = Name }}.
{ok, Config} = application:get_env(arweave, config),
{ok, #state{
name = Name,
request_packed_chunks = Config#config.data_sync_request_packed_chunks
}}.

handle_call(Request, _From, State) ->
?LOG_WARNING([{event, unhandled_call}, {module, ?MODULE}, {request, Request}]),
Expand All @@ -54,7 +59,7 @@ handle_cast({read_range, Args}, State) ->

handle_cast({sync_range, Args}, State) ->
StartTime = erlang:monotonic_time(),
SyncResult = sync_range(Args),
SyncResult = sync_range(Args, State),
EndTime = erlang:monotonic_time(),
case SyncResult of
recast ->
Expand Down Expand Up @@ -150,7 +155,7 @@ read_range2(MessagesRemaining, {Start, End, OriginStoreID, TargetStoreID, SkipSm
{absolute_end_offset, AbsoluteOffset},
{chunk_data_key, ar_util:encode(ChunkDataKey)},
{reason, io_lib:format("~p", [Error])}]),
read_range2(MessagesRemaining,
read_range2(MessagesRemaining,
{Start + ChunkSize, End, OriginStoreID, TargetStoreID, SkipSmall});
{ok, {Chunk, DataPath}} ->
case ar_sync_record:is_recorded(AbsoluteOffset, ar_data_sync,
Expand Down Expand Up @@ -200,14 +205,14 @@ read_range2(MessagesRemaining, {Start, End, OriginStoreID, TargetStoreID, SkipSm
{Start + ChunkSize, End, OriginStoreID, TargetStoreID, SkipSmall})
end.

sync_range({Start, End, _Peer, _TargetStoreID, _RetryCount}) when Start >= End ->
sync_range({Start, End, _Peer, _TargetStoreID, _RetryCount}, _State) when Start >= End ->
ok;
sync_range({Start, End, Peer, _TargetStoreID, 0}) ->
sync_range({Start, End, Peer, _TargetStoreID, 0}, _State) ->
?LOG_DEBUG([{event, sync_range_retries_exhausted},
{peer, ar_util:format_peer(Peer)},
{start_offset, Start}, {end_offset, End}]),
{error, timeout};
sync_range({Start, End, Peer, TargetStoreID, RetryCount} = Args) ->
sync_range({Start, End, Peer, TargetStoreID, RetryCount} = Args, State) ->
IsChunkCacheFull =
case ar_data_sync:is_chunk_cache_full() of
true ->
Expand Down Expand Up @@ -238,7 +243,7 @@ sync_range({Start, End, Peer, TargetStoreID, RetryCount} = Args) ->
true ->
ok;
false ->
case ar_http_iface_client:get_chunk_binary(Peer, Start2, any) of
case ar_http_iface_client:get_chunk_binary(Peer, Start2, get_target_packing(TargetStoreID, State#state.request_packed_chunks)) of
{ok, #{ chunk := Chunk } = Proof, _Time, _TransferSize} ->
%% In case we fetched a packed small chunk,
%% we may potentially skip some chunks by
Expand All @@ -250,7 +255,7 @@ sync_range({Start, End, Peer, TargetStoreID, RetryCount} = Args) ->
gen_server:cast(list_to_atom("ar_data_sync_" ++ Label),
{store_fetched_chunk, Peer, Start2 - 1, Proof}),
ar_data_sync:increment_chunk_cache_size(),
sync_range({Start3, End, Peer, TargetStoreID, RetryCount});
sync_range({Start3, End, Peer, TargetStoreID, RetryCount}, State);
{error, timeout} ->
?LOG_DEBUG([{event, timeout_fetching_chunk},
{peer, ar_util:format_peer(Peer)},
Expand All @@ -267,3 +272,6 @@ sync_range({Start, End, Peer, TargetStoreID, RetryCount} = Args) ->
end
end
end.

get_target_packing(StoreID, true) -> ar_storage_module:get_packing(StoreID);
get_target_packing(_StoreID, false) -> any.
Loading