more mining stats

.gitignore

@@ -3,6 +3,9 @@
 *.dat
 .vscode
 slave.out
+cm_miner_1.out
+cm_miner_2.out
+cm_miner_3.out
 *.code-workspace
 .arweave.plt
 testlog
@@ -15,6 +18,9 @@ apps/arweave/c_src/**/*.o
 apps/arweave/c_src/tests/tests
 data_test_master
 data_test_slave
+data_test_cm_miner_1
+data_test_cm_miner_2
+data_test_cm_miner_3
 erl_crash.dump
 tags
 metrics
@@ -28,4 +34,7 @@ logs
 ebin
 metrics_master
 metrics_slave
+metrics_cm_miner_1
+metrics_cm_miner_2
+metrics_cm_miner_3
 release/output

apps/arweave/include/ar_config.hrl

@@ -107,6 +107,9 @@
 -define(DEFAULT_BLOCK_THROTTLE_BY_SOLUTION_INTERVAL_MS, 2000).
 -endif.
 
+-define(DEFAULT_CM_POLL_INTERVAL, 60000).
+-define(DEFAULT_CM_STAT_INTERVAL, 60000).
+
 %% @doc Startup options with default values.
 -record(config, {
 	init = false,
@@ -192,7 +195,13 @@
 	defragmentation_modules = [],
 	block_throttle_by_ip_interval = ?DEFAULT_BLOCK_THROTTLE_BY_IP_INTERVAL_MS,
 	block_throttle_by_solution_interval = ?DEFAULT_BLOCK_THROTTLE_BY_SOLUTION_INTERVAL_MS,
-	p3 = #p3_config{}
+	p3 = #p3_config{},
+	coordinated_mining = false,
+	coordinated_mining_secret = not_set,
+	cm_exit_peer = not_set,
+	cm_peers = [],
+	cm_poll_interval = ?DEFAULT_CM_POLL_INTERVAL,
+	cm_stat_interval = ?DEFAULT_CM_STAT_INTERVAL
 }).
 
 -endif.

apps/arweave/include/ar_consensus.hrl

@@ -16,11 +16,17 @@
 %% of equal size. A miner can search for a solution in each of the partitions
 %% in parallel, per mining address.
 -ifdef(DEBUG).
--define(PARTITION_SIZE, 1800000).
+-define(PARTITION_SIZE, 2097152). % 8 * 256 * 1024
 -else.
 -define(PARTITION_SIZE, 3600000000000). % 90% of 4 TB.
 -endif.
 
+-define(PARTITION_NUMBER(Offset), (Offset div ?PARTITION_SIZE)).
+%% MAX_PARTITION_NUMBER excludes the last partition as it may be incomplete and therefore provides
+%% a mining advantage (e.g. it can fit in RAM)
+-define(MAX_PARTITION_NUMBER(PartitionUpperBound),
+	max(0, PartitionUpperBound div ?PARTITION_SIZE - 1)).
+
 %% The size of a recall range. The first range is randomly chosen from the given
 %% mining partition. The second range is chosen from the entire weave.
 -ifdef(DEBUG).

apps/arweave/include/ar_data_sync.hrl

@@ -27,7 +27,7 @@
 
 %% The time to wait until the next full disk pool scan.
 -ifdef(DEBUG).
--define(DISK_POOL_SCAN_DELAY_MS, 1000).
+-define(DISK_POOL_SCAN_DELAY_MS, 2000).
 -else.
 -define(DISK_POOL_SCAN_DELAY_MS, 10000).
 -endif.

apps/arweave/include/ar_mining.hrl

