authorgravatar for sachabarsayuracko@gmail.comglowsquid <sachabarsayuracko@gmail.com> 2026-04-16 19:47:46+02:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-04-16 19:47:46+02:00
log0177cb57c03d0e10b4d0ef6a11ebf16de3dfd2b5
treec599aa3d7c8b8dea2ae82d8b5c176c98f6154e55
parentffcfeb919fc08724eea0346af8ea5531d62e2734

std.http.Client: make mutexes cancelable (#31880)

Makes most of the mutexes cancelable with the exception of the ones in deinit functions. I also fixed the compilation errors with `ConnectionPool.resize`. Reviewed-on: https://codeberg.org/ziglang/zig/pulls/31880 Reviewed-by: Andrew Kelley <andrew@ziglang.org> Co-authored-by: glowsquid <sachabarsayuracko@gmail.com> Co-committed-by: glowsquid <sachabarsayuracko@gmail.com>

1 files changed, 21 insertions(+), 24 deletions(-)

lib/std/http/Client.zig+21-24
...@@ -4,8 +4,6 @@...@@ -4,8 +4,6 @@
4//!4//!
5//! TLS support may be disabled via `std.options.http_disable_tls`.5//! TLS support may be disabled via `std.options.http_disable_tls`.
6//!6//!
7//! TODO all the lockUncancelable in this file should be changed to regular lock and
8//! `error.Canceled` added to more error sets.
9const Client = @This();7const Client = @This();
108
11const builtin = @import("builtin");9const builtin = @import("builtin");
...@@ -84,8 +82,8 @@ pub const ConnectionPool = struct {...@@ -84,8 +82,8 @@ pub const ConnectionPool = struct {
84 /// If no connection is found, null is returned.82 /// If no connection is found, null is returned.
85 ///83 ///
86 /// Threadsafe.84 /// Threadsafe.
87 pub fn findConnection(pool: *ConnectionPool, io: Io, criteria: Criteria) ?*Connection {85 pub fn findConnection(pool: *ConnectionPool, io: Io, criteria: Criteria) Io.Cancelable!?*Connection {
88 pool.mutex.lockUncancelable(io);86 try pool.mutex.lock(io);
89 defer pool.mutex.unlock(io);87 defer pool.mutex.unlock(io);
9088
91 var next = pool.free.last;89 var next = pool.free.last;
...@@ -113,8 +111,8 @@ pub const ConnectionPool = struct {...@@ -113,8 +111,8 @@ pub const ConnectionPool = struct {
113 }111 }
114112
115 /// Acquires an existing connection from the connection pool. This function is threadsafe.113 /// Acquires an existing connection from the connection pool. This function is threadsafe.
116 pub fn acquire(pool: *ConnectionPool, io: Io, connection: *Connection) void {114 pub fn acquire(pool: *ConnectionPool, io: Io, connection: *Connection) Io.Cancelable!void {
117 pool.mutex.lockUncancelable(io);115 try pool.mutex.lock(io);
118 defer pool.mutex.unlock(io);116 defer pool.mutex.unlock(io);
119117
120 return pool.acquireUnsafe(connection);118 return pool.acquireUnsafe(connection);
...@@ -150,8 +148,8 @@ pub const ConnectionPool = struct {...@@ -150,8 +148,8 @@ pub const ConnectionPool = struct {
150 }148 }
151149
152 /// Adds a newly created node to the pool of used connections. This function is threadsafe.150 /// Adds a newly created node to the pool of used connections. This function is threadsafe.
153 pub fn addUsed(pool: *ConnectionPool, io: Io, connection: *Connection) void {151 pub fn addUsed(pool: *ConnectionPool, io: Io, connection: *Connection) Io.Cancelable!void {
154 pool.mutex.lockUncancelable(io);152 try pool.mutex.lock(io);
155 defer pool.mutex.unlock(io);153 defer pool.mutex.unlock(io);
156154
157 pool.used.append(&connection.pool_node);155 pool.used.append(&connection.pool_node);
...@@ -162,18 +160,15 @@ pub const ConnectionPool = struct {...@@ -162,18 +160,15 @@ pub const ConnectionPool = struct {
162 /// If the new size is smaller than the current size, then idle connections will be closed until the pool is the new size.160 /// If the new size is smaller than the current size, then idle connections will be closed until the pool is the new size.
163 ///161 ///
164 /// Threadsafe.162 /// Threadsafe.
165 pub fn resize(pool: *ConnectionPool, io: Io, allocator: Allocator, new_size: usize) void {163 pub fn resize(pool: *ConnectionPool, io: Io, new_size: usize) Io.Cancelable!void {
166 pool.mutex.lockUncancelable(io);164 try pool.mutex.lock(io);
167 defer pool.mutex.unlock(io);165 defer pool.mutex.unlock(io);
168166
169 const next = pool.free.first;
170 _ = next;
171 while (pool.free_len > new_size) {167 while (pool.free_len > new_size) {
172 const popped = pool.free.popFirst() orelse unreachable;168 const popped: *Connection = @alignCast(@fieldParentPtr("pool_node", pool.free.popFirst().?));
173 pool.free_len -= 1;169 pool.free_len -= 1;
174170
175 popped.data.close(allocator);171 popped.destroy(io);
176 allocator.destroy(popped);
177 }172 }
178173
179 pool.free_size = new_size;174 pool.free_size = new_size;
...@@ -1323,7 +1318,7 @@ pub fn initDefaultProxies(client: *Client, arena: Allocator, environ_map: *const...@@ -1323,7 +1318,7 @@ pub fn initDefaultProxies(client: *Client, arena: Allocator, environ_map: *const
1323 const io = client.io;1318 const io = client.io;
13241319
1325 // Prevent any new connections from being created.1320 // Prevent any new connections from being created.
1326 client.connection_pool.mutex.lockUncancelable(io);1321 try client.connection_pool.mutex.lock(io);
1327 defer client.connection_pool.mutex.unlock(io);1322 defer client.connection_pool.mutex.unlock(io);
13281323
1329 assert(client.connection_pool.used.first == null); // There are active requests.1324 assert(client.connection_pool.used.first == null); // There are active requests.
...@@ -1418,7 +1413,7 @@ pub const basic_authorization = struct {...@@ -1418,7 +1413,7 @@ pub const basic_authorization = struct {
14181413
1419pub const ConnectTcpError = error{1414pub const ConnectTcpError = error{
1420 TlsInitializationFailed,1415 TlsInitializationFailed,
1421} || Allocator.Error || HostName.ConnectError;1416} || Allocator.Error || HostName.ConnectError || Io.Cancelable;
14221417
1423/// Reuses a `Connection` if one matching `host` and `port` is already open.1418/// Reuses a `Connection` if one matching `host` and `port` is already open.
1424///1419///
...@@ -1451,7 +1446,7 @@ pub fn connectTcpOptions(client: *Client, options: ConnectTcpOptions) ConnectTcp...@@ -1451,7 +1446,7 @@ pub fn connectTcpOptions(client: *Client, options: ConnectTcpOptions) ConnectTcp
1451 const proxied_host = options.proxied_host orelse host;1446 const proxied_host = options.proxied_host orelse host;
1452 const proxied_port = options.proxied_port orelse port;1447 const proxied_port = options.proxied_port orelse port;
14531448
1454 if (client.connection_pool.findConnection(io, .{1449 if (try client.connection_pool.findConnection(io, .{
1455 .host = proxied_host,1450 .host = proxied_host,
1456 .port = proxied_port,1451 .port = proxied_port,
1457 .protocol = protocol,1452 .protocol = protocol,
...@@ -1469,18 +1464,20 @@ pub fn connectTcpOptions(client: *Client, options: ConnectTcpOptions) ConnectTcp...@@ -1469,18 +1464,20 @@ pub fn connectTcpOptions(client: *Client, options: ConnectTcpOptions) ConnectTcp
1469 error.Canceled => |e| return e,1464 error.Canceled => |e| return e,
1470 else => return error.TlsInitializationFailed,1465 else => return error.TlsInitializationFailed,
1471 };1466 };
1472 client.connection_pool.addUsed(io, &tc.connection);1467 errdefer tc.destroy();
1468 try client.connection_pool.addUsed(io, &tc.connection);
1473 return &tc.connection;1469 return &tc.connection;
1474 },1470 },
1475 .plain => {1471 .plain => {
1476 const pc = try Connection.Plain.create(client, proxied_host, proxied_port, stream);1472 const pc = try Connection.Plain.create(client, proxied_host, proxied_port, stream);
1477 client.connection_pool.addUsed(io, &pc.connection);1473 errdefer pc.destroy();
1474 try client.connection_pool.addUsed(io, &pc.connection);
1478 return &pc.connection;1475 return &pc.connection;
1479 },1476 },
1480 }1477 }
1481}1478}
14821479
1483pub const ConnectUnixError = Allocator.Error || std.posix.SocketError || error{NameTooLong} || std.posix.ConnectError;1480pub const ConnectUnixError = Allocator.Error || std.posix.SocketError || error{NameTooLong} || std.posix.ConnectError || Io.Cancelable;
14841481
1485/// Connect to `path` as a unix domain socket. This will reuse a connection if one is already open.1482/// Connect to `path` as a unix domain socket. This will reuse a connection if one is already open.
1486///1483///
...@@ -1488,7 +1485,7 @@ pub const ConnectUnixError = Allocator.Error || std.posix.SocketError || error{N...@@ -1488,7 +1485,7 @@ pub const ConnectUnixError = Allocator.Error || std.posix.SocketError || error{N
1488pub fn connectUnix(client: *Client, path: []const u8) ConnectUnixError!*Connection {1485pub fn connectUnix(client: *Client, path: []const u8) ConnectUnixError!*Connection {
1489 const io = client.io;1486 const io = client.io;
14901487
1491 if (client.connection_pool.findConnection(io, .{1488 if (try client.connection_pool.findConnection(io, .{
1492 .host = path,1489 .host = path,
1493 .port = 0,1490 .port = 0,
1494 .protocol = .plain,1491 .protocol = .plain,
...@@ -1512,7 +1509,7 @@ pub fn connectUnix(client: *Client, path: []const u8) ConnectUnixError!*Connecti...@@ -1512,7 +1509,7 @@ pub fn connectUnix(client: *Client, path: []const u8) ConnectUnixError!*Connecti
1512 };1509 };
1513 errdefer client.allocator.free(conn.data.host);1510 errdefer client.allocator.free(conn.data.host);
15141511
1515 client.connection_pool.addUsed(conn);1512 try client.connection_pool.addUsed(conn);
15161513
1517 return &conn.data;1514 return &conn.data;
1518}1515}
...@@ -1530,7 +1527,7 @@ pub fn connectProxied(...@@ -1530,7 +1527,7 @@ pub fn connectProxied(
1530 const io = client.io;1527 const io = client.io;
1531 if (!proxy.supports_connect) return error.TunnelNotSupported;1528 if (!proxy.supports_connect) return error.TunnelNotSupported;
15321529
1533 if (client.connection_pool.findConnection(io, .{1530 if (try client.connection_pool.findConnection(io, .{
1534 .host = proxied_host,1531 .host = proxied_host,
1535 .port = proxied_port,1532 .port = proxied_port,
1536 .protocol = proxy.protocol,1533 .protocol = proxy.protocol,