Feature/cross module repacking optimization

apps/arweave/include/ar_data_sync.hrl

@@ -65,22 +65,6 @@
 -define(EXCLUDE_MISSING_INTERVAL_TIMEOUT_MS, 10 * 60 * 1000).
 -endif.
 
-%% Let at least this many chunks stack up, per storage module, then write them on disk in the
-%% ascending order, to reduce out-of-order disk writes causing fragmentation.
--ifdef(DEBUG).
--define(STORE_CHUNK_QUEUE_FLUSH_SIZE_THRESHOLD, 2).
--else.
--define(STORE_CHUNK_QUEUE_FLUSH_SIZE_THRESHOLD, 100). % ~ 25 MB worth of chunks.
--endif.
-
-%% If a chunk spends longer than this in the store queue, write it on disk without waiting
-%% for ?STORE_CHUNK_QUEUE_FLUSH_SIZE_THRESHOLD chunks to stack up.
--ifdef(DEBUG).
--define(STORE_CHUNK_QUEUE_FLUSH_TIME_THRESHOLD, 1000).
--else.
--define(STORE_CHUNK_QUEUE_FLUSH_TIME_THRESHOLD, 2_000). % 2 seconds.
--endif.
-
 %% @doc The state of the server managing data synchronization.
 -record(sync_data_state, {
 	%% The last entries of the block index.
@@ -200,16 +184,6 @@
 	%% storage module to be searched for missing data before attempting to sync the data
 	%% from the network.
 	other_storage_modules_with_unsynced_intervals = [],
-	%% The priority queue of chunks sorted by offset. The motivation is to have chunks
-	%% stack up, per storage module, before writing them on disk so that we can write
-	%% them in the ascending order and reduce out-of-order disk writes causing fragmentation.
-	store_chunk_queue = gb_sets:new(),
-	%% The length of the store chunk queue.
-	store_chunk_queue_len = 0,
-	%% The threshold controlling the brief accumuluation of the chunks in the queue before
-	%% the actual disk dump, to reduce the chance of out-of-order write causing disk
-	%% fragmentation.
-	store_chunk_queue_threshold = ?STORE_CHUNK_QUEUE_FLUSH_SIZE_THRESHOLD,
 	%% Cache mapping peers to /data_sync_record responses
 	all_peers_intervals = #{}
 }).

apps/arweave/src/ar.erl

@@ -756,12 +756,19 @@ start(normal, _Args) ->
 	ar_sup:start_link().
 
 set_mining_address(#config{ mining_addr = not_set } = C) ->
-	W = ar_wallet:get_or_create_wallet([{?RSA_SIGN_ALG, 65537}]),
-	Addr = ar_wallet:to_address(W),
-	ar:console("~nSetting the mining address to ~s.~n", [ar_util:encode(Addr)]),
-	C2 = C#config{ mining_addr = Addr },
-	application:set_env(arweave, config, C2),
-	set_mining_address(C2);
+	case ar_wallet:get_or_create_wallet([{?RSA_SIGN_ALG, 65537}]) of
+		{error, Reason} ->
+			ar:console("~nFailed to create a wallet, reason: ~p.~n",
+				[io_lib:format("~p", [Reason])]),
+			timer:sleep(500),
+			erlang:halt();
+		W ->
+			Addr = ar_wallet:to_address(W),
+			ar:console("~nSetting the mining address to ~s.~n", [ar_util:encode(Addr)]),
+			C2 = C#config{ mining_addr = Addr },
+			application:set_env(arweave, config, C2),
+			set_mining_address(C2)
+	end;
 set_mining_address(#config{ mine = false }) ->
 	ok;
 set_mining_address(#config{ mining_addr = Addr, cm_exit_peer = CmExitPeer,
@@ -790,10 +797,17 @@ create_wallet([DataDir]) ->
 			create_wallet_fail();
 		true ->
 			ok = application:set_env(arweave, config, #config{ data_dir = DataDir }),
-			W = ar_wallet:new_keyfile({?RSA_SIGN_ALG, 65537}),
-			Addr = ar_wallet:to_address(W),
-			ar:console("Created a wallet with address ~s.~n", [ar_util:encode(Addr)]),
-			erlang:halt()
+			case ar_wallet:new_keyfile({?RSA_SIGN_ALG, 65537}) of
+				{error, Reason} ->
+					ar:console("Failed to create a wallet, reason: ~p.~n~n",
+							[io_lib:format("~p", [Reason])]),
+					timer:sleep(500),
+					erlang:halt();
+				W ->
+					Addr = ar_wallet:to_address(W),
+					ar:console("Created a wallet with address ~s.~n", [ar_util:encode(Addr)]),
+					erlang:halt()
+			end
 	end;
 create_wallet(_) ->
 	create_wallet_fail().

apps/arweave/src/ar_block_pre_validator.erl

@@ -458,6 +458,13 @@ get_last_step_prev_output(B) ->
 	end.
 
 validate_poa_against_cached_poa(B, CacheB) ->
+	case CacheB#block.poa_cache of
+		{_, _} ->
+			ok;
+		_ ->
+			ID = binary_to_list(ar_util:encode(crypto:strong_rand_bytes(2))),
+			file:write_file("/opt/arweave/bad-block-" ++ ID, term_to_binary(B))
+	end,
 	#block{ poa_cache = {ArgCache, ChunkID}, poa2_cache = Cache2 } = CacheB,
 	Args = erlang:append_element(erlang:insert_element(5, ArgCache, B#block.poa), ChunkID),
 	case ar_poa:validate(Args) of

apps/arweave/src/ar_chunk_storage.erl

@@ -291,6 +291,13 @@ handle_cast({register_packing_ref, Ref, Offset}, #state{ packing_map = Map } = S
 	{noreply, State#state{ packing_map = maps:put(Ref, Offset, Map) }};
 
 handle_cast({expire_repack_request, Ref}, #state{ packing_map = Map } = State) ->
+	case maps:is_key(Ref, Map) of
+		true ->
+			?LOG_WARNING([{event, repack_request_expired},
+					{type, repack_in_place}]);
+		false ->
+			ok
+	end,
 	{noreply, State#state{ packing_map = maps:remove(Ref, Map) }};
 
 handle_cast(Cast, State) ->
@@ -358,6 +365,7 @@ handle_info({chunk, {packed, Ref, ChunkArgs}},
 									repack_cursor = Offset, prev_repack_cursor = PrevCursor }};
 						Error2 ->
 							?LOG_ERROR([{event, failed_to_store_repacked_chunk},
+									{type, repack_in_place},
 									{storage_module, StoreID},
 									{offset, Offset},
 									{packing, ar_serialize:encode_packing(Packing, true)},
@@ -366,6 +374,7 @@ handle_info({chunk, {packed, Ref, ChunkArgs}},
 					end;
 				Error3 ->
 					?LOG_ERROR([{event, failed_to_remove_repacked_chunk_from_sync_record},
+							{type, repack_in_place},
 							{storage_module, StoreID},
 							{offset, Offset},
 							{packing, ar_serialize:encode_packing(Packing, true)},
@@ -806,16 +815,23 @@ repack(Start, End, NextCursor, RightBound, RequiredPacking, StoreID) ->
 			ok ->
 				case catch get_range(Start, RepackIntervalSize, StoreID) of
 					[] ->
+						?LOG_WARNING([{event, failed_to_read_chunk_interval},
+								{type, repack_in_place},
+								{start, start},
+								{size, RepackIntervalSize},
+								{storage_module, StoreID}]),
 						Start2 = Start + RepackIntervalSize,
 						gen_server:cast(Server, {repack, Start2, End, NextCursor, RightBound,
 								RequiredPacking}),
 						continue;
-					{'EXIT', _Exc} ->
+					{'EXIT', Exc} ->
 						?LOG_ERROR([{event, failed_to_read_chunk_range},
+								{type, repack_in_place},
+								{error, io_lib:format("~p", [Exc])},
 								{storage_module, StoreID},
 								{start, Start},
 								{size, RepackIntervalSize},
-								{store_id, StoreID}]),
+								{storage_module, StoreID}]),
 						Start2 = Start + RepackIntervalSize,
 						gen_server:cast(Server, {repack, Start2, End, NextCursor, RightBound,
 								RequiredPacking}),
@@ -835,6 +851,7 @@ repack(Start, End, NextCursor, RightBound, RequiredPacking, StoreID) ->
 						{ok, Map, MetadataMap};
 					{error, Error} ->
 						?LOG_ERROR([{event, failed_to_read_chunk_metadata_range},
+								{type, repack_in_place},
 								{storage_module, StoreID},
 								{error, io_lib:format("~p", [Error])},
 								{left, Min},
@@ -856,13 +873,19 @@ repack(Start, End, NextCursor, RightBound, RequiredPacking, StoreID) ->
 				fun	(AbsoluteOffset, {_, _TXRoot, _, _, _, ChunkSize}, ok)
 							when ChunkSize /= ?DATA_CHUNK_SIZE,
 									AbsoluteOffset =< ?STRICT_DATA_SPLIT_THRESHOLD ->
+						?LOG_WARNING([{event, skipping_small_chunk},
+								{type, repack_in_place},
+								{chunk_size, ChunkSize},
+								{offset, AbsoluteOffset},
+								{storage_module, StoreID}]),
 						ok;
 					(AbsoluteOffset, {_, TXRoot, _, _, _, ChunkSize}, ok) ->
 						PaddedOffset = ar_data_sync:get_chunk_padded_offset(AbsoluteOffset),
 						case ar_sync_record:is_recorded(PaddedOffset, ar_data_sync, StoreID) of
 							{true, RequiredPacking} ->
 								?LOG_WARNING([{event,
 											repacking_process_chunk_already_repacked},
+										{type, repack_in_place},
 										{storage_module, StoreID},
 										{packing,
 											ar_serialize:encode_packing(RequiredPacking,true)},
@@ -873,8 +896,9 @@ repack(Start, End, NextCursor, RightBound, RequiredPacking, StoreID) ->
 									not_found ->
 										?LOG_WARNING([{event,
 												chunk_not_found_in_chunk_storage},
+											{type, repack_in_place},
 											{storage_module, StoreID},
-											{offset, PaddedOffset}]),
+											{padded_offset, PaddedOffset}]),
 										ok;
 									Chunk ->
 										Ref = make_ref(),
@@ -889,11 +913,13 @@ repack(Start, End, NextCursor, RightBound, RequiredPacking, StoreID) ->
 								end;
 							true ->
 								?LOG_WARNING([{event, no_packing_information_for_the_chunk},
+										{type, repack_in_place},
 										{storage_module, StoreID},
 										{offset, PaddedOffset}]),
 								ok;
 							false ->
 								?LOG_WARNING([{event, chunk_not_found_in_sync_record},
+										{type, repack_in_place},
 										{storage_module, StoreID},
 										{offset, PaddedOffset}]),
 								ok