pool

package
v3.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 8, 2026 License: BSD-2-Clause Imports: 11 Imported by: 0

Documentation

Overview

Package pool provides methods to work with a Tarantool cluster considering master discovery.

Main features:

- Return available connection from pool according to round-robin strategy.

- Automatic master discovery by mode parameter.

Since: 1.6.0

Index

Examples

Constants

View Source 
const (
	// LogMsgConnectFailed is a log message emitted when a connection
	// attempt to a pool instance fails.
	LogMsgConnectFailed = "connect failed"
	// LogMsgInitWatchersFailed is a log message emitted when initialization
	// of watchers for a pool instance connection fails.
	LogMsgInitWatchersFailed = "failed to initialize watchers"
	// LogMsgStoringConnectionCanceled is a log message emitted when storing
	// a discovered connection is canceled by the Handler.Discovered callback.
	LogMsgStoringConnectionCanceled = "storing connection canceled"
	// LogMsgDeactivatingFailed is a log message emitted when the
	// Handler.Deactivated callback returns an error during connection
	// deactivation.
	LogMsgDeactivatingFailed = "handler deactivation failed"
	// LogMsgStoringConnectionFailed is a log message emitted when storing
	// a connection fails due to a role detection error.
	LogMsgStoringConnectionFailed = "storing connection failed"
	// LogMsgReconnectFailed is a log message emitted when a reconnection
	// attempt for a pool instance fails.
	LogMsgReconnectFailed = "reconnect failed"
	// LogMsgReopenFailed is a log message emitted when reopening a
	// connection for a pool instance fails.
	LogMsgReopenFailed = "reopen connection failed"
)
View Source 
const (
	// LogKeyInstance is a log key for a pool instance name.
	LogKeyInstance = "instance"
	// LogKeyError is a log key for an error value.
	// It intentionally mirrors tarantool.LogKeyError to avoid an
	// inter-package dependency.
	LogKeyError = "error"
)

Variables

View Source 
var (
	ErrWrongCheckTimeout = errors.New("wrong check timeout, must be greater than 0")
	ErrTooManyArgs       = errors.New("too many arguments")
	ErrIncorrectResponse = errors.New("incorrect response format")
	ErrIncorrectStatus   = errors.New("incorrect instance status: status should be `running`")
	ErrNoRwInstance      = errors.New("can't find rw instance in pool")
	ErrNoRoInstance      = errors.New("can't find ro instance in pool")
	ErrNoHealthyInstance = errors.New("can't find healthy instance in pool")
	ErrExists            = errors.New("endpoint exists")
	ErrClosed            = errors.New("pool is closed")
	ErrUnknownRequest    = errors.New("the passed connected request doesn't belong to " +
		"the current connection pool")
	ErrContextCanceled = errors.New("operation was canceled")
	// ErrOptsReconnect is returned if tarantool.Opts.Reconnect is set for an
	// instance connection. It must not be used with ConnectionPool: the pool
	// itself manages reconnection via pool.Opts.CheckTimeout, so an internal
	// reconnect creates a conflict with the pool's own reconnection logic.
	ErrOptsReconnect = errors.New(
		"tarantool.Opts.Reconnect must not be used with ConnectionPool")
	// ErrOptsMaxReconnects is returned if tarantool.Opts.MaxReconnects is set
	// for an instance connection. It must not be used with ConnectionPool: the
	// pool itself manages connection lifetime, so an internal reconnect limit
	// permanently closes a Connection while the pool will create a new one
	// anyway.
	ErrOptsMaxReconnects = errors.New(
		"tarantool.Opts.MaxReconnects must not be used with ConnectionPool")
	// ErrOptsNotify is returned if tarantool.Opts.Notify is set for an instance
	// connection. It must not be used with ConnectionPool: events from an
	// individual connection are misleading in a pool context (e.g., Disconnected
	// does not mean the endpoint is unavailable, since the pool may have already
	// reconnected); use pool.ConnectionHandler to track endpoint availability
	// instead.
	ErrOptsNotify = errors.New(
		"tarantool.Opts.Notify must not be used with ConnectionPool")
)

Functions

This section is empty.

Types

type ConnectorAdapter 

type ConnectorAdapter struct {
	// contains filtered or unexported fields
}

ConnectorAdapter allows to use Pooler as Connector.

Example 
connPool, err := examplePool(testRoles, connOpts)
if err != nil {
	fmt.Println(err)
}
defer func() { _ = connPool.Close() }()

adapter := pool.NewConnectorAdapter(connPool, pool.ModeRW)
var connector tarantool.Connector = adapter

// Ping an ModeRW instance to check connection.
data, err := connector.Do(tarantool.NewPingRequest()).Get()
fmt.Println("Ping Data", data)
fmt.Println("Ping Error", err)

Output:
Ping Data []
Ping Error <nil>

func NewConnectorAdapter 

func NewConnectorAdapter(pool Pooler, mode Mode) *ConnectorAdapter

NewConnectorAdapter creates a new ConnectorAdapter object for a pool and with a mode. All requests to the pool will be executed in the specified mode.

func (*ConnectorAdapter) Close 

func (c *ConnectorAdapter) Close() error

Close reports if the connector is closed by user or all connections in the specified mode closed.

func (*ConnectorAdapter) ConfiguredTimeout 

func (c *ConnectorAdapter) ConfiguredTimeout() time.Duration

ConfiguredTimeout returns a timeout from connections config.

func (*ConnectorAdapter) ConnectedNow 

func (c *ConnectorAdapter) ConnectedNow() bool

ConnectedNow reports if connections is established at the moment.

func (*ConnectorAdapter) Do 

func (c *ConnectorAdapter) Do(req tarantool.Request) tarantool.Future

Do performs a request asynchronously on the connection.

func (*ConnectorAdapter) NewPrepared 

func (c *ConnectorAdapter) NewPrepared(expr string) (*tarantool.Prepared, error)

NewPrepared passes a sql statement to Tarantool for preparation synchronously.

func (*ConnectorAdapter) NewStream 

func (c *ConnectorAdapter) NewStream() (*tarantool.Stream, error)

NewStream creates new Stream object for connection.

Since v. 2.10.0, Tarantool supports streams and interactive transactions over them. To use interactive transactions, memtx_use_mvcc_engine box option should be set to true.

Since 1.7.0.

func (*ConnectorAdapter) NewWatcher 

func (c *ConnectorAdapter) NewWatcher(key string,
	callback tarantool.WatchCallback) (tarantool.Watcher, error)

NewWatcher creates new Watcher object for the pool.

Since 1.10.0.

type Handler 

type Handler interface {
	// Discovered is called when a connection with a role has been detected
	// (for the first time or when a role of a connection has been changed),
	// but is not yet available to send requests. It allows for a client to
	// initialize the connection before using it in a pool.
	//
	// The client code may cancel adding a connection to the pool. The client
	// need to return an error from the Discovered call for that. In this case
	// the pool will close connection and will try to reopen it later.
	Discovered(name string, conn *tarantool.Connection, role Role) error
	// Deactivated is called when a connection with a role has become
	// unavaileble to send requests. It happens if the connection is closed or
	// the connection role is switched.
	//
	// So if a connection switches a role, a pool calls:
	// Deactivated() + Discovered().
	//
	// Deactivated will not be called if a previous Discovered() call returns
	// an error. Because in this case, the connection does not become available
	// for sending requests.
	Deactivated(name string, conn *tarantool.Connection, role Role) error
}

Handler provides callbacks for components interested in handling changes of connections in a Pool.

type Info 

type Info struct {
	ConnectedNow bool
	Role         Role
	Instance     Instance
}

Info structure for information about connection statuses:

- ConnectedNow reports if connection is established at the moment.

- Role reports master/replica role of instance.

type Instance 

type Instance struct {
	// Name is an instance name. The name must be unique.
	Name string
	// Dialer will be used to create a connection to the instance.
	Dialer tarantool.Dialer
	// Opts configures a connection to the instance. tarantool.Opts.Reconnect,
	// tarantool.Opts.MaxReconnects and tarantool.Opts.Notify must not be set:
	// the pool manages reconnection and connection lifetime itself, and
	// individual connection events are misleading in a pool context. Use
	// pool.ConnectionHandler to track endpoint availability.
	Opts tarantool.Opts
}

Instance describes a single instance configuration in the pool.

type Mode 

type Mode uint32

Mode is a request mode for a connection pool.

Default mode for each request table: 

  Request   Default mode
---------- --------------
| call    | no default  |
| eval    | no default  |
| execute | no default  |
| ping    | no default  |
| insert  | ModeRW      |
| delete  | ModeRW      |
| replace | ModeRW      |
| update  | ModeRW      |
| upsert  | ModeRW      |
| select  | ModeAny     |
| get     | ModeAny     |

const (
	ModeAny      Mode = iota // The request can be executed on any instance (master or replica).
	ModeRW                   // The request can only be executed on master.
	ModeRO                   // The request can only be executed on replica.
	ModePreferRW             // If there is one, otherwise fallback to a writeable one (master).
	ModePreferRO             // If there is one, otherwise fallback to a read only one (replica).
)

type Opts 

type Opts struct {
	// Timeout for timer to reopen connections that have been closed by some
	// events and to relocate connection between subpools if ro/rw role has
	// been updated.
	CheckTimeout time.Duration
	// Handler provides an ability to handle connection updates.
	Handler Handler
	// Logger is a structured logger for the pool. If nil, logs are
	// discarded. If non-nil, the logger is automatically wrapped with
	// WithGroup("tarantool.pool") to namespace library attributes.
	//
	// If a logger is not set in connection opts, the pool logger will
	// be set for each connection automatically.
	Logger *slog.Logger
}

Opts provides additional options (configurable via NewWithOpts).

type Pool 

type Pool struct {
	// contains filtered or unexported fields
}

Pool is a connection pool for a Tarantool cluster.

Main features:

- Return available connection from pool according to round-robin strategy.

- Automatic master discovery by mode parameter.

func New 

func New(ctx context.Context, instances []Instance) (*Pool, error)

New creates pool for instances with specified instances. Instances must have unique names.

func NewWithOpts 

func NewWithOpts(ctx context.Context, instances []Instance,
	opts Opts) (*Pool, error)

NewWithOpts creates pool for instances with specified instances and opts. Instances must have unique names.

func (*Pool) Add 

func (p *Pool) Add(ctx context.Context, instance Instance) error

Add adds a new instance into the pool. The pool will try to connect to the instance later if it is unable to establish a connection.

The function may return an error and don't add the instance into the pool if the context has been cancelled or on concurrent Close()/CloseGraceful() call.

func (*Pool) Close 

func (p *Pool) Close() error

Close closes connections in the Pool.

func (*Pool) CloseGraceful 

func (p *Pool) CloseGraceful() error

CloseGraceful closes connections in the Pool gracefully. It waits for all requests to complete.

Example (Force)

ExamplePool_CloseGraceful_force demonstrates how to force close a connection pool with graceful close in progress after a while. 

connPool, err := examplePool(testRoles, connOpts)
if err != nil {
	fmt.Println(err)
	return
}

eval := `local fiber = require('fiber')
	local time = ...
	fiber.sleep(time)
`
req := tarantool.NewEvalRequest(eval).Args([]any{10})
fut := connPool.Do(req, pool.ModeAny)

done := make(chan struct{})
go func() {
	defer close(done)

	err := connPool.CloseGraceful()
	if err != nil {
		fmt.Println(err)
		return
	}
	fmt.Println("Pool.CloseGraceful() done!")
}()

select {
case <-done:
case <-time.After(3 * time.Second):
	fmt.Println("Force Pool.Close()!")
	_ = connPool.Close()
}
<-done

fmt.Println("Result:")
fmt.Println(fut.Get())

Output:
Force Pool.Close()!
Pool.CloseGraceful() done!
Result:
[] connection closed: connection closed by client

func (*Pool) ConfiguredTimeout 

func (p *Pool) ConfiguredTimeout(mode Mode) (time.Duration, error)

ConfiguredTimeout gets timeout of current connection.

func (*Pool) ConnectedNow 

func (p *Pool) ConnectedNow(mode Mode) (bool, error)

ConnectedNow gets connected status of pool.

func (*Pool) Do 

func (p *Pool) Do(req tarantool.Request, userMode Mode) tarantool.Future

Do sends the request and returns a future. For requests that belong to the only one connection (e.g. Unprepare or ExecutePrepared) the argument of type Mode is unused.

Example 
connPool, err := examplePool(testRoles, connOpts)
if err != nil {
	fmt.Println(err)
}
defer func() { _ = connPool.Close() }()

modes := []pool.Mode{
	pool.ModeAny,
	pool.ModeRW,
	pool.ModeRO,
	pool.ModePreferRW,
	pool.ModePreferRO,
}
for _, m := range modes {
	// It could be any request object.
	req := tarantool.NewPingRequest()
	_, err := connPool.Do(req, m).Get()
	fmt.Println("Ping Error", err)
}

Output:
Ping Error <nil>
Ping Error <nil>
Ping Error <nil>
Ping Error <nil>
Ping Error <nil>

func (*Pool) DoOn 

func (p *Pool) DoOn(req tarantool.Request, name string) tarantool.Future

DoOn sends the request into a target instance and returns a future.

func (*Pool) Info 

func (p *Pool) Info() map[string]Info

Info gets information of connections (connected status, ro/rw role).

func (*Pool) NewPrepared 

func (p *Pool) NewPrepared(expr string, userMode Mode) (*tarantool.Prepared, error)

NewPrepared passes a sql statement to Tarantool for preparation synchronously.

Example 
connPool, err := examplePool(testRoles, connOpts)
if err != nil {
	fmt.Println(err)
}
defer func() { _ = connPool.Close() }()

stmt, err := connPool.NewPrepared("SELECT 1", pool.ModeAny)
if err != nil {
	fmt.Println(err)
}

executeReq := tarantool.NewExecutePreparedRequest(stmt)
unprepareReq := tarantool.NewUnprepareRequest(stmt)

_, err = connPool.Do(executeReq, pool.ModeAny).Get()
if err != nil {
	fmt.Printf("Failed to execute prepared stmt")
}
_, err = connPool.Do(unprepareReq, pool.ModeAny).Get()
if err != nil {
	fmt.Printf("Failed to prepare")
}

func (*Pool) NewStream 

func (p *Pool) NewStream(userMode Mode) (*tarantool.Stream, error)

NewStream creates new Stream object for connection selected by userMode from pool.

Since v. 2.10.0, Tarantool supports streams and interactive transactions over them. To use interactive transactions, memtx_use_mvcc_engine box option should be set to true.

Since 1.7.0.

func (*Pool) NewWatcher 

func (p *Pool) NewWatcher(key string,
	callback tarantool.WatchCallback, mode Mode) (tarantool.Watcher, error)

NewWatcher creates a new Watcher object for the connection pool. A watcher could be created only for instances with the support.

The behavior is same as if Connection.NewWatcher() called for each connection with a suitable mode.

Keep in mind that garbage collection of a watcher handle doesn’t lead to the watcher’s destruction. In this case, the watcher remains registered. You need to call Unregister() directly.

Unregister() guarantees that there will be no the watcher's callback calls after it, but Unregister() call from the callback leads to a deadlock.

See: https://www.tarantool.io/en/doc/latest/reference/reference_lua/box_events/#box-watchers

Since 1.10.0.

Example 
const key = "foo"
const value = "bar"

connPool, err := examplePool(testRoles, connOpts)
if err != nil {
	fmt.Println(err)
}
defer func() { _ = connPool.Close() }()

callback := func(event tarantool.WatchEvent) {
	fmt.Printf("event connection: %s\n", event.Conn.Addr())
	fmt.Printf("event key: %s\n", event.Key)
	fmt.Printf("event value: %v\n", event.Value)
}
mode := pool.ModeAny
watcher, err := connPool.NewWatcher(key, callback, mode)
if err != nil {
	fmt.Printf("Unexpected error: %s\n", err)
	return
}
defer watcher.Unregister()

_, _ = connPool.Do(tarantool.NewBroadcastRequest(key).Value(value), mode).Get()
time.Sleep(time.Second)

func (*Pool) Remove 

func (p *Pool) Remove(name string) error

Remove removes an endpoint with the name from the pool. The call closes an active connection gracefully.

type Pooler 

type Pooler interface {
	TopologyEditor

	ConnectedNow(mode Mode) (bool, error)
	Close() error
	// CloseGraceful closes connections in the Pool gracefully. It waits
	// for all requests to complete.
	CloseGraceful() error
	ConfiguredTimeout(mode Mode) (time.Duration, error)
	NewPrepared(expr string, mode Mode) (*tarantool.Prepared, error)
	NewStream(mode Mode) (*tarantool.Stream, error)
	NewWatcher(key string, callback tarantool.WatchCallback,
		mode Mode) (tarantool.Watcher, error)
	Do(req tarantool.Request, mode Mode) (fut tarantool.Future)
}

Pooler is the interface that must be implemented by a connection pool.

type Role 

type Role uint32

Role describes a role of an instance by its mode. 

const (
	// RoleUnknown - the connection pool was unable to detect the instance mode.
	RoleUnknown Role = iota // unknown
	// RoleMaster - the instance is in read-write mode.
	RoleMaster // master
	// RoleReplica - the instance is in read-only mode.
	RoleReplica // replica
)

func (Role) String 

func (i Role) String() string

type TopologyEditor 

type TopologyEditor interface {
	Add(ctx context.Context, instance Instance) error
	Remove(name string) error
}

TopologyEditor is the interface that must be implemented by a connection pool. It describes edit topology methods.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.