@@ -0,0 +1,56 @@
+-ifndef(AR_MINING_HRL).
+-define(AR_MINING_HRL, true).
+
+%% fields prefixed with cm_ are only set when a solution is distributed across miners as part
+%% of a coordinated mining set.
+-record(mining_candidate, {
+	cache_ref = not_set, %% not serialized
+	chunk1 = not_set, %% not serialized
+	chunk2 = not_set, %% not serialized
+	chunk1_store_id = not_set,
+	chunk2_store_id = not_set,
+	cm_diff = not_set, %% serialized. set to the difficulty used by the H1 miner
+	cm_lead_peer = not_set, %% not serialized. if set, this candidate came from another peer
+	h0 = not_set, %% serialized
+	h1 = not_set, %% serialized
+	h2 = not_set, %% serialized
+	mining_address = not_set, %% serialized
+	next_seed = not_set, %% serialized
+	nonce = not_set, %% serialized
+	nonce_limiter_output = not_set, %% serialized
+	partition_number = not_set, %% serialized
+	partition_number2 = not_set, %% serialized
+	partition_upper_bound = not_set, %% serialized
+	poa2 = not_set, %% serialized
+	preimage = not_set, %% serialized. this can be either the h1 or h2 preimage
+	seed = not_set, %% serialized
+	session_ref = not_set, %% not serialized
+	start_interval_number = not_set, %% serialized
+	step_number = not_set %% serialized
+}).
+
+-record(mining_solution, {
+	last_step_checkpoints = not_set,
+	merkle_rebase_threshold = not_set,
+	next_seed = not_set,
+	nonce = not_set,
+	nonce_limiter_output = not_set,
+	partition_number = not_set,
+	poa1 = not_set,
+	poa2 = not_set,
+	preimage = not_set,
+	recall_byte1 = undefined, %% undefined instead of not_set for compatibility with existing code
+	recall_byte2 = undefined,
+	solution_hash = not_set,
+	start_interval_number = not_set,
+	step_number = not_set,
+	steps = not_set,
+	%% Not used in block, but cached to improve validation and logging:
+	seed = not_set,
+	mining_address = not_set,
+	partition_upper_bound = not_set
+}).
+
+-define(MINING_HASH_MAX_BUCKET, 256).
+
+-endif.

apps/arweave/src/ar.erl

@@ -46,7 +46,6 @@
 		ar_poa,
 		ar_pricing,
 		ar_retarget,
-		ar_serialize,
 		ar_storage,
 		ar_sync_buckets,
 		ar_tx,
@@ -58,6 +57,7 @@
 		%% test modules
 		ar_base64_compatibility_tests,
 		ar_config_tests,
+		ar_coordinated_mining_tests,
 		ar_data_sync_tests,
 		ar_difficulty_tests,
 		ar_fork_recovery_tests,
@@ -68,13 +68,16 @@
 		ar_mempool_tests,
 		ar_mine_randomx_tests,
 		ar_mine_vdf_tests,
+		ar_mining_io_tests,
+		ar_mining_server_tests,
 		ar_multiple_txs_per_wallet_tests,
 		ar_node_tests,
 		ar_poa_tests,
 		ar_poller_tests,
 		ar_post_block_tests,
 		ar_pricing_tests,
 		ar_semaphore_tests,
+		ar_serialize_tests,
 		ar_tx_blacklist_tests,
 		ar_tx_replay_pool_tests,
 		ar_vdf_server_tests,
@@ -315,10 +318,27 @@ show_help() ->
 			{"block_throttle_by_solution_interval (number)",
 					io_lib:format("The number of milliseconds that have to pass before "
 							"we accept another block with the same solution hash. "
-							"Default: ~B.", [?DEFAULT_BLOCK_THROTTLE_BY_SOLUTION_INTERVAL_MS])},
+							"Default: ~B.",
+							[?DEFAULT_BLOCK_THROTTLE_BY_SOLUTION_INTERVAL_MS])},
 			{"defragment_module",
 				"Run defragmentation of the chunk storage files from the given storage module."
-				" Assumes the run_defragmentation flag is provided."}
+				" Assumes the run_defragmentation flag is provided."},
+			{"coordinated_mining", "Enable coordinated mining. You need to also set "
+					"coordinated_mining_secret, cm_peer, and cm_exit_peer."},
+			{"coordinated_mining_secret", "Coordinated mining secret for authenticated "
+					"requests between private peers. You need to also set coordinated_mining, "
+					"cm_peer, and cm_exit_peer."},
+			{"cm_poll_interval", io_lib:format("The frequency in milliseconds of asking the "
+					"other nodes in the coordinated mining setup about their partition "
+					"tables. Default is ~B.", [?DEFAULT_CM_POLL_INTERVAL])},
+			{"cm_stat_interval", io_lib:format("The frequency in milliseconds of printing the "
+					"coordinated mining statistics. Default is ~B.",
+					[?DEFAULT_CM_STAT_INTERVAL])},
+			{"cm_peer (IP:port)", "The peer(s) to mine in coordination with. You need to also "
+					"set coordinated_mining, coordinated_mining_secret, and cm_exit_peer."},
+			{"cm_exit_peer (IP:port)", "The peer to send mining solutions to in the "
+					"coordinated mining mode. You need to also set coordinated_mining, "
+					"coordinated_mining_secret, and cm_peer."}
 		]
 	),
 	erlang:halt().
