bundle

Documentation

Overview

Package bundle provides multi-streaming transport with the functionality to dynamically (un)register receive endpoints, establish long-lived flows, and more.

• Copyright (c) 2018-2024, NVIDIA CORPORATION. All rights reserved.

Package bundle provides multi-streaming transport with the functionality to dynamically (un)register receive endpoints, establish long-lived flows, and more.

• Copyright (c) 2018-2023, NVIDIA CORPORATION. All rights reserved.

Index

• func IsErrDestinationMissing(e error) bool
• type Args
• type DataMover
  • func NewDataMover(trname string, recvCB transport.RecvObj, owt cmn.OWT, extra Extra) (*DataMover, error)
  • func (dm *DataMover) ACK(hdr *transport.ObjHdr, cb transport.ObjSentCB, tsi *meta.Snode) error
  • func (dm *DataMover) Abort()
  • func (dm *DataMover) Bcast(obj *transport.Obj, roc cos.ReadOpenCloser) error
  • func (dm *DataMover) Close(err error)
  • func (dm *DataMover) GetXact() core.Xact
  • func (dm *DataMover) NetC() string
  • func (dm *DataMover) NetD() string
  • func (dm *DataMover) OWT() cmn.OWT
  • func (dm *DataMover) Open()
  • func (dm *DataMover) Quiesce(d time.Duration) core.QuiRes
  • func (dm *DataMover) RegRecv() (err error)
  • func (dm *DataMover) Send(obj *transport.Obj, roc cos.ReadOpenCloser, tsi *meta.Snode) (err error)
  • func (dm *DataMover) SetXact(xctn core.Xact)
  • func (dm *DataMover) String() string
  • func (dm *DataMover) UnregRecv()
• type ErrDestinationMissing
  • func (e *ErrDestinationMissing) Error() string
• type Extra
• type Stats
• type Streams
  • func New(cl transport.Client, args Args) (sb *Streams)
  • func (sb *Streams) Abort()
  • func (sb *Streams) Close(gracefully bool)
  • func (sb *Streams) GetStats() Stats
  • func (sb *Streams) ListenSmapChanged()
  • func (sb *Streams) Resync()
  • func (sb *Streams) Send(obj *transport.Obj, roc cos.ReadOpenCloser, nodes ...*meta.Snode) (err error)
  • func (sb *Streams) Smap() *meta.Smap
  • func (sb *Streams) String() string
  • func (sb *Streams) Trname() string
  • func (sb *Streams) UsePDU() bool

Functions

func IsErrDestinationMissing

func IsErrDestinationMissing(e error) bool

Types

type Args

type Args struct {
	Extra        *transport.Extra // additional parameters
	Net          string           // one of cmn.KnownNetworks, empty defaults to cmn.NetIntraData
	Trname       string           // transport endpoint name
	Ntype        int              // core.Target (0) by default
	Multiplier   int              // so-many TCP connections per Rx endpoint, with round-robin
	ManualResync bool             // auto-resync by default
}

type DataMover

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

func NewDataMover

func NewDataMover(trname string, recvCB transport.RecvObj, owt cmn.OWT, extra Extra) (*DataMover, error)

func (*DataMover) ACK

func (dm *DataMover) ACK(hdr *transport.ObjHdr, cb transport.ObjSentCB, tsi *meta.Snode) error

func (*DataMover) Abort

func (dm *DataMover) Abort()

func (*DataMover) Bcast

func (dm *DataMover) Bcast(obj *transport.Obj, roc cos.ReadOpenCloser) error

func (*DataMover) Close

func (dm *DataMover) Close(err error)

func (*DataMover) GetXact

func (dm *DataMover) GetXact() core.Xact

func (*DataMover) NetC

func (dm *DataMover) NetC() string

func (*DataMover) NetD

func (dm *DataMover) NetD() string

func (*DataMover) OWT

func (dm *DataMover) OWT() cmn.OWT

func (*DataMover) Open

func (dm *DataMover) Open()

func (*DataMover) Quiesce

func (dm *DataMover) Quiesce(d time.Duration) core.QuiRes

quiesce *local* Rx

func (*DataMover) RegRecv

func (dm *DataMover) RegRecv() (err error)

register user's receive-data (and, optionally, receive-ack) wrappers

func (*DataMover) Send

func (dm *DataMover) Send(obj *transport.Obj, roc cos.ReadOpenCloser, tsi *meta.Snode) (err error)

func (*DataMover) SetXact

func (dm *DataMover) SetXact(xctn core.Xact)

xaction that drives and utilizes this data mover

func (*DataMover) String

func (dm *DataMover) String() string

func (*DataMover) UnregRecv

func (dm *DataMover) UnregRecv()

type ErrDestinationMissing

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

func (*ErrDestinationMissing) Error

func (e *ErrDestinationMissing) Error() string

type Extra

type Extra struct {
	RecvAck     transport.RecvObj
	Config      *cmn.Config
	Compression string
	Multiplier  int
	SizePDU     int32
	MaxHdrSize  int32
}

additional (and optional) params for new data mover

type Stats

type Stats map[string]*transport.Stats // by DaemonID

type Streams

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

func New

func New(cl transport.Client, args Args) (sb *Streams)

func (*Streams) Abort

func (sb *Streams) Abort()

func (*Streams) Close

func (sb *Streams) Close(gracefully bool)

Close closes all contained streams and unregisters the bundle from Smap listeners; graceful=true blocks until all pending objects get completed (for "completion", see transport/README.md)

func (*Streams) GetStats

func (sb *Streams) GetStats() Stats

func (*Streams) ListenSmapChanged

func (sb *Streams) ListenSmapChanged()

keep streams to => (clustered nodes as per rxNodeType) in sync at all times

func (*Streams) Resync

func (sb *Streams) Resync()

Resync streams asynchronously is a slowpath; is called under lock; NOTE: calls stream.Stop()

func (*Streams) Send

func (sb *Streams) Send(obj *transport.Obj, roc cos.ReadOpenCloser, nodes ...*meta.Snode) (err error)

when (nodes == nil) transmit via all established streams in a bundle otherwise, restrict to the specified subset (nodes)

func (*Streams) Smap

func (sb *Streams) Smap() *meta.Smap

func (*Streams) String

func (sb *Streams) String() string

func (*Streams) Trname

func (sb *Streams) Trname() string

func (*Streams) UsePDU

func (sb *Streams) UsePDU() bool