ConnectionPool

pub const ConnectionPool = struct

File

Code

pub const ConnectionPool = struct {
    mutex: Io.Mutex = .init,
    /// Open connections that are currently in use.
    used: std.DoublyLinkedList = .{},
    /// Open connections that are not currently in use.
    free: std.DoublyLinkedList = .{},
    free_len: usize = 0,
    free_size: usize = 32,

    /// The criteria for a connection to be considered a match.
    pub const Criteria = struct {
        host: HostName,
        port: u16,
        protocol: Protocol,
    };

    /// Finds and acquires a connection from the connection pool matching the criteria.
    /// If no connection is found, null is returned.
    ///
    /// Threadsafe.
    pub fn findConnection(pool: *ConnectionPool, io: Io, criteria: Criteria) Io.Cancelable!?*Connection {
        try pool.mutex.lock(io);
        defer pool.mutex.unlock(io);

        var next = pool.free.last;
        while (next) |node| : (next = node.prev) {
            const connection: *Connection = @alignCast(@fieldParentPtr("pool_node", node));
            if (connection.protocol != criteria.protocol) continue;
            if (connection.port != criteria.port) continue;

            // Domain names are case-insensitive (RFC 5890, Section 2.3.2.4)
            if (!connection.host().eql(criteria.host)) continue;

            pool.acquireUnsafe(connection);
            return connection;
        }

        return null;
    }

    /// Acquires an existing connection from the connection pool. This function is not threadsafe.
    pub fn acquireUnsafe(pool: *ConnectionPool, connection: *Connection) void {
        pool.free.remove(&connection.pool_node);
        pool.free_len -= 1;

        pool.used.append(&connection.pool_node);
    }

    /// Acquires an existing connection from the connection pool. This function is threadsafe.
    pub fn acquire(pool: *ConnectionPool, io: Io, connection: *Connection) Io.Cancelable!void {
        try pool.mutex.lock(io);
        defer pool.mutex.unlock(io);

        return pool.acquireUnsafe(connection);
    }

    /// Tries to release a connection back to the connection pool.
    /// If the connection is marked as closing, it will be closed instead.
    ///
    /// Threadsafe.
    pub fn release(pool: *ConnectionPool, connection: *Connection, io: Io) void {
        pool.mutex.lockUncancelable(io);
        defer pool.mutex.unlock(io);

        pool.used.remove(&connection.pool_node);

        if (connection.closing or pool.free_size == 0) return connection.destroy(io);

        if (pool.free_len >= pool.free_size) {
            const popped: *Connection = @alignCast(@fieldParentPtr("pool_node", pool.free.popFirst().?));
            pool.free_len -= 1;

            popped.destroy(io);
        }

        if (connection.proxied) {
            // proxied connections go to the end of the queue, always try direct connections first
            pool.free.prepend(&connection.pool_node);
        } else {
            pool.free.append(&connection.pool_node);
        }

        pool.free_len += 1;
    }

    /// Adds a newly created node to the pool of used connections. This function is threadsafe.
    pub fn addUsed(pool: *ConnectionPool, io: Io, connection: *Connection) Io.Cancelable!void {
        try pool.mutex.lock(io);
        defer pool.mutex.unlock(io);

        pool.used.append(&connection.pool_node);
    }

    /// Resizes the connection pool.
    ///
    /// If the new size is smaller than the current size, then idle connections will be closed until the pool is the new size.
    ///
    /// Threadsafe.
    pub fn resize(pool: *ConnectionPool, io: Io, new_size: usize) Io.Cancelable!void {
        try pool.mutex.lock(io);
        defer pool.mutex.unlock(io);

        while (pool.free_len > new_size) {
            const popped: *Connection = @alignCast(@fieldParentPtr("pool_node", pool.free.popFirst().?));
            pool.free_len -= 1;

            popped.destroy(io);
        }

        pool.free_size = new_size;
    }

    /// Frees the connection pool and closes all connections within.
    ///
    /// All future operations on the connection pool will deadlock.
    ///
    /// Threadsafe.
    pub fn deinit(pool: *ConnectionPool, io: Io) void {
        pool.mutex.lockUncancelable(io);

        var next = pool.free.first;
        while (next) |node| {
            const connection: *Connection = @alignCast(@fieldParentPtr("pool_node", node));
            next = node.next;
            connection.destroy(io);
        }

        next = pool.used.first;
        while (next) |node| {
            const connection: *Connection = @alignCast(@fieldParentPtr("pool_node", node));
            next = node.next;
            connection.destroy(io);
        }

        pool.* = undefined;
    }
}