| ... | @@ -46,9 +46,9 @@ https_proxy: ?*Proxy = null, | ... | @@ -46,9 +46,9 @@ https_proxy: ?*Proxy = null, |
| 46 | pub const ConnectionPool = struct { | 46 | pub const ConnectionPool = struct { |
| 47 | mutex: std.Thread.Mutex = .{}, | 47 | mutex: std.Thread.Mutex = .{}, |
| 48 | /// Open connections that are currently in use. | 48 | /// Open connections that are currently in use. |
| 49 | used: Queue = .{}, | 49 | used: std.DoublyLinkedList = .{}, |
| 50 | /// Open connections that are not currently in use. | 50 | /// Open connections that are not currently in use. |
| 51 | free: Queue = .{}, | 51 | free: std.DoublyLinkedList = .{}, |
| 52 | free_len: usize = 0, | 52 | free_len: usize = 0, |
| 53 | free_size: usize = 32, | 53 | free_size: usize = 32, |
| 54 | | 54 | |
| ... | @@ -59,9 +59,6 @@ pub const ConnectionPool = struct { | ... | @@ -59,9 +59,6 @@ pub const ConnectionPool = struct { |
| 59 | protocol: Connection.Protocol, | 59 | protocol: Connection.Protocol, |
| 60 | }; | 60 | }; |
| 61 | | 61 | |
| 62 | const Queue = std.DoublyLinkedList(Connection); | | |
| 63 | pub const Node = Queue.Node; | | |
| 64 | | | |
| 65 | /// Finds and acquires a connection from the connection pool matching the criteria. This function is threadsafe. | 62 | /// Finds and acquires a connection from the connection pool matching the criteria. This function is threadsafe. |
| 66 | /// If no connection is found, null is returned. | 63 | /// If no connection is found, null is returned. |
| 67 | pub fn findConnection(pool: *ConnectionPool, criteria: Criteria) ?*Connection { | 64 | pub fn findConnection(pool: *ConnectionPool, criteria: Criteria) ?*Connection { |
| ... | @@ -70,33 +67,34 @@ pub const ConnectionPool = struct { | ... | @@ -70,33 +67,34 @@ pub const ConnectionPool = struct { |
| 70 | | 67 | |
| 71 | var next = pool.free.last; | 68 | var next = pool.free.last; |
| 72 | while (next) |node| : (next = node.prev) { | 69 | while (next) |node| : (next = node.prev) { |
| 73 | if (node.data.protocol != criteria.protocol) continue; | 70 | const connection: *Connection = @fieldParentPtr("pool_node", node); |
| 74 | if (node.data.port != criteria.port) continue; | 71 | if (connection.protocol != criteria.protocol) continue; |
| | 72 | if (connection.port != criteria.port) continue; |
| 75 | | 73 | |
| 76 | // Domain names are case-insensitive (RFC 5890, Section 2.3.2.4) | 74 | // Domain names are case-insensitive (RFC 5890, Section 2.3.2.4) |
| 77 | if (!std.ascii.eqlIgnoreCase(node.data.host, criteria.host)) continue; | 75 | if (!std.ascii.eqlIgnoreCase(connection.host, criteria.host)) continue; |
| 78 | | 76 | |
| 79 | pool.acquireUnsafe(node); | 77 | pool.acquireUnsafe(connection); |
| 80 | return &node.data; | 78 | return connection; |
| 81 | } | 79 | } |
| 82 | | 80 | |
| 83 | return null; | 81 | return null; |
| 84 | } | 82 | } |
| 85 | | 83 | |
| 86 | /// Acquires an existing connection from the connection pool. This function is not threadsafe. | 84 | /// Acquires an existing connection from the connection pool. This function is not threadsafe. |
| 87 | pub fn acquireUnsafe(pool: *ConnectionPool, node: *Node) void { | 85 | pub fn acquireUnsafe(pool: *ConnectionPool, connection: *Connection) void { |
| 88 | pool.free.remove(node); | 86 | pool.free.remove(&connection.pool_node); |
| 89 | pool.free_len -= 1; | 87 | pool.free_len -= 1; |
| 90 | | 88 | |
| 91 | pool.used.append(node); | 89 | pool.used.append(&connection.pool_node); |
| 92 | } | 90 | } |
| 93 | | 91 | |
| 94 | /// Acquires an existing connection from the connection pool. This function is threadsafe. | 92 | /// Acquires an existing connection from the connection pool. This function is threadsafe. |
| 95 | pub fn acquire(pool: *ConnectionPool, node: *Node) void { | 93 | pub fn acquire(pool: *ConnectionPool, connection: *Connection) void { |
| 96 | pool.mutex.lock(); | 94 | pool.mutex.lock(); |
| 97 | defer pool.mutex.unlock(); | 95 | defer pool.mutex.unlock(); |
| 98 | | 96 | |
| 99 | return pool.acquireUnsafe(node); | 97 | return pool.acquireUnsafe(connection); |
| 100 | } | 98 | } |
| 101 | | 99 | |
| 102 | /// Tries to release a connection back to the connection pool. This function is threadsafe. | 100 | /// Tries to release a connection back to the connection pool. This function is threadsafe. |
| ... | @@ -108,38 +106,37 @@ pub const ConnectionPool = struct { | ... | @@ -108,38 +106,37 @@ pub const ConnectionPool = struct { |
| 108 | pool.mutex.lock(); | 106 | pool.mutex.lock(); |
| 109 | defer pool.mutex.unlock(); | 107 | defer pool.mutex.unlock(); |
| 110 | | 108 | |
| 111 | const node: *Node = @fieldParentPtr("data", connection); | 109 | pool.used.remove(&connection.pool_node); |
| 112 | | | |
| 113 | pool.used.remove(node); | | |
| 114 | | 110 | |
| 115 | if (node.data.closing or pool.free_size == 0) { | 111 | if (connection.closing or pool.free_size == 0) { |
| 116 | node.data.close(allocator); | 112 | connection.close(allocator); |
| 117 | return allocator.destroy(node); | 113 | return allocator.destroy(connection); |
| 118 | } | 114 | } |
| 119 | | 115 | |
| 120 | if (pool.free_len >= pool.free_size) { | 116 | if (pool.free_len >= pool.free_size) { |
| 121 | const popped = pool.free.popFirst() orelse unreachable; | 117 | const popped: *Connection = @fieldParentPtr("pool_node", pool.free.popFirst().?); |
| 122 | pool.free_len -= 1; | 118 | pool.free_len -= 1; |
| 123 | | 119 | |
| 124 | popped.data.close(allocator); | 120 | popped.close(allocator); |
| 125 | allocator.destroy(popped); | 121 | allocator.destroy(popped); |
| 126 | } | 122 | } |
| 127 | | 123 | |
| 128 | if (node.data.proxied) { | 124 | if (connection.proxied) { |
| 129 | pool.free.prepend(node); // proxied connections go to the end of the queue, always try direct connections first | 125 | // proxied connections go to the end of the queue, always try direct connections first |
| | 126 | pool.free.prepend(&connection.pool_node); |
| 130 | } else { | 127 | } else { |
| 131 | pool.free.append(node); | 128 | pool.free.append(&connection.pool_node); |
| 132 | } | 129 | } |
| 133 | | 130 | |
| 134 | pool.free_len += 1; | 131 | pool.free_len += 1; |
| 135 | } | 132 | } |
| 136 | | 133 | |
| 137 | /// Adds a newly created node to the pool of used connections. This function is threadsafe. | 134 | /// Adds a newly created node to the pool of used connections. This function is threadsafe. |
| 138 | pub fn addUsed(pool: *ConnectionPool, node: *Node) void { | 135 | pub fn addUsed(pool: *ConnectionPool, connection: *Connection) void { |
| 139 | pool.mutex.lock(); | 136 | pool.mutex.lock(); |
| 140 | defer pool.mutex.unlock(); | 137 | defer pool.mutex.unlock(); |
| 141 | | 138 | |
| 142 | pool.used.append(node); | 139 | pool.used.append(&connection.pool_node); |
| 143 | } | 140 | } |
| 144 | | 141 | |
| 145 | /// Resizes the connection pool. This function is threadsafe. | 142 | /// Resizes the connection pool. This function is threadsafe. |
| ... | @@ -170,18 +167,18 @@ pub const ConnectionPool = struct { | ... | @@ -170,18 +167,18 @@ pub const ConnectionPool = struct { |
| 170 | | 167 | |
| 171 | var next = pool.free.first; | 168 | var next = pool.free.first; |
| 172 | while (next) |node| { | 169 | while (next) |node| { |
| 173 | defer allocator.destroy(node); | 170 | const connection: *Connection = @fieldParentPtr("pool_node", node); |
| 174 | next = node.next; | 171 | next = node.next; |
| 175 | | 172 | connection.close(allocator); |
| 176 | node.data.close(allocator); | 173 | allocator.destroy(connection); |
| 177 | } | 174 | } |
| 178 | | 175 | |
| 179 | next = pool.used.first; | 176 | next = pool.used.first; |
| 180 | while (next) |node| { | 177 | while (next) |node| { |
| 181 | defer allocator.destroy(node); | 178 | const connection: *Connection = @fieldParentPtr("pool_node", node); |
| 182 | next = node.next; | 179 | next = node.next; |
| 183 | | 180 | connection.close(allocator); |
| 184 | node.data.close(allocator); | 181 | allocator.destroy(node); |
| 185 | } | 182 | } |
| 186 | | 183 | |
| 187 | pool.* = undefined; | 184 | pool.* = undefined; |
| ... | @@ -194,6 +191,9 @@ pub const Connection = struct { | ... | @@ -194,6 +191,9 @@ pub const Connection = struct { |
| 194 | /// undefined unless protocol is tls. | 191 | /// undefined unless protocol is tls. |
| 195 | tls_client: if (!disable_tls) *std.crypto.tls.Client else void, | 192 | tls_client: if (!disable_tls) *std.crypto.tls.Client else void, |
| 196 | | 193 | |
| | 194 | /// Entry in `ConnectionPool.used` or `ConnectionPool.free`. |
| | 195 | pool_node: std.DoublyLinkedList.Node, |
| | 196 | |
| 197 | /// The protocol that this connection is using. | 197 | /// The protocol that this connection is using. |
| 198 | protocol: Protocol, | 198 | protocol: Protocol, |
| 199 | | 199 | |
| ... | @@ -1326,9 +1326,8 @@ pub fn connectTcp(client: *Client, host: []const u8, port: u16, protocol: Connec | ... | @@ -1326,9 +1326,8 @@ pub fn connectTcp(client: *Client, host: []const u8, port: u16, protocol: Connec |
| 1326 | if (disable_tls and protocol == .tls) | 1326 | if (disable_tls and protocol == .tls) |
| 1327 | return error.TlsInitializationFailed; | 1327 | return error.TlsInitializationFailed; |
| 1328 | | 1328 | |
| 1329 | const conn = try client.allocator.create(ConnectionPool.Node); | 1329 | const conn = try client.allocator.create(Connection); |
| 1330 | errdefer client.allocator.destroy(conn); | 1330 | errdefer client.allocator.destroy(conn); |
| 1331 | conn.* = .{ .data = undefined }; | | |
| 1332 | | 1331 | |
| 1333 | const stream = net.tcpConnectToHost(client.allocator, host, port) catch |err| switch (err) { | 1332 | const stream = net.tcpConnectToHost(client.allocator, host, port) catch |err| switch (err) { |
| 1334 | error.ConnectionRefused => return error.ConnectionRefused, | 1333 | error.ConnectionRefused => return error.ConnectionRefused, |
| ... | @@ -1343,21 +1342,23 @@ pub fn connectTcp(client: *Client, host: []const u8, port: u16, protocol: Connec | ... | @@ -1343,21 +1342,23 @@ pub fn connectTcp(client: *Client, host: []const u8, port: u16, protocol: Connec |
| 1343 | }; | 1342 | }; |
| 1344 | errdefer stream.close(); | 1343 | errdefer stream.close(); |
| 1345 | | 1344 | |
| 1346 | conn.data = .{ | 1345 | conn.* = .{ |
| 1347 | .stream = stream, | 1346 | .stream = stream, |
| 1348 | .tls_client = undefined, | 1347 | .tls_client = undefined, |
| 1349 | | 1348 | |
| 1350 | .protocol = protocol, | 1349 | .protocol = protocol, |
| 1351 | .host = try client.allocator.dupe(u8, host), | 1350 | .host = try client.allocator.dupe(u8, host), |
| 1352 | .port = port, | 1351 | .port = port, |
| | 1352 | |
| | 1353 | .pool_node = .{}, |
| 1353 | }; | 1354 | }; |
| 1354 | errdefer client.allocator.free(conn.data.host); | 1355 | errdefer client.allocator.free(conn.host); |
| 1355 | | 1356 | |
| 1356 | if (protocol == .tls) { | 1357 | if (protocol == .tls) { |
| 1357 | if (disable_tls) unreachable; | 1358 | if (disable_tls) unreachable; |
| 1358 | | 1359 | |
| 1359 | conn.data.tls_client = try client.allocator.create(std.crypto.tls.Client); | 1360 | conn.tls_client = try client.allocator.create(std.crypto.tls.Client); |
| 1360 | errdefer client.allocator.destroy(conn.data.tls_client); | 1361 | errdefer client.allocator.destroy(conn.tls_client); |
| 1361 | | 1362 | |
| 1362 | const ssl_key_log_file: ?std.fs.File = if (std.options.http_enable_ssl_key_log_file) ssl_key_log_file: { | 1363 | const ssl_key_log_file: ?std.fs.File = if (std.options.http_enable_ssl_key_log_file) ssl_key_log_file: { |
| 1363 | const ssl_key_log_path = std.process.getEnvVarOwned(client.allocator, "SSLKEYLOGFILE") catch |err| switch (err) { | 1364 | const ssl_key_log_path = std.process.getEnvVarOwned(client.allocator, "SSLKEYLOGFILE") catch |err| switch (err) { |
| ... | @@ -1375,19 +1376,19 @@ pub fn connectTcp(client: *Client, host: []const u8, port: u16, protocol: Connec | ... | @@ -1375,19 +1376,19 @@ pub fn connectTcp(client: *Client, host: []const u8, port: u16, protocol: Connec |
| 1375 | } else null; | 1376 | } else null; |
| 1376 | errdefer if (ssl_key_log_file) |key_log_file| key_log_file.close(); | 1377 | errdefer if (ssl_key_log_file) |key_log_file| key_log_file.close(); |
| 1377 | | 1378 | |
| 1378 | conn.data.tls_client.* = std.crypto.tls.Client.init(stream, .{ | 1379 | conn.tls_client.* = std.crypto.tls.Client.init(stream, .{ |
| 1379 | .host = .{ .explicit = host }, | 1380 | .host = .{ .explicit = host }, |
| 1380 | .ca = .{ .bundle = client.ca_bundle }, | 1381 | .ca = .{ .bundle = client.ca_bundle }, |
| 1381 | .ssl_key_log_file = ssl_key_log_file, | 1382 | .ssl_key_log_file = ssl_key_log_file, |
| 1382 | }) catch return error.TlsInitializationFailed; | 1383 | }) catch return error.TlsInitializationFailed; |
| 1383 | // This is appropriate for HTTPS because the HTTP headers contain | 1384 | // This is appropriate for HTTPS because the HTTP headers contain |
| 1384 | // the content length which is used to detect truncation attacks. | 1385 | // the content length which is used to detect truncation attacks. |
| 1385 | conn.data.tls_client.allow_truncation_attacks = true; | 1386 | conn.tls_client.allow_truncation_attacks = true; |
| 1386 | } | 1387 | } |
| 1387 | | 1388 | |
| 1388 | client.connection_pool.addUsed(conn); | 1389 | client.connection_pool.addUsed(conn); |
| 1389 | | 1390 | |
| 1390 | return &conn.data; | 1391 | return conn; |
| 1391 | } | 1392 | } |
| 1392 | | 1393 | |
| 1393 | pub const ConnectUnixError = Allocator.Error || std.posix.SocketError || error{NameTooLong} || std.posix.ConnectError; | 1394 | pub const ConnectUnixError = Allocator.Error || std.posix.SocketError || error{NameTooLong} || std.posix.ConnectError; |
Andrew Kelley <andrew@ziglang.org> 2025-04-03 15:57:35-07:00
lib