connector.go

// Copyright (C) 2021 Storj Labs, Inc.
// See LICENSE for copying information.

package server

import (
	"context"
	"crypto/tls"

	"github.com/zeebo/errs"

	"storj.io/common/memory"
	"storj.io/common/rpc"
	"storj.io/storj/private/quic"
)

// HybridConnector implements a dialer that creates a connection using either
// quic or tcp.
type HybridConnector struct {
	quic *quic.Connector
	tcp  *rpc.TCPConnector
}

// NewDefaultHybridConnector instantiates a new instance of HybridConnector with
// provided quic and tcp connectors.
// If a nil value is provided for either connector, a default connector will be
// created instead.
// See func DialContext for more details.
func NewDefaultHybridConnector(qc *quic.Connector, tc *rpc.TCPConnector) HybridConnector {
	if qc == nil {
		connector := quic.NewDefaultConnector(nil)
		qc = &connector
	}
	if tc == nil {
		connector := rpc.NewDefaultTCPConnector(nil)
		tc = &connector
	}

	return HybridConnector{
		quic: qc,
		tcp:  tc,
	}
}

// DialContext creates a connection using either quic or tcp.
// It tries to dial through both connector and returns the first established
// connection. If both connections are established, it will return quic connection.
// An error is returned if both connector failed.
func (c HybridConnector) DialContext(ctx context.Context, tlsConfig *tls.Config, address string) (_ rpc.ConnectorConn, err error) {
	defer mon.Task()(&ctx)(&err)

	if tlsConfig == nil {
		return nil, Error.New("tls config is not set")
	}

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	var tcpConn, quicConn rpc.ConnectorConn
	errChan := make(chan error)
	readyChan := make(chan struct{})

	go func() {
		var err error
		quicConn, err = c.quic.DialContext(ctx, tlsConfig.Clone(), address)
		if err != nil {
			errChan <- err
			return
		}

		readyChan <- struct{}{}
	}()
	go func() {
		var err error
		tcpConn, err = c.tcp.DialContext(ctx, tlsConfig.Clone(), address)
		if err != nil {
			errChan <- err
			return
		}

		readyChan <- struct{}{}
	}()

	var errors []error
	var numFinished int
	// makre sure both dial is finished either with an established connection or
	// an error. It allows us to appropriately close tcp connection if both
	// connections are ready around the same time
	for numFinished < 2 {
		select {
		case <-readyChan:
			numFinished++
			// if one connection is ready, we want to cancel the other dial if
			// the connection isn't ready
			cancel()
		case err := <-errChan:
			numFinished++
			errors = append(errors, err)
		}
	}

	// we want to prioritize quic conn if both connections are available
	if quicConn != nil {
		if tcpConn != nil {
			_ = tcpConn.Close()
		}

		mon.Event("hybrid_connector_established_quic_connection")
		return quicConn, nil
	}

	if tcpConn != nil {
		mon.Event("hybrid_connector_established_tcp_connection")
		return tcpConn, nil
	}

	mon.Event("hybrid_connector_established_no_connection")

	return nil, errs.Combine(errors...)
}

// SetQUICTransferRate returns a connector with the given transfer rate.
func (c *HybridConnector) SetQUICTransferRate(rate memory.Size) {
	updated := c.quic.SetTransferRate(rate)
	c.quic = &updated
}

// SetTCPTransferRate returns a connector with the given transfer rate.
func (c *HybridConnector) SetTCPTransferRate(rate memory.Size) {
	c.tcp.TransferRate = rate
}