@@ -551,6 +571,35 @@ parse_cli_args(["defragment_module", DefragModuleString | Rest], C) ->
 		io:format("~ndefragment_module value must be in the [number],[address] format.~n~n"),
 		erlang:halt()
 	end;
+parse_cli_args(["coordinated_mining" | Rest], C) ->
+	parse_cli_args(Rest, C#config{ coordinated_mining = true });
+parse_cli_args(["coordinated_mining_secret", CMSecret | Rest], C)
+		when length(CMSecret) >= ?INTERNAL_API_SECRET_MIN_LEN ->
+	parse_cli_args(Rest, C#config{ coordinated_mining_secret = list_to_binary(CMSecret) });
+parse_cli_args(["coordinated_mining_secret", _ | _], _) ->
+	io:format("~nThe coordinated_mining_secret must be at least ~B characters long.~n~n",
+			[?INTERNAL_API_SECRET_MIN_LEN]),
+	erlang:halt();
+parse_cli_args(["cm_poll_interval", Num | Rest], C) ->
+	parse_cli_args(Rest, C#config{ cm_poll_interval = list_to_integer(Num) });
+parse_cli_args(["cm_stat_interval", Num | Rest], C) ->
+	parse_cli_args(Rest, C#config{ cm_stat_interval = list_to_integer(Num) });
+parse_cli_args(["cm_peer", Peer | Rest], C = #config{ cm_peers = Ps }) ->
+	case ar_util:safe_parse_peer(Peer) of
+		{ok, ValidPeer} ->
+			parse_cli_args(Rest, C#config{ cm_peers = [ValidPeer|Ps] });
+		{error, _} ->
+			io:format("Peer ~p is invalid.~n", [Peer]),
+			parse_cli_args(Rest, C)
+	end;
+parse_cli_args(["cm_exit_peer", Peer | Rest], C) ->
+	case ar_util:safe_parse_peer(Peer) of
+		{ok, ValidPeer} ->
+			parse_cli_args(Rest, C#config{ cm_exit_peer = ValidPeer });
+		{error, _} ->
+			io:format("Peer ~p is invalid.~n", [Peer]),
+			parse_cli_args(Rest, C)
+	end;
 parse_cli_args([Arg | _Rest], _O) ->
 	io:format("~nUnknown argument: ~s.~n", [Arg]),
 	show_help().
@@ -675,16 +724,21 @@ set_mining_address(#config{ mining_addr = not_set } = C) ->
 	set_mining_address(C2);
 set_mining_address(#config{ mine = false }) ->
 	ok;
-set_mining_address(#config{ mining_addr = Addr }) ->
+set_mining_address(#config{ mining_addr = Addr, cm_exit_peer = CmExitPeer }) ->
 	case ar_wallet:load_key(Addr) of
 		not_found ->
-			ar:console("~nThe mining key for the address ~s was not found."
-				" Make sure you placed the file in [data_dir]/~s (the node is looking for"
-				" [data_dir]/~s/[mining_addr].json or "
-				"[data_dir]/~s/arweave_keyfile_[mining_addr].json file)."
-				" Do not specify \"mining_addr\" if you want one to be generated.~n~n",
-				[ar_util:encode(Addr), ?WALLET_DIR, ?WALLET_DIR, ?WALLET_DIR]),
-			erlang:halt();
+			case CmExitPeer of
+				not_set ->
+					ar:console("~nThe mining key for the address ~s was not found."
+						" Make sure you placed the file in [data_dir]/~s (the node is looking for"
+						" [data_dir]/~s/[mining_addr].json or "
+						"[data_dir]/~s/arweave_keyfile_[mining_addr].json file)."
+						" Do not specify \"mining_addr\" if you want one to be generated.~n~n",
+						[ar_util:encode(Addr), ?WALLET_DIR, ?WALLET_DIR, ?WALLET_DIR]),
+					erlang:halt();
+				_ ->
+					ok
+			end;
 		_Key ->
 			ok
 	end.

apps/arweave/src/ar_block.erl

@@ -690,10 +690,10 @@ hash_list_gen_test_() ->
 test_hash_list_gen() ->
 	[B0] = ar_weave:init([]),
 	ar_test_node:start(B0),
-	ar_node:mine(),
+	ar_test_node:mine(),
 	BI1 = ar_test_node:wait_until_height(1),
 	B1 = ar_storage:read_block(hd(BI1)),
-	ar_node:mine(),
+	ar_test_node:mine(),
 	BI2 = ar_test_node:wait_until_height(2),
 	B2 = ar_storage:read_block(hd(BI2)),
 	?assertEqual([B0#block.indep_hash], generate_hash_list_for_block(B1, BI2)),

apps/arweave/src/ar_block_propagation_worker.erl

@@ -49,9 +49,13 @@ handle_cast({send_block2, Peer, SendAnnouncementFun, SendFun, RetryCount, From},
 		{ok, {{<<"200">>, _}, _, Body, _, _}} ->
 			case catch ar_serialize:binary_to_block_announcement_response(Body) of
 				{'EXIT', Reason} ->
+					?LOG_INFO([{event, send_announcement_response}, {peer, ar_util:format_peer(Peer)},
+						{exit, Reason}]),
 					ar_peers:issue_warning(Peer, block_announcement, Reason),
 					From ! {worker_sent_block, self()};
 				{error, Reason} ->
+					?LOG_INFO([{event, send_announcement_response}, {peer, ar_util:format_peer(Peer)},
+						{error, Reason}]),
 					ar_peers:issue_warning(Peer, block_announcement, Reason),
 					From ! {worker_sent_block, self()};
 				{ok, #block_announcement_response{ missing_tx_indices = L,

apps/arweave/src/ar_chunk_storage.erl

@@ -87,7 +87,7 @@ get(Byte, StoreID) ->
 			not_found;
 		{_End, IntervalStart} ->
 			Start = Byte - (Byte - IntervalStart) rem ?DATA_CHUNK_SIZE,
-			LeftBorder = Start - Start rem ?CHUNK_GROUP_SIZE,
+			LeftBorder = ar_util:floor_int(Start, ?CHUNK_GROUP_SIZE),
 			case get(Byte, Start, LeftBorder, StoreID, 1) of
 				[] ->
 					not_found;
@@ -113,7 +113,7 @@ get_range(Start, Size, StoreID) ->
 			Start2 = max(Start, IntervalStart),
 			Size2 = Start + Size - Start2,
 			BucketStart = Start2 - (Start2 - IntervalStart) rem ?DATA_CHUNK_SIZE,
-			LeftBorder = BucketStart - BucketStart rem ?CHUNK_GROUP_SIZE,
+			LeftBorder = ar_util:floor_int(BucketStart, ?CHUNK_GROUP_SIZE),
 			End = Start2 + Size2,
 			LastBucketStart = (End - 1) - ((End - 1)- IntervalStart) rem ?DATA_CHUNK_SIZE,
 			case LastBucketStart >= LeftBorder + ?CHUNK_GROUP_SIZE of
@@ -404,7 +404,7 @@ handle_store_chunk(Offset, Chunk, FileIndex, StoreID) ->
 
 get_key(Offset) ->
 	StartOffset = Offset - ?DATA_CHUNK_SIZE,
-	StartOffset - StartOffset rem ?CHUNK_GROUP_SIZE.
+	ar_util:floor_int(StartOffset, ?CHUNK_GROUP_SIZE).
 
 store_chunk(Key, Offset, Chunk, FileIndex, StoreID) ->
 	Filepath = filepath(Key, FileIndex, StoreID),
@@ -440,7 +440,7 @@ store_chunk(Key, Offset, Chunk, Filepath) ->
 
 store_chunk2(Key, Offset, Chunk, Filepath, F) ->
 	StartOffset = Offset - ?DATA_CHUNK_SIZE,
-	LeftChunkBorder = StartOffset - StartOffset rem ?DATA_CHUNK_SIZE,
+	LeftChunkBorder = ar_util:floor_int(StartOffset, ?DATA_CHUNK_SIZE),
 	ChunkOffset = StartOffset - LeftChunkBorder,
 	RelativeOffset = LeftChunkBorder - Key,
 	Position = RelativeOffset + ?OFFSET_SIZE * (RelativeOffset div ?DATA_CHUNK_SIZE),
@@ -472,7 +472,7 @@ delete_chunk(Offset, Key, Filepath) ->
 	case file:open(Filepath, [read, write, raw]) of
 		{ok, F} ->
 			StartOffset = Offset - ?DATA_CHUNK_SIZE,
-			LeftChunkBorder = StartOffset - StartOffset rem ?DATA_CHUNK_SIZE,
+			LeftChunkBorder = ar_util:floor_int(StartOffset, ?DATA_CHUNK_SIZE),
 			RelativeOffset = LeftChunkBorder - Key,
 			Position = RelativeOffset + ?OFFSET_SIZE * (RelativeOffset div ?DATA_CHUNK_SIZE),
 			ZeroChunk =
@@ -526,7 +526,7 @@ read_chunk(Byte, Start, Key, Filepath, ChunkCount) ->
 	end.
 
 read_chunk2(Byte, Start, Key, File, ChunkCount) ->
-	LeftChunkBorder = Start - Start rem ?DATA_CHUNK_SIZE,
+	LeftChunkBorder = ar_util:floor_int(Start, ?DATA_CHUNK_SIZE),
 	RelativeOffset = LeftChunkBorder - Key,
 	Position = RelativeOffset + ?OFFSET_SIZE * RelativeOffset div ?DATA_CHUNK_SIZE,
 	read_chunk3(Byte, Position, LeftChunkBorder, File, ChunkCount).

apps/arweave/src/ar_config.erl

@@ -527,6 +527,44 @@ parse_options([{<<"p3">>, {P3Config}} | Rest], Config) ->
 			P3Config}
 	end;
 
+parse_options([{<<"coordinated_mining">>, true} | Rest], Config) ->
+	parse_options(Rest, Config#config{ coordinated_mining = true });
+parse_options([{<<"coordinated_mining">>, false} | Rest], Config) ->
+	parse_options(Rest, Config);
+parse_options([{<<"coordinated_mining">>, Opt} | _], _) ->
+	{error, {bad_type, coordinated_mining, boolean}, Opt};
+
+parse_options([{<<"coordinated_mining_secret">>, CMSecret} | Rest], Config) when is_binary(CMSecret), byte_size(CMSecret) >= ?INTERNAL_API_SECRET_MIN_LEN ->
+	parse_options(Rest, Config#config{ coordinated_mining_secret = CMSecret });
+parse_options([{<<"coordinated_mining_secret">>, CMSecret} | _], _) ->
+	{error, {bad_type, coordinated_mining_secret, string}, CMSecret};
+
+parse_options([{<<"cm_poll_interval">>, CMPollInterval} | Rest], Config) when is_integer(CMPollInterval) ->
+	parse_options(Rest, Config#config{ cm_poll_interval = CMPollInterval });
+parse_options([{<<"cm_poll_interval">>, CMPollInterval} | _], _) ->
+	{error, {bad_type, cm_poll_interval, number}, CMPollInterval};
+
+parse_options([{<<"cm_stat_interval">>, CMStatInterval} | Rest], Config) when is_integer(CMStatInterval) ->
+	parse_options(Rest, Config#config{ cm_stat_interval = CMStatInterval });
+parse_options([{<<"cm_stat_interval">>, CMStatInterval} | _], _) ->
+	{error, {bad_type, cm_stat_interval, number}, CMStatInterval};
+
+parse_options([{<<"cm_peers">>, Peers} | Rest], Config) when is_list(Peers) ->
+	case parse_peers(Peers, []) of
+		{ok, ParsedPeers} ->
+			parse_options(Rest, Config#config{ cm_peers = ParsedPeers });
+		error ->
+			{error, bad_peers, Peers}
+	end;
+
+parse_options([{<<"cm_exit_peer">>, Peer} | Rest], Config) ->
+	case ar_util:safe_parse_peer(Peer) of
+		{ok, ParsedPeer} ->
+			parse_options(Rest, Config#config{ cm_exit_peer = ParsedPeer });
+		{error, _} ->
+			{error, bad_cm_exit_peer, Peer}
+	end;
+
 parse_options([Opt | _], _) ->
 	{error, unknown, Opt};
 parse_options([], Config) ->