Merge pull request #145 from Syndica/ink/better-threads1

improvement(gossip,tar): Work towards improving thread management

build.zig.zon

@@ -28,8 +28,8 @@
             .hash = "1220fe113318a795d366cd63d2d3f1abacfaa6ff0739bc3240eaf7b68b06d21e4917",
         },
         .zstd = .{
-            .url = "https://github.com/Syndica/zstd.zig/archive/d9f86cad45380bf6b5d1e83ac8514cbddf3c4a38.tar.gz",
-            .hash = "1220be2064e0060e3836ee349595119c9d5874b58ee76147897d4070206f9e9e5db6",
+            .url = "https://github.com/Syndica/zstd.zig/archive/a052e839a3dfc44feb00c2eb425815baa3c76e0d.tar.gz",
+            .hash = "122001d56e43ef94e31243739ae83d7508abf0b8102795aff1ac91446e7ff450d875",
         },
         .curl = .{
             .url = "https://github.com/jiacai2050/zig-curl/archive/7b1e1c6adb1daca48bbae6bd18fc1386dc6076ab.tar.gz",

src/accountsdb/snapshots.zig

@@ -1016,6 +1016,7 @@ pub fn parallelUnpackZstdTarBall(
         0,
     );
     var tar_stream = try ZstdReader.init(memory);
+    defer tar_stream.deinit();
     const n_files_estimate: usize = if (full_snapshot) 421_764 else 100_000; // estimate
 
     try parallelUntarToFileSystem(

src/cmd/cmd.zig

@@ -39,7 +39,7 @@ const requestIpEcho = @import("../net/echo.zig").requestIpEcho;
 const servePrometheus = @import("../prometheus/http.zig").servePrometheus;
 const parallelUnpackZstdTarBall = @import("../accountsdb/snapshots.zig").parallelUnpackZstdTarBall;
 const downloadSnapshotsFromGossip = @import("../accountsdb/download.zig").downloadSnapshotsFromGossip;
-const SOCKET_TIMEOUT = @import("../net/socket_utils.zig").SOCKET_TIMEOUT;
+const SOCKET_TIMEOUT_US = @import("../net/socket_utils.zig").SOCKET_TIMEOUT_US;
 
 const config = @import("config.zig");
 // var validator_config = config.current;
@@ -376,6 +376,7 @@ fn gossip() !void {
         &.{},
     );
     defer gossip_service.deinit();
+
     try runGossipWithConfigValues(&gossip_service);
 }
 
@@ -411,7 +412,7 @@ fn validator() !void {
     // repair
     var repair_socket = try Socket.create(network.AddressFamily.ipv4, network.Protocol.udp);
     try repair_socket.bindToPort(repair_port);
-    try repair_socket.setReadTimeout(SOCKET_TIMEOUT);
+    try repair_socket.setReadTimeout(SOCKET_TIMEOUT_US);
 
     var repair_svc = try initRepair(
         logger,
@@ -571,8 +572,16 @@ fn initRepair(
 }
 
 fn runGossipWithConfigValues(gossip_service: *GossipService) !void {
+    // TODO: use better allocator, unless GPA becomes more performant.
+    var gp_message_allocator: std.heap.GeneralPurposeAllocator(.{}) = .{};
+    defer _ = gp_message_allocator.deinit();
+
     const gossip_config = config.current.gossip;
-    return gossip_service.run(gossip_config.spy_node, gossip_config.dump);
+    return gossip_service.run(.{
+        .message_allocator = gp_message_allocator.allocator(),
+        .spy_node = gossip_config.spy_node,
+        .dump = gossip_config.dump,
+    });
 }
 
 /// determine our shred version and ip. in the solana-labs client, the shred version

src/gossip/fuzz.zig

@@ -308,7 +308,13 @@ pub fn run() !void {
         .noop,
     );
 
-    var fuzz_handle = try std.Thread.spawn(.{}, GossipService.run, .{ &gossip_service_fuzzer, true, false });
+    const fuzz_handle = try std.Thread.spawn(.{}, GossipService.run, .{
+        &gossip_service_fuzzer, .{
+            .message_allocator = allocator,
+            .spy_node = true,
+            .dump = false,
+        },
+    });
 
     const SLEEP_TIME = 0;
     // const SLEEP_TIME = std.time.ns_per_ms * 10;