Milvex.ConnectionPool (milvex v0.14.2)

Copy Markdown

Round-robin pool of Milvex.Connection processes.

Each pooled connection maintains its own gRPC channel, i.e. its own HTTP/2 connection. HTTP/2 servers cap the number of concurrent streams per connection (SETTINGS_MAX_CONCURRENT_STREAMS), and strict clients such as the Mint adapter reject requests over that limit with :too_many_concurrent_requests. Pooling multiplies the effective concurrent-stream budget by the pool size.

A pool pid or registered name can be used anywhere a connection is expected:

{:ok, pool} = Milvex.ConnectionPool.start_link(host: "localhost", pool_size: 4)
{:ok, channel, config} = Milvex.Connection.get_channel(pool)
{:ok, results} = Milvex.search(pool, "movies", vectors, vector_field: "embedding")

The usual entry point is Milvex.Connection.start_link/1 with a :pool_size option greater than 1, which delegates here.

Behavior

Channel lookup never goes through the pool process. Each pooled connection publishes its channel to a shared ETS table when it connects and removes it when it disconnects. Callers pick a channel round-robin in their own process using an :atomics counter, so get_channel/2 is lock-free and the pool process is never a bottleneck.

  • get_channel/2 picks connections round-robin. If the picked connection is not currently connected, the remaining connections are tried before returning a retriable :not_connected error.
  • Each connection reconnects independently with the backoff configured via Milvex.Config.
  • connected?/2 returns true if at least one pooled connection is connected.
  • disconnect/1 stops the pool and all pooled connections.

Summary

Types

entry()

Lock-free pool entry published via :persistent_term.

t()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

connected?(pool, opts \\ [])

Checks if at least one pooled connection is established.

disconnect(pool)

Disconnects all pooled connections and stops the pool.

get_channel(pool, opts \\ [])

Gets a gRPC channel from the pool, round-robin.

start_link(opts)

Starts a pool of connections to a Milvus server.

Types

entry()

@type entry() :: %{
  table: :ets.tid(),
  counter: :atomics.atomics_ref(),
  size: pos_integer(),
  config: Milvex.Config.t()
}

Lock-free pool entry published via :persistent_term.

t()

@type t() :: %Milvex.ConnectionPool{entry: entry(), supervisor: pid() | nil}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

connected?(pool, opts \\ [])

@spec connected?(
  GenServer.server(),
  keyword()
) :: boolean()

Checks if at least one pooled connection is established.

disconnect(pool)

@spec disconnect(GenServer.server()) :: :ok

Disconnects all pooled connections and stops the pool.

get_channel(pool, opts \\ [])

@spec get_channel(
  GenServer.server(),
  keyword()
) ::
  {:ok, Milvex.Connection.channel(), Milvex.Config.t()}
  | {:error, Milvex.Error.t()}

Gets a gRPC channel from the pool, round-robin.

Lock-free: reads the pool's shared channel table directly from the caller process without going through the pool process.

Returns {:ok, channel, config} if any pooled connection is connected, or {:error, error} otherwise. Exits with :noproc if the pool is not running.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

Starts a pool of connections to a Milvus server.

Options

  • :name - Optional name to register the pool process
  • :pool_size - Number of connections to start (default: 1)
  • All other options are passed to Milvex.Config.parse/1