Add a connect_on_init_count pool option

#64

Author: karlseguin

readme.md

@@ -59,10 +59,11 @@ Initializes a connection pool. Pool options are:
 * `size` - Number of connections to maintain. Defaults to `10`
 * `auth`: - See [Conn.auth](#authopts-opts-void)
 * `connect`: - See the [Conn.open](#openallocator-stdmemallocator-opts-opts-conn)
-* `timeout` - The amount of time, in milliseconds, to wait for a connection to be available when `acquire()` is called.
+* `timeout`: - The amount of time, in milliseconds, to wait for a connection to be available when `acquire()` is called.
+* `connect_on_init_count`:  - The # of connections in the pool to eagerly connect during `init`. Defaults to `null` which will initiliaze all connections (`size`). The background reconnector is used to setup the remaining (`size - connect_on_init_count`) connections. This can be set to `0`, to prevent `init` from failing except in extreme cases (i.e. OOM), but that will hide any configuration/connection issue until the first query is executed.
 
-### initUri(allocator: std.mem.Allocator, uri: std.Uri, size: u16, pool_timeout_ms: u32) !*Pool
-Initializes a connection pool using a std.Uri.
+### initUri(allocator: std.mem.Allocator, uri: std.Uri, opts: Opts) !*Pool
+Initializes a connection pool using a std.Uri. When using this function, the `auth` and `connect` fields of `opts` should **not** be set, as these will automatically set based on the provided `uri`.
 
 ```zig
 const uri = try std.Uri.parse("postgresql://username:password@localhost:5432/database_name");

src/conn.zig

@@ -108,7 +108,7 @@ pub const Conn = struct {
     };
 
     pub fn openAndAuthUri(allocator: Allocator, uri: std.Uri) !Conn {
-        var po = try lib.parseOpts(uri, allocator, 0, 0);
+        var po = try lib.parseOpts(uri, allocator);
         defer po.deinit();
         return try openAndAuth(allocator, po.opts.connect, po.opts.auth);
     }

src/lib.zig

@@ -108,7 +108,7 @@ pub const ParsedOpts = struct {
     }
 };
 
-pub fn parseOpts(uri: std.Uri, allocator: std.mem.Allocator, size: u16, pool_timeout_ms: u32) !ParsedOpts {
+pub fn parseOpts(uri: std.Uri, allocator: std.mem.Allocator) !ParsedOpts {
     if (!std.mem.eql(u8, uri.scheme, "postgresql")) {
         return error.InvalidUriScheme;
     }
@@ -144,7 +144,8 @@ pub fn parseOpts(uri: std.Uri, allocator: std.mem.Allocator, size: u16, pool_tim
 
     const path = std.mem.trimLeft(u8, try uri.path.toRawMaybeAlloc(aa), "/");
     return .{ .arena = arena, .opts = .{
-        .size = size,
+        .size = 0,
+        .timeout = 0,
         .auth = .{
             .username = if (uri.user) |user| try user.toRawMaybeAlloc(aa) else "postgres",
             .password = if (uri.password) |password| try password.toRawMaybeAlloc(aa) else null,
@@ -156,7 +157,6 @@ pub fn parseOpts(uri: std.Uri, allocator: std.mem.Allocator, size: u16, pool_tim
             .port = uri.port orelse null,
             .host = if (uri.host) |host| try host.toRawMaybeAlloc(aa) else null,
         },
-        .timeout = pool_timeout_ms,
     } };
 }
 
@@ -242,22 +242,22 @@ const TestCase = struct {
 };
 
 const valid_tcs: [2]TestCase = .{
-    .{ .uri = "postgresql:///", .expected_opts = .{ .size = 10, .auth = .{ .username = "postgres" }, .connect = .{}, .timeout = 5000 } },
-    .{ .uri = "postgresql://user:pass@somehost:1234/somedb?tcp_user_timeout=5678", .expected_opts = .{ .size = 10, .auth = .{
+    .{ .uri = "postgresql:///", .expected_opts = .{ .size = 0, .auth = .{ .username = "postgres" }, .connect = .{}, .timeout = 0 } },
+    .{ .uri = "postgresql://user:pass@somehost:1234/somedb?tcp_user_timeout=5678", .expected_opts = .{ .size = 0, .auth = .{
         .username = "user",
         .password = "pass",
         .database = "somedb",
         .timeout = 5678,
     }, .connect = .{
         .host = "somehost",
         .port = 1234,
-    }, .timeout = 5000 } },
+    }, .timeout = 0 } },
 };
 
 test "URI: parse valid" {
     const a = std.testing.allocator;
     for (valid_tcs) |tc| {
-        var po = parseOpts(try std.Uri.parse(tc.uri), a, 10, 5000) catch |e| {
+        var po = parseOpts(try std.Uri.parse(tc.uri), a) catch |e| {
             std.log.err("failed to parse URI {s}", .{tc.uri});
             return e;
         };
@@ -267,9 +267,9 @@ test "URI: parse valid" {
 }
 
 test "URI: invalid scheme" {
-    try std.testing.expectError(error.InvalidUriScheme, parseOpts(try std.Uri.parse("foobar:///"), std.testing.allocator, 0, 0));
+    try std.testing.expectError(error.InvalidUriScheme, parseOpts(try std.Uri.parse("foobar:///"), std.testing.allocator));
 }
 
 test "URI: invalid params" {
-    try std.testing.expectError(error.UnsupportedConnectionParam, parseOpts(try std.Uri.parse("postgresql:///?bar=baz"), std.testing.allocator, 0, 0));
+    try std.testing.expectError(error.UnsupportedConnectionParam, parseOpts(try std.Uri.parse("postgresql:///?bar=baz"), std.testing.allocator));
 }

src/pool.zig

@@ -29,11 +29,15 @@ pub const Pool = struct {
         auth: auth.Opts = .{},
         connect: Conn.Opts = .{},
         timeout: u32 = 10 * std.time.ms_per_s,
+        connect_on_init_count: ?u16 = null,
     };
 
-    pub fn initUri(allocator: Allocator, uri: std.Uri, size: u16, pool_timeout_ms: u32) !*Pool {
-        var po = try lib.parseOpts(uri, allocator, size, pool_timeout_ms);
+
+    pub fn initUri(allocator: Allocator, uri: std.Uri, opts: Opts) !*Pool {
+        var po = try lib.parseOpts(uri, allocator);
         defer po.deinit();
+        po.size = opts.size;
+        po.timeout = opts.timeout;
         return Pool.init(allocator, po.opts);
     }
 
@@ -60,16 +64,17 @@ pub const Pool = struct {
             }
         }
         errdefer lib.freeSSLContext(ssl_ctx);
+        const connect_on_init_count = opts.connect_on_init_count orelse size;
 
         pool.* = .{
             ._cond = .{},
             ._mutex = .{},
             ._conns = conns,
             ._arena = arena,
             ._opts = opts_copy,
-            ._available = size,
             ._ssl_ctx = ssl_ctx,
             ._allocator = allocator,
+            ._available = connect_on_init_count,
             ._reconnector = Reconnector.init(pool),
             ._timeout = @as(u64, @intCast(opts.timeout)) * std.time.ns_per_ms,
         };
@@ -81,11 +86,16 @@ pub const Pool = struct {
             }
         }
 
-        for (0..size) |i| {
+        for (0..connect_on_init_count) |i| {
             conns[i] = try newConnection(pool, true);
             opened_connections += 1;
         }
 
+        const lazy_start_count = size - connect_on_init_count;
+        for (0..lazy_start_count) |_| {
+            try pool._reconnector.reconnect();
+        }
+
         return pool;
     }
 
@@ -218,6 +228,7 @@ const Reconnector = struct {
         const retry_delay = 2 * std.time.ns_per_s;
 
         self.mutex.lock();
+        defer self.mutex.unlock();
         loop: while (self.count > 0) {
             const stopped = self.stopped;
             self.mutex.unlock();
@@ -235,6 +246,9 @@ const Reconnector = struct {
             self.mutex.lock();
             self.count -= 1;
         }
+
+        self.thread.?.detach();
+        self.thread = null;
     }
 
     fn stop(self: *Reconnector) void {
@@ -248,11 +262,11 @@ const Reconnector = struct {
 
     fn reconnect(self: *Reconnector) !void {
         self.mutex.lock();
+        defer self.mutex.unlock();
         self.count += 1;
         if (self.thread == null) {
             self.thread = try Thread.spawn(.{ .stack_size = 1024 * 1024 }, Reconnector.run, .{self});
         }
-        self.mutex.unlock();
     }
 };
 
@@ -291,6 +305,7 @@ test "Pool" {
     var pool = try Pool.init(t.allocator, .{
         .size = 2,
         .auth = t.authOpts(.{}),
+        .connect_on_init_count = 1,
     });
     defer pool.deinit();