Merge branch 'main' of https://github.com/Syndica/sig into 19/snapsho…

…t-loading-fixes

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/download.zig

@@ -10,7 +10,7 @@ const Logger = @import("../trace/log.zig").Logger;
 const socket_tag = @import("../gossip/data.zig").socket_tag;
 const Hash = @import("../core/hash.zig").Hash;
 
-const DOWNLOAD_PROGRESS_UPDATES_NS = 10 * std.time.ns_per_s;
+const DOWNLOAD_PROGRESS_UPDATES_NS = 30 * std.time.ns_per_s;
 
 const PeerSnapshotHash = struct {
     contact_info: ContactInfo,
@@ -322,6 +322,10 @@ const DownloadProgress = struct {
         };
     }
 
+    pub fn deinit(self: *Self) void {
+        std.posix.munmap(self.mmap);
+    }
+
     pub fn bufferWriteCallback(ptr: [*c]c_char, size: c_uint, nmemb: c_uint, user_data: *anyopaque) callconv(.C) c_uint {
         const len = size * nmemb;
         var self: *Self = @alignCast(@ptrCast(user_data));
@@ -397,7 +401,7 @@ pub fn downloadFile(
     allocator: std.mem.Allocator,
     logger: Logger,
     url: [:0]const u8,
-    output_dir: []const u8,
+    output_dir_str: []const u8,
     filename: []const u8,
     min_mb_per_second: ?usize,
 ) !void {
@@ -423,11 +427,19 @@ pub fn downloadFile(
     easy.timeout_ms = std.time.ms_per_hour * 5; // 5 hours is probs too long but its ok
     var download_progress = try DownloadProgress.init(
         logger,
-        output_dir,
+        output_dir_str,
         filename,
         download_size,
         min_mb_per_second,
     );
+    errdefer {
+        // NOTE: this shouldnt fail because we open the dir in DownloadProgress.init
+        const output_dir = std.fs.cwd().openDir(output_dir_str, .{}) catch {
+            std.debug.panic("failed to open output dir: {s}", .{output_dir_str});
+        };
+        output_dir.deleteFile(filename) catch {};
+    }
+    defer download_progress.deinit();
 
     try setNoBody(easy, false); // full download
     try easy.setUrl(url);

src/accountsdb/snapshots.zig

@@ -1023,6 +1023,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,16 +39,22 @@ 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;
 
 const ACCOUNT_INDEX_BINS = @import("../accountsdb/db.zig").ACCOUNT_INDEX_BINS;
 const socket_tag = @import("../gossip/data.zig").socket_tag;
 
+// TODO: use better allocator, unless GPA becomes more performant.
+
 var gpa = std.heap.GeneralPurposeAllocator(.{}){};
 const gpa_allocator = gpa.allocator();
+
+var gossip_value_gpa: std.heap.GeneralPurposeAllocator(.{}) = .{};
+const gossip_value_gpa_allocator = gossip_value_gpa.allocator();
+
 const base58Encoder = base58.Encoder.init(.{});
 
 const gossip_host = struct {
@@ -376,6 +382,7 @@ fn gossip() !void {
         &.{},
     );
     defer gossip_service.deinit();
+
     try runGossipWithConfigValues(&gossip_service);
 }
 
@@ -411,7 +418,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,
@@ -531,6 +538,7 @@ fn initGossip(
 
     return try GossipService.init(
         gpa_allocator,
+        gossip_value_gpa_allocator,
         contact_info,
         my_keypair,
         entrypoints,
@@ -572,7 +580,10 @@ fn initRepair(
 
 fn runGossipWithConfigValues(gossip_service: *GossipService) !void {
     const gossip_config = config.current.gossip;
-    return gossip_service.run(gossip_config.spy_node, gossip_config.dump);
+    return gossip_service.run(.{
+        .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

@@ -300,6 +300,7 @@ pub fn run() !void {
 
     var fuzz_exit = AtomicBool.init(false);
     var gossip_service_fuzzer = try GossipService.init(
+        allocator,
         allocator,
         fuzz_contact_info,
         fuzz_keypair,
@@ -308,7 +309,12 @@ 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, .{
+            .spy_node = true,
+            .dump = false,
+        },
+    });
 
     const SLEEP_TIME = 0;
     // const SLEEP_TIME = std.time.ns_per_ms * 10;