feat(turbine): implement retransmit stage #247
Conversation
0xNineteen
changed the title
yewman
force-pushed
the
harnew/turbine
branch
4 times, most recently
from
2ddc787
to
f56a27d
Compare
yewman
force-pushed
the
harnew/turbine
branch
from
66bba6c
to
0d16755
Compare
yewman
force-pushed
the
harnew/turbine
branch
5 times, most recently
from
2cd6003
to
4a9a900
Compare
yewman
force-pushed
the
harnew/turbine
branch
5 times, most recently
from
96a8034
to
03fa6d5
Compare
…pc and bank code to contexts where it is relevant (#296)
yewman
force-pushed
the
harnew/turbine
branch
from
8d5c5c0
to
91ca003
Compare
There was a problem hiding this comment.
PRNG naming convention: try to cooperate with the changes of #304, and use the appropriate naming conventions for prng-related declarations.
Named parameters: we need more of them, we have a lot of very long functions with parameter types that are not always unambiguous.
Naming & style guide: this review comments on a few improvable names.
Unmanaged data structures: I'd like to see more in here, I've written down a number of comments related to this. I recommend addressing other comments before this.
The rest is miscellaneous or adjacent to the above.
After addressing all of that, I think this PR will look great.
| pub fn pubkey(self: Node) Pubkey { | ||
| return switch (self.id) { | ||
| .contact_info => |ci| ci.pubkey, | ||
| .pubkey => |pk| pk, | ||
| }; | ||
| } | ||
|
|
||
| pub fn contactInfo(self: Node) ?ThreadSafeContactInfo { | ||
| return switch (self.id) { | ||
| .contact_info => |ci| ci, | ||
| .pubkey => null, | ||
| }; | ||
| } | ||
|
|
||
| pub fn tvuAddress(self: Node) ?SocketAddr { | ||
| return switch (self.id) { | ||
| .contact_info => |ci| ci.tvu_addr, | ||
| .pubkey => null, | ||
| }; | ||
| } |
There was a problem hiding this comment.
It would make more sense for all three of these to be methods on NodeId.
| fn computeRetransmitParent( | ||
| fanout: usize, | ||
| index_: usize, | ||
| nodes: []const Node, | ||
| ) ?Pubkey { | ||
| var index = index_; | ||
| const offset = (index -| 1) % fanout; | ||
| index = if (index == 0) return null else (index - 1) / fanout; | ||
| index = index - (index -| 1) % fanout; | ||
| index = if (index == 0) index else index + offset; | ||
| return nodes[index].pubkey(); | ||
| } |
There was a problem hiding this comment.
How about changing this to just return index, allowing the caller to index nodes themselves? This would play well with my other suggestion about using multiarraylist, where it'd allow the caller to just do nodes.items(.id)[computeRetransmitParent(fanout, index_)].pubkey() (in tandem with my other request to move the pubkey method into NodeId).
| num_known_nodes: usize, | ||
| num_unknown_staked_nodes: usize, | ||
| known_nodes_unstaked_ratio: struct { u64, u64 }, |
There was a problem hiding this comment.
Named parameter struct for at least these three parameters. The four numbers in TestEnvironment.init(allocator, rng, 1_000, 100, .{ 1, 7 }) feel pretty magical.
| }; | ||
| } | ||
|
|
||
| pub fn hash(self: *AHasher, comptime T: type, data: *const T) void { |
There was a problem hiding this comment.
This function should be removed, write should be made pub and renamed to update, and the current update should be renamed to updateInt.
This will allow a user to use std.hash.autoHash[Strat] for more complex data.
| inline fn hashSlice(self: *AHasher, comptime T: type, data: []const T) void { | ||
| self.update(data.len); | ||
| self.write(@as([]const u8, data)); | ||
| } |
There was a problem hiding this comment.
This should be removed in light of my other comment.
Implementation appears to be working correctly.
Most logic is well tested, however, there are some areas that will receive some extra attention regarding unit testing for equivalence with agave.
Metrics from 1 hour runtime
We can make these a little nicer but are they at least demonstrate functionality for now.
✅ turbine/retransmit_service.sig
Core logic mirrored from agave.
✅ turbine/turbine_tree.zig
✅ turbine/shred_deduper.zig
Closely mirrored from agave.
ShredDeduper provides a means to probabilistically deduplicate incoming shreds based on their shred id and the raw bytes. Shred id's are allowed up to a specified number of duplicates, raw bytes are not allowed any duplicates.
✅ utils/deduper.zig
Closely mirrored from agave.
Deduper is a generic probabilistic deduplication implementation.
✅ utils/ahash.zig
Closely mirrored from ahash rust crate.
Currently only includes functionality required for the Deduper.
✅ rand/weighted_shuffle.zig
Closely mirrored from agave.
WeightedShuffle implements an iterator over indices shuffled according to their weight in an underlying weights array.