address race in geyser (tsan failing on ci)

src/accountsdb/db.zig

@@ -3474,12 +3474,18 @@ test "geyser stream on load" {
     // start the geyser writer
     try geyser_writer.spawnIOLoop();
 
-    const reader_handle = try std.Thread.spawn(.{}, sig.geyser.core.streamReader, .{
+    var reader = try sig.geyser.GeyserReader.init(
         allocator,
+        geyser_pipe_path,
+        &geyser_exit,
+        .{},
+    );
+    defer reader.deinit();
+
+    const reader_handle = try std.Thread.spawn(.{}, sig.geyser.core.streamReader, .{
+        &reader,
         .noop,
         &geyser_exit,
-        geyser_pipe_path,
-        null,
         null,
     });
     defer reader_handle.join();
@@ -3505,7 +3511,7 @@ test "geyser stream on load" {
         snapshot.accounts_db_fields,
         1,
         allocator,
-        1_500,
+        500,
     );
 }
 

src/geyser/benchmark.zig

@@ -61,10 +61,18 @@ pub fn runBenchmark(logger: sig.trace.Logger) !void {
 
     exit.* = std.atomic.Value(bool).init(false);
 
+    var reader = try sig.geyser.GeyserReader.init(
+        allocator,
+        PIPE_PATH,
+        exit,
+        .{},
+    );
+    defer reader.deinit();
+
     const reader_handle = try std.Thread.spawn(
         .{},
         geyser.core.streamReader,
-        .{ allocator, logger, exit, PIPE_PATH, MEASURE_RATE, null },
+        .{ &reader, logger, exit, MEASURE_RATE },
     );
     const writer_handle = try std.Thread.spawn(.{}, streamWriter, .{ allocator, exit });
 

src/geyser/core.zig

@@ -479,16 +479,11 @@ pub fn openPipe(pipe_path: []const u8) !std.fs.File {
 }
 
 pub fn streamReader(
-    allocator: std.mem.Allocator,
+    reader: *GeyserReader,
     logger: sig.trace.Logger,
     exit: *std.atomic.Value(bool),
-    pipe_path: []const u8,
     measure_rate: ?sig.time.Duration,
-    allocator_config: ?GeyserReader.AllocatorConfig,
 ) !void {
-    var reader = try sig.geyser.GeyserReader.init(allocator, pipe_path, exit, allocator_config orelse .{});
-    defer reader.deinit();
-
     var bytes_read: usize = 0;
     var timer = try sig.time.Timer.start();
 

src/geyser/main.zig

@@ -356,15 +356,22 @@ pub fn benchmark() !void {
     logger.info().logf("using pipe path: {s}", .{pipe_path});
 
     var exit = std.atomic.Value(bool).init(false);
-    try sig.geyser.core.streamReader(
+
+    var reader = try sig.geyser.GeyserReader.init(
         allocator,
-        logger,
-        &exit,
         pipe_path,
-        sig.time.Duration.fromSecs(config.measure_rate_secs),
+        &exit,
         .{
             .io_buf_len = 1 << 30,
             .bincode_buf_len = 1 << 30,
         },
     );
+    defer reader.deinit();
+
+    try sig.geyser.core.streamReader(
+        &reader,
+        logger,
+        &exit,
+        sig.time.Duration.fromSecs(config.measure_rate_secs),
+    );
 }