@@ -58,10 +58,28 @@ const sizeofh = int(unsafe.Sizeof(Obj{}))
5858// Correct usage: consume ObjHdr synchronously and inline inside the recv() callback.
5959
6060type (
61- // advanced usage: additional stream control
61+ // object-sent callback that has the following signature can optionally be defined on a:
62+ // a) per-stream basis (via NewStream constructor - see Extra struct above)
63+ // b) for a given object that is being sent (for instance, to support a call-per-batch semantics)
64+ // Naturally, object callback "overrides" the per-stream one: when object callback is defined
65+ // (i.e., non-nil), the stream callback is ignored/skipped.
66+ // NOTE: if defined, the callback executes asynchronously as far as the sending part is concerned
67+ SentCB func (* ObjHdr , io.ReadCloser , any , error )
68+
69+ // scope: stream
70+ TermedCB func (error )
71+
72+ // usage and scope:
73+ // - entire stream's lifetime (all Send() calls)
74+ // - additional stream control
75+ // - global or optional params (to override defaults)
76+ Parent struct {
77+ Xact core.Xact // sender ID; abort
78+ SentCB SentCB // to free SGLs, close files, etc. cleanup
79+ TermedCB TermedCB // when err-ed
80+ }
6281 Extra struct {
63- Xact core.Xact // usage: sender ID; abort
64- Callback ObjSentCB // typical usage: to free SGLs, close files
82+ Parent * Parent
6583 Config * cmn.Config // (to optimize-out GCO.Get())
6684 Compression string // see CompressAlways, etc. enum
6785 IdleTeardown time.Duration // when exceeded, causes PUT to terminate (and to renew upon the very next send)
@@ -70,11 +88,7 @@ type (
7088 MaxHdrSize int32 // overrides config.Transport.MaxHeaderSize
7189 }
7290
73- // receive-side session stats indexed by session ID (see recv.go for "uid")
74- // optional, currently tests only
75- RxStats map [uint64 ]* Stats
76-
77- // object header
91+ // _object_ header (not to confuse w/ objects in buckets)
7892 ObjHdr struct {
7993 Bck cmn.Bck
8094 ObjName string
@@ -86,35 +100,27 @@ type (
86100 }
87101 // object to transmit
88102 Obj struct {
89- Reader io.ReadCloser // reader (to read the object, and close when done)
90- CmplArg any // optional context passed to the ObjSentCB callback
91- Callback ObjSentCB // called when the last byte is sent _or_ when the stream terminates (see term.reason)
92- prc * atomic.Int64 // private; if present, ref-counts so that we call ObjSentCB only once
93- Hdr ObjHdr
94- }
95-
96- // object-sent callback that has the following signature can optionally be defined on a:
97- // a) per-stream basis (via NewStream constructor - see Extra struct above)
98- // b) for a given object that is being sent (for instance, to support a call-per-batch semantics)
99- // Naturally, object callback "overrides" the per-stream one: when object callback is defined
100- // (i.e., non-nil), the stream callback is ignored/skipped.
101- // NOTE: if defined, the callback executes asynchronously as far as the sending part is concerned
102- ObjSentCB func (* ObjHdr , io.ReadCloser , any , error )
103-
104- Msg struct {
105- SID string
106- Body []byte
107- Opcode int
103+ Reader io.ReadCloser // reader (to read the object, and close when done)
104+ CmplArg any // optional context passed to the SentCB callback
105+ SentCB SentCB // called when the last byte is sent _or_ when the stream terminates (see term.reason)
106+ prc * atomic.Int64 // private; if present, ref-counts so that we call SentCB only once
107+ Hdr ObjHdr
108108 }
109109
110110 // stream collector
111111 StreamCollector struct {}
112112
113113 // Rx callbacks
114114 RecvObj func (hdr * ObjHdr , objReader io.Reader , err error ) error
115- RecvMsg func (msg Msg , err error ) error
116115)
117116
117+ // receive-side session stats indexed by session ID (see recv.go for "uid")
118+ // optional, currently tests only
119+ type (
120+ RxStats map [uint64 ]* Stats
121+ )
122+
123+ // shared data mover (SDM)
118124type (
119125 Receiver interface {
120126 ID () string
@@ -134,7 +140,9 @@ func NewObjStream(client Client, dstURL, dstID string, extra *Extra) (s *Stream)
134140 }
135141 s = & Stream {streamBase : * newBase (client , dstURL , dstID , extra )}
136142 s .streamBase .streamer = s
137- s .callback = extra .Callback
143+ if extra .Parent != nil {
144+ s .sentCB = extra .Parent .SentCB
145+ }
138146 if extra .Compressed () {
139147 s .initCompression (extra )
140148 }
@@ -162,10 +170,10 @@ func NewObjStream(client Client, dstURL, dstID string, extra *Extra) (s *Stream)
162170// when the header's Dsize field is set to zero), the reader is not required and the
163171// corresponding argument in Send() can be set to nil.
164172// - object reader is *always* closed irrespectively of whether the Send() succeeds
165- // or fails. On success, if send-completion (ObjSentCB ) callback is provided
173+ // or fails. On success, if send-completion (SentCB ) callback is provided
166174// (i.e., non-nil), the closing is done by doCmpl().
167175// - Optional reference counting is also done by (and in) the doCmpl, so that the
168- // ObjSentCB gets called if and only when the refcount (if provided i.e., non-nil)
176+ // SentCB gets called if and only when the refcount (if provided i.e., non-nil)
169177// reaches zero.
170178// - For every transmission of every object there's always an doCmpl() completion
171179// (with its refcounting and reader-closing). This holds true in all cases including
0 commit comments