From 2ae76283e52dab6a723c5afaba0fd37302c7d039 Mon Sep 17 00:00:00 2001 From: Leonard Liubich Date: Tue, 28 Jul 2026 11:41:08 +0300 Subject: [PATCH 1/2] sn/object: Refactor GET for 1 data part case It's planned to modify concurrent code for multiple parts. This will allow to not touch the code of a simple corner case. Signed-off-by: Leonard Liubich --- pkg/services/object/get/ec.go | 32 +++++++++++++++++++++++++++----- 1 file changed, 27 insertions(+), 5 deletions(-) diff --git a/pkg/services/object/get/ec.go b/pkg/services/object/get/ec.go index 8b60d16a67..98b3c796d0 100644 --- a/pkg/services/object/get/ec.go +++ b/pkg/services/object/get/ec.go @@ -1367,6 +1367,32 @@ func (s *Service) getECPartRangeFromNode(ctx context.Context, cnr cid.ID, parent } func (s *Service) streamECObject(ctx context.Context, transport GetECRequestTransport, rule iec.Rule, ruleIdx int, sortedNodes []netmap.NodeInfo) error { + if rule.DataPartNum == 1 { // no need in parallelism + // part payload size is ignored, but in practice the costs are small + copiedHdr, fullPldLen, _, copiedPldLen, err := s.streamFirstECPart(ctx, transport, rule, ruleIdx, sortedNodes) + if err != nil { + if errors.Is(err, ErrResponded) { + return nil + } + return fmt.Errorf("part#0: %w", err) + } + + if !copiedHdr { + return partialObjectCopy{} + } + + if copiedPldLen == fullPldLen { + return nil + } + + return partialObjectCopy{ + copiedHeader: true, + copiedPayloadLength: copiedPldLen, + } + } + + // parallel collection of multiple data parts + copiedHdr, fullPldLen, partPldLen, copiedPldLen, err := s.streamFirstECPart(ctx, transport, rule, ruleIdx, sortedNodes) if err != nil { if errors.Is(err, ErrResponded) { @@ -1379,16 +1405,12 @@ func (s *Service) streamECObject(ctx context.Context, transport GetECRequestTran return partialObjectCopy{} } - if rule.DataPartNum == 1 && copiedPldLen == partPldLen { - return nil - } - partial := partialObjectCopy{ copiedHeader: true, copiedPayloadLength: copiedPldLen, } - if copiedPldLen < partPldLen || rule.DataPartNum == 1 { + if copiedPldLen < partPldLen { return partial } From 6f866b8e55023fe3221b20ca05d593607a5eabbe Mon Sep 17 00:00:00 2001 From: Leonard Liubich Date: Tue, 28 Jul 2026 13:56:36 +0300 Subject: [PATCH 2/2] sn/object: Do not wait until first EC data part payload is fully copied Previously, secondary EC data chunks were blocked until the first one was completely received and sent to the response stream. However, in the current implementation, only the first part's and parent payload lengths are required to request secondary chunks (using ranged GET). Both of these values are obtained from header of the first part received using GET. Therefore, secondary routines may not wait for first part's payload to be fully copied and be unlocked earlier. This acceleration will be more noticeable with the growth of the object size (and consequently its data parts). Signed-off-by: Leonard Liubich --- pkg/services/object/get.go | 100 +++++++++++-------- pkg/services/object/get/ec.go | 151 ++++++++++++++++------------- pkg/services/object/get/service.go | 4 +- 3 files changed, 144 insertions(+), 111 deletions(-) diff --git a/pkg/services/object/get.go b/pkg/services/object/get.go index e9733d2caf..c21bf2b3c2 100644 --- a/pkg/services/object/get.go +++ b/pkg/services/object/get.go @@ -340,7 +340,7 @@ type getECTransport struct { } // CopyLocalECPartParentHeaderAndPayload implements [getsvc.GetECRequestTransport]. -func (x *getECTransport) CopyLocalECPartParentHeaderAndPayload(ctx context.Context, storage *engine.StorageEngine, partInfo iec.PartInfo) (bool, uint64, uint64, uint64, error) { +func (x *getECTransport) CopyLocalECPartParentHeaderAndPayload(ctx context.Context, storage *engine.StorageEngine, partInfo iec.PartInfo, interceptLens func(gotPartPldLen uint64, gotParentPldLen uint64)) (bool, uint64, error) { logError := func(msg string, err error) { x.server.log.Warn(msg, zap.Stringer("container", x.requestContainer), zap.Stringer("parent", x.requestObject), zap.Int("ruleIdx", partInfo.RuleIndex), zap.Int("partIdx", partInfo.Index), zap.Error(err)) @@ -353,19 +353,19 @@ func (x *getECTransport) CopyLocalECPartParentHeaderAndPayload(ctx context.Conte if err != nil { var splitErr *object.SplitInfoError if errors.Is(err, apistatus.ErrObjectAlreadyRemoved) || errors.As(err, &splitErr) { - return false, 0, 0, 0, err + return false, 0, err } if !errors.Is(err, apistatus.ErrObjectNotFound) { logError("local storage failure (read EC part)", err) } - return false, 0, 0, 0, nil + return false, 0, nil } defer stream.Close() _, _, partHdrf, err := iobject.GetNonPayloadFieldBounds(buf[:prefixLen]) if err != nil { - return false, 0, 0, 0, fmt.Errorf("parse first %d bytes of object protobuf: %w", prefixLen, err) + return false, 0, fmt.Errorf("parse first %d bytes of object protobuf: %w", prefixLen, err) } partHdrBuf := buf[partHdrf.ValueFrom:partHdrf.To] @@ -373,30 +373,32 @@ func (x *getECTransport) CopyLocalECPartParentHeaderAndPayload(ctx context.Conte typ, err := iobject.GetTypeHeader(partHdrBuf) if err != nil { logError("invalid local object header (get type)", err) - return false, 0, 0, 0, nil + return false, 0, nil } if typ == object.TypeLink { - return false, 0, 0, 0, getsvc.ErrLinker + return false, 0, getsvc.ErrLinker } partPldLen, err := iobject.GetPayloadLengthHeader(partHdrBuf) if err != nil { logError("invalid local object header (get payload length)", err) - return false, 0, 0, 0, nil + return false, 0, nil } parentIDf, parentSigf, parentHdrf, err := iobject.GetParentNonPayloadFieldBoundsHeader(partHdrBuf) if err != nil { logError("invalid local object header (get parent fields)", err) - return false, 0, 0, 0, nil + return false, 0, nil } parentPldLen, err := iobject.GetPayloadLengthHeader(partHdrBuf[parentHdrf.ValueFrom:parentHdrf.To]) if err != nil { logError("invalid local object header (get payload length from parent header)", err) - return false, 0, 0, 0, nil + return false, 0, nil } + interceptLens(partPldLen, parentPldLen) + var n int if !parentIDf.IsMissing() { @@ -418,13 +420,13 @@ func (x *getECTransport) CopyLocalECPartParentHeaderAndPayload(ctx context.Conte if err != nil { var e copyReadError if !errors.As(err, &e) { - return false, 0, 0, 0, err + return false, 0, err } logError("local storage stream failure (read EC part)", err) - return true, parentPldLen, partPldLen, uint64(e.written), nil + return true, uint64(e.written), nil } - return true, parentPldLen, partPldLen, partPldLen, nil + return true, partPldLen, nil } // CopyLocalECPartRange implements [getsvc.GetECRequestTransport]. @@ -495,9 +497,8 @@ func (x *getECTransport) initGetPartRequest(partInfo iec.PartInfo) error { } // CopyRemoteECPartParentHeaderAndPayload implements [getsvc.GetECRequestTransport]. -func (x *getECTransport) CopyRemoteECPartParentHeaderAndPayload(ctx context.Context, conn clientcore.MultiAddressClient, partInfo iec.PartInfo) (bool, uint64, uint64, uint64, error) { +func (x *getECTransport) CopyRemoteECPartParentHeaderAndPayload(ctx context.Context, conn clientcore.MultiAddressClient, partInfo iec.PartInfo, interceptLens func(gotPartPldLen uint64, gotParentPldLen uint64)) (bool, uint64, error) { var copiedHdr bool - var parentPldLen uint64 var partPldLen uint64 var copiedPartPld uint64 @@ -508,7 +509,10 @@ func (x *getECTransport) CopyRemoteECPartParentHeaderAndPayload(ctx context.Cont } var err error - copiedHdr, parentPldLen, partPldLen, copiedPartPld, err = x.copyRemotePart(ctx, conn) + copiedHdr, copiedPartPld, err = x.copyRemotePart(ctx, conn, func(gotPartPldLen uint64, gotParentPldLen uint64) { + partPldLen = gotPartPldLen + interceptLens(gotPartPldLen, gotParentPldLen) + }) if err != nil { return err } @@ -537,13 +541,13 @@ func (x *getECTransport) CopyRemoteECPartParentHeaderAndPayload(ctx context.Cont return clientcore.ErrSkipConnection }) if err != nil && !errors.Is(err, clientcore.ErrAllConnectionsSkipped) { - return false, 0, 0, 0, err + return false, 0, err } - return copiedHdr, parentPldLen, partPldLen, copiedPartPld, nil + return copiedHdr, copiedPartPld, nil } -func (x *getECTransport) copyRemotePart(ctx context.Context, conn *grpc.ClientConn) (bool, uint64, uint64, uint64, error) { +func (x *getECTransport) copyRemotePart(ctx context.Context, conn *grpc.ClientConn, interceptLens func(gotPartPldLen uint64, gotParentPldLen uint64)) (bool, uint64, error) { ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -551,11 +555,11 @@ func (x *getECTransport) copyRemotePart(ctx context.Context, conn *grpc.ClientCo if err != nil { err = igrpc.ConvertContextStatus(err) if errors.Is(err, ctx.Err()) { - return false, 0, 0, 0, err + return false, 0, err } // TODO: if error is due to incorrect request, error should be returned. How to catch this? x.server.log.Warn("GET object API failure (call)", zap.String("node", conn.Target()), zap.Error(err)) - return false, 0, 0, 0, nil + return false, 0, nil } var copiedHdr bool @@ -569,7 +573,7 @@ func (x *getECTransport) copyRemotePart(ctx context.Context, conn *grpc.ClientCo if err = stream.RecvMsg(&respBuf); err != nil { err = igrpc.ConvertContextStatus(err) if errors.Is(err, ctx.Err()) { - return false, 0, 0, 0, err + return false, 0, err } if !errors.Is(err, io.EOF) { x.server.log.Warn("GET object API failure (receive message)", zap.String("node", conn.Target()), zap.Error(err)) @@ -580,46 +584,50 @@ func (x *getECTransport) copyRemotePart(ctx context.Context, conn *grpc.ClientCo code, body, err := handleResponseCodeAndBody(respBuf) if err != nil { respBuf.Free() - return false, 0, 0, 0, err + return false, 0, err } if code == protostatus.ObjectNotFound { respBuf.Free() if headWas { - return false, 0, 0, 0, errors.New("received object not found status after header") + return false, 0, errors.New("received object not found status after header") } - return false, 0, 0, 0, nil + return false, 0, nil } if code != protostatus.OK { if err = x.responseStream.SendMsg(respBuf); err != nil { - return false, 0, 0, 0, fmt.Errorf("%w: %w", getsvc.ErrResponseStreamFailure, err) + return false, 0, fmt.Errorf("%w: %w", getsvc.ErrResponseStreamFailure, err) } - return false, 0, 0, 0, getsvc.ErrResponded + return false, 0, getsvc.ErrResponded } num, fld, err := handleGetResponseBodyOneof(&headWas, body) if err != nil { respBuf.Free() - return false, 0, 0, 0, err + return false, 0, err } switch num { default: respBuf.Free() - return false, 0, 0, 0, errors.New("none of the supported oneof fields are specified") + return false, 0, errors.New("none of the supported oneof fields are specified") case protoobject.FieldGetResponseBodyInit: var parentID, parentSig, parentHdr iprotobuf.BuffersSlice - parentID, parentSig, parentHdr, parentPldLen, partPldLen, err = handleGetECPartResponseInit(fld) + parentID, parentSig, parentHdr, err = handleGetECPartResponseInit(fld, func(gotPartPldLen uint64, gotParentPldLen uint64) { + partPldLen = gotPartPldLen + parentPldLen = gotParentPldLen + interceptLens(gotPartPldLen, gotParentPldLen) + }) if err != nil { respBuf.Free() - return false, 0, 0, 0, err + return false, 0, err } err = x.server.writeInitGetResponseBuffers(x.responseStream, parentID, parentSig, parentHdr, x.signResponses) respBuf.Free() if err != nil { - return false, 0, 0, 0, err + return false, 0, err } copiedHdr = true @@ -627,24 +635,24 @@ func (x *getECTransport) copyRemotePart(ctx context.Context, conn *grpc.ClientCo copiedPartPldLen += uint64(fld.Len()) if copiedPartPldLen > partPldLen { respBuf.Free() - return false, 0, 0, 0, fmt.Errorf("part payload overflow: full %d bytes, copied %d", partPldLen, copiedPartPldLen) + return false, 0, fmt.Errorf("part payload overflow: full %d bytes, copied %d", partPldLen, copiedPartPldLen) } if copiedPartPldLen > parentPldLen { respBuf.Free() - return false, 0, 0, 0, fmt.Errorf("parent payload overflow: full %d bytes, copied %d", parentPldLen, copiedPartPldLen) + return false, 0, fmt.Errorf("parent payload overflow: full %d bytes, copied %d", parentPldLen, copiedPartPldLen) } if err = x.responseStream.SendMsg(respBuf); err != nil { - return false, 0, 0, 0, fmt.Errorf("%w: %w", getsvc.ErrResponseStreamFailure, err) + return false, 0, fmt.Errorf("%w: %w", getsvc.ErrResponseStreamFailure, err) } case protoobject.FieldGetResponseBodySplitInfo: err := handleSplitInfo(fld, true) respBuf.Free() - return false, 0, 0, 0, err + return false, 0, err } } - return copiedHdr, parentPldLen, partPldLen, copiedPartPldLen, nil + return copiedHdr, copiedPartPldLen, nil } func (x *getECTransport) copyRemotePartRange(ctx context.Context, conn *grpc.ClientConn, partInfo iec.PartInfo, off, ln uint64, controlCh <-chan bool) (uint64, error) { @@ -759,10 +767,12 @@ func (x *getECTransport) copyRemotePartRange(ctx context.Context, conn *grpc.Cli return copied, nil } -func handleGetECPartResponseInit(buffers iprotobuf.BuffersSlice) (iprotobuf.BuffersSlice, iprotobuf.BuffersSlice, iprotobuf.BuffersSlice, uint64, uint64, error) { +func handleGetECPartResponseInit(buffers iprotobuf.BuffersSlice, interceptLens func(gotPartPldLen uint64, gotParentPldLen uint64)) (iprotobuf.BuffersSlice, iprotobuf.BuffersSlice, iprotobuf.BuffersSlice, error) { var parentID, parentSig, parentHdr iprotobuf.BuffersSlice var parentPldLen uint64 + var parentPldLenDone bool var partPldLen uint64 + var partPldLenDone bool var opts protoscan.ScanMessageOptions opts.InterceptNested = func(num protowire.Number, buffers iprotobuf.BuffersSlice) error { @@ -774,6 +784,10 @@ func handleGetECPartResponseInit(buffers iprotobuf.BuffersSlice) (iprotobuf.Buff opts.InterceptUint64 = func(num protowire.Number, u uint64) error { if num == protoobject.FieldHeaderPayloadLength { partPldLen = u + partPldLenDone = true + if parentPldLenDone { + interceptLens(partPldLen, parentPldLen) + } } return nil } @@ -798,6 +812,10 @@ func handleGetECPartResponseInit(buffers iprotobuf.BuffersSlice) (iprotobuf.Buff opts.InterceptUint64 = func(num protowire.Number, u uint64) error { if num == protoobject.FieldHeaderPayloadLength { parentPldLen = u + parentPldLenDone = true + if partPldLenDone { + interceptLens(partPldLen, parentPldLen) + } } return nil } @@ -822,10 +840,14 @@ func handleGetECPartResponseInit(buffers iprotobuf.BuffersSlice) (iprotobuf.Buff err := protoscan.ScanMessage(buffers, protoscan.ObjectGetResponseInitScheme, opts) if err != nil { - return iprotobuf.BuffersSlice{}, iprotobuf.BuffersSlice{}, iprotobuf.BuffersSlice{}, 0, 0, err + return iprotobuf.BuffersSlice{}, iprotobuf.BuffersSlice{}, iprotobuf.BuffersSlice{}, err + } + + if !partPldLenDone || !parentPldLenDone { + interceptLens(partPldLen, parentPldLen) } - return parentID, parentSig, parentHdr, parentPldLen, partPldLen, nil + return parentID, parentSig, parentHdr, nil } func (x *getECTransport) CopyRemoteECPartRange(ctx context.Context, conn clientcore.MultiAddressClient, partInfo iec.PartInfo, off uint64, ln uint64, controlCh <-chan bool) (uint64, error) { diff --git a/pkg/services/object/get/ec.go b/pkg/services/object/get/ec.go index 98b3c796d0..6ab9768658 100644 --- a/pkg/services/object/get/ec.go +++ b/pkg/services/object/get/ec.go @@ -1367,9 +1367,13 @@ func (s *Service) getECPartRangeFromNode(ctx context.Context, cnr cid.ID, parent } func (s *Service) streamECObject(ctx context.Context, transport GetECRequestTransport, rule iec.Rule, ruleIdx int, sortedNodes []netmap.NodeInfo) error { + var parentPldLen uint64 + if rule.DataPartNum == 1 { // no need in parallelism // part payload size is ignored, but in practice the costs are small - copiedHdr, fullPldLen, _, copiedPldLen, err := s.streamFirstECPart(ctx, transport, rule, ruleIdx, sortedNodes) + copiedHdr, copiedPldLen, err := s.streamFirstECPart(ctx, transport, rule, ruleIdx, sortedNodes, func(_ uint64, gotParentPldLen uint64) { + parentPldLen = gotParentPldLen + }) if err != nil { if errors.Is(err, ErrResponded) { return nil @@ -1381,7 +1385,7 @@ func (s *Service) streamECObject(ctx context.Context, transport GetECRequestTran return partialObjectCopy{} } - if copiedPldLen == fullPldLen { + if copiedPldLen == parentPldLen { return nil } @@ -1393,53 +1397,10 @@ func (s *Service) streamECObject(ctx context.Context, transport GetECRequestTran // parallel collection of multiple data parts - copiedHdr, fullPldLen, partPldLen, copiedPldLen, err := s.streamFirstECPart(ctx, transport, rule, ruleIdx, sortedNodes) - if err != nil { - if errors.Is(err, ErrResponded) { - return nil - } - return fmt.Errorf("part#0: %w", err) - } - - if !copiedHdr { - return partialObjectCopy{} - } - - partial := partialObjectCopy{ - copiedHeader: true, - copiedPayloadLength: copiedPldLen, - } - - if copiedPldLen < partPldLen { - return partial - } - - if rule.DataPartNum == 2 { - partInfo := iec.PartInfo{ - RuleIndex: ruleIdx, - Index: 1, - } - copiedPartPld, err := s.streamECPartRangePrefix(ctx, transport, rule, partInfo, sortedNodes, fullPldLen-partPldLen, nil) - if err != nil { - if errors.Is(err, ErrResponded) { - return nil - } - return fmt.Errorf("part#1: %w", err) - } - - partial.copiedPayloadLength += copiedPartPld - if partial.copiedPayloadLength > fullPldLen { - return fmt.Errorf("payload overflow: full %d bytes, copied %d", fullPldLen, partial.copiedPayloadLength) - } - - if partial.copiedPayloadLength == fullPldLen { - return nil - } - - return partial - } + var partPldLen uint64 + lensGotCh := make(chan struct{}) - controlChs := make([]chan bool, rule.DataPartNum-2) // -1 because 1st part already copied, -1 because 2nd part doesn't need a trigger + controlChs := make([]chan bool, rule.DataPartNum-1) // -1 because 1st part doesn't need a trigger for i := range controlChs { controlChs[i] = make(chan bool, 1) } @@ -1448,29 +1409,38 @@ func (s *Service) streamECObject(ctx context.Context, transport GetECRequestTran for j := i + 1; j < int(rule.DataPartNum); j++ { // non-blocking because multiple routines can call this while it's enough to write once select { - case controlChs[j-2] <- true: + case controlChs[j-1] <- true: default: } } } + ctx, cancel := context.WithCancel(ctx) + defer cancel() + var resErr error + var copiedPldLen uint64 var wg sync.WaitGroup for i := 1; i < int(rule.DataPartNum); i++ { wg.Go(func() { + // wait for part#0 routine to fetch both part and parent payload length + select { + case <-ctx.Done(): + return // result already reached + case <-lensGotCh: + } + var ln uint64 if i < int(rule.DataPartNum)-1 { + // TODO: consider requesting (0,0) to not wait for part#0 by default. On failure, [X:] range can be used for continuation ln = partPldLen } else { // last part can be suffixed with zeros which should not be transmitted - ln = fullPldLen - partPldLen*uint64(rule.DataPartNum-1) + ln = parentPldLen - partPldLen*uint64(rule.DataPartNum-1) } - var controlCh <-chan bool - if i > 1 { - controlCh = controlChs[i-2] - } + controlCh := controlChs[i-1] partInfo := iec.PartInfo{ RuleIndex: ruleIdx, @@ -1488,10 +1458,10 @@ func (s *Service) streamECObject(ctx context.Context, transport GetECRequestTran if copiedPartPld > 0 { // data is copied sequentially, so no concurrency here - partial.copiedPayloadLength += copiedPartPld - if partial.copiedPayloadLength > fullPldLen { + copiedPldLen += copiedPartPld + if copiedPldLen > parentPldLen { abortNextTo(i) - resErr = fmt.Errorf("payload overflow: full %d bytes, copied %d", fullPldLen, partial.copiedPayloadLength) + resErr = fmt.Errorf("payload overflow: full %d bytes, copied %d", parentPldLen, copiedPldLen) return } } @@ -1501,11 +1471,44 @@ func (s *Service) streamECObject(ctx context.Context, transport GetECRequestTran return } + // TODO: double-check this happens fast enough after current routine received the last chunk if i < int(rule.DataPartNum)-1 { - close(controlChs[i-1]) + close(controlChs[i]) } }) } + + var lensGot atomic.Bool + + copiedHdr, copiedPldLen, err := s.streamFirstECPart(ctx, transport, rule, ruleIdx, sortedNodes, func(gotPartPldLen uint64, gotParentPldLen uint64) { + if lensGot.Swap(true) { + return + } + partPldLen = gotPartPldLen + parentPldLen = gotParentPldLen + close(lensGotCh) + }) + if err != nil { + if errors.Is(err, ErrResponded) { + return nil + } + return fmt.Errorf("part#0: %w", err) + } + + if !copiedHdr { + return partialObjectCopy{} + } + + if copiedPldLen < partPldLen { + return partialObjectCopy{ + copiedHeader: true, + copiedPayloadLength: copiedPldLen, + } + } + + // unlock part#1 routine + close(controlChs[0]) + wg.Wait() if resErr != nil { @@ -1515,17 +1518,19 @@ func (s *Service) streamECObject(ctx context.Context, transport GetECRequestTran return resErr } - if partial.copiedPayloadLength == fullPldLen { + if copiedPldLen == parentPldLen { return nil } - return partial + return partialObjectCopy{ + copiedHeader: true, + copiedPayloadLength: copiedPldLen, + } } -func (s *Service) streamFirstECPart(ctx context.Context, transport GetECRequestTransport, rule iec.Rule, ruleIdx int, sortedNodes []netmap.NodeInfo) (bool, uint64, uint64, uint64, error) { +func (s *Service) streamFirstECPart(ctx context.Context, transport GetECRequestTransport, rule iec.Rule, ruleIdx int, sortedNodes []netmap.NodeInfo, interceptLens func(gotPartPldLen uint64, gotParentPldLen uint64)) (bool, uint64, error) { var err error var copiedHdr bool - var parentPldLen uint64 var partPldLen uint64 var copiedPartPld uint64 @@ -1539,22 +1544,28 @@ func (s *Service) streamFirstECPart(ctx context.Context, transport GetECRequestT if !copiedHdr { if local { - copiedHdr, parentPldLen, partPldLen, copiedPartPld, err = transport.CopyLocalECPartParentHeaderAndPayload(ctx, s.localObjects.(*engine.StorageEngine), partInfo) + copiedHdr, copiedPartPld, err = transport.CopyLocalECPartParentHeaderAndPayload(ctx, s.localObjects.(*engine.StorageEngine), partInfo, func(gotPartPldLen uint64, gotParentPldLen uint64) { + partPldLen = gotPartPldLen + interceptLens(gotPartPldLen, gotParentPldLen) + }) } else { conn, connErr := s.conns.(*clientCacheWrapper).connect(ctx, sortedNodes[nodeIdx]) if connErr != nil { connErr = igrpc.ConvertContextStatus(connErr) if errors.Is(connErr, ctx.Err()) { - return false, 0, 0, 0, connErr + return false, 0, connErr } s.logSNConnFailure(sortedNodes[nodeIdx], connErr) continue } - copiedHdr, parentPldLen, partPldLen, copiedPartPld, err = transport.CopyRemoteECPartParentHeaderAndPayload(ctx, conn, partInfo) + copiedHdr, copiedPartPld, err = transport.CopyRemoteECPartParentHeaderAndPayload(ctx, conn, partInfo, func(gotPartPldLen uint64, gotParentPldLen uint64) { + partPldLen = gotPartPldLen + interceptLens(partPldLen, gotParentPldLen) + }) } if err != nil { - return false, 0, 0, 0, err + return false, 0, err } // TODO: verify partPldLen against parentPldLen @@ -1574,7 +1585,7 @@ func (s *Service) streamFirstECPart(ctx context.Context, transport GetECRequestT if connErr != nil { connErr = igrpc.ConvertContextStatus(connErr) if errors.Is(connErr, ctx.Err()) { - return false, 0, 0, 0, connErr + return false, 0, connErr } s.logSNConnFailure(sortedNodes[nodeIdx], connErr) continue @@ -1583,12 +1594,12 @@ func (s *Service) streamFirstECPart(ctx context.Context, transport GetECRequestT copiedFromNode, err = transport.CopyRemoteECPartRange(ctx, conn, partInfo, copiedPartPld, partPldLen-copiedPartPld, nil) } if err != nil { - return false, 0, 0, 0, err + return false, 0, err } copiedPartPld += copiedFromNode if copiedPartPld > partPldLen { - return false, 0, 0, 0, fmt.Errorf("part payload overflow: full %d bytes, copied %d", partPldLen, copiedPartPld) + return false, 0, fmt.Errorf("part payload overflow: full %d bytes, copied %d", partPldLen, copiedPartPld) } if copiedPartPld == partPldLen { @@ -1596,7 +1607,7 @@ func (s *Service) streamFirstECPart(ctx context.Context, transport GetECRequestT } } - return copiedHdr, parentPldLen, partPldLen, copiedPartPld, nil + return copiedHdr, copiedPartPld, nil } func (s *Service) streamECPartRangePrefix(ctx context.Context, transport GetECRequestTransport, rule iec.Rule, partInfo iec.PartInfo, sortedNodes []netmap.NodeInfo, ln uint64, controlCh <-chan bool) (uint64, error) { diff --git a/pkg/services/object/get/service.go b/pkg/services/object/get/service.go index 247d5f867a..6f708b8b63 100644 --- a/pkg/services/object/get/service.go +++ b/pkg/services/object/get/service.go @@ -93,9 +93,9 @@ type GetECRequestTransport interface { // Otherwise, no error is returned. Copying can be incomplete in this case. // // CopyRemoteECPartParentHeaderAndPayload is never called concurrently. - CopyRemoteECPartParentHeaderAndPayload(ctx context.Context, conn clientcore.MultiAddressClient, partInfo iec.PartInfo) (bool, uint64, uint64, uint64, error) + CopyRemoteECPartParentHeaderAndPayload(ctx context.Context, conn clientcore.MultiAddressClient, partInfo iec.PartInfo, interceptLens func(gotPartPldLen uint64, gotParentPldLen uint64)) (bool, uint64, error) // TODO: upd docs. // CopyLocalECPartParentHeaderAndPayload works like CopyRemoteECPartParentHeaderAndPayload but locally. - CopyLocalECPartParentHeaderAndPayload(ctx context.Context, storage *engine.StorageEngine, partInfo iec.PartInfo) (bool, uint64, uint64, uint64, error) + CopyLocalECPartParentHeaderAndPayload(ctx context.Context, storage *engine.StorageEngine, partInfo iec.PartInfo, interceptLens func(gotPartPldLen uint64, gotParentPldLen uint64)) (bool, uint64, error) // CopyRemoteECPartRange requests specified payload range of originally // requested object's EC part identified by partInfo pair from remote storage // node using conn to it. If succeeded, CopyRemoteECPartRange sends with payload