[#1605] node: Stream forwarder requests

Do not hold objects in memory in the forwarded Get/GetRange requests.

Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>
This commit is contained in:
Pavel Karpy 2022-07-15 09:55:53 +03:00 committed by fyrchik
parent 89ebc278b8
commit a421344727
2 changed files with 27 additions and 26 deletions

View file

@ -36,8 +36,14 @@ func (exec *execCtx) processNode(ctx context.Context, info client.NodeInfo) bool
case err == nil: case err == nil:
exec.status = statusOK exec.status = statusOK
exec.err = nil exec.err = nil
// both object and err are nil only if the original
// request was forwarded to another node and the object
// has already been streamed to the requesting party
if obj != nil {
exec.collectedObject = obj exec.collectedObject = obj
exec.writeCollectedObject() exec.writeCollectedObject()
}
case errors.As(err, &errRemoved): case errors.As(err, &errRemoved):
exec.status = statusINHUMED exec.status = statusINHUMED
exec.err = errRemoved exec.err = errRemoved

View file

@ -54,12 +54,14 @@ func (s *Service) toPrm(req *objectV2.GetRequest, stream objectSvc.GetObjectStre
return nil, err return nil, err
} }
streamWrapper := &streamObjectWriter{stream}
p := new(getsvc.Prm) p := new(getsvc.Prm)
p.SetCommonParameters(commonPrm) p.SetCommonParameters(commonPrm)
p.WithAddress(addr) p.WithAddress(addr)
p.WithRawFlag(body.GetRaw()) p.WithRawFlag(body.GetRaw())
p.SetObjectWriter(&streamObjectWriter{stream}) p.SetObjectWriter(streamWrapper)
if !commonPrm.LocalOnly() { if !commonPrm.LocalOnly() {
var onceResign sync.Once var onceResign sync.Once
@ -105,8 +107,6 @@ func (s *Service) toPrm(req *objectV2.GetRequest, stream objectSvc.GetObjectStre
var ( var (
headWas bool headWas bool
payload []byte
obj = new(objectV2.Object)
resp = new(objectV2.GetResponse) resp = new(objectV2.GetResponse)
) )
@ -149,29 +149,30 @@ func (s *Service) toPrm(req *objectV2.GetRequest, stream objectSvc.GetObjectStre
headWas = true headWas = true
obj := new(objectV2.Object)
obj.SetObjectID(v.GetObjectID()) obj.SetObjectID(v.GetObjectID())
obj.SetSignature(v.GetSignature()) obj.SetSignature(v.GetSignature())
obj.SetHeader(v.GetHeader())
hdr := v.GetHeader() if err = streamWrapper.WriteHeader(object.NewFromV2(obj)); err != nil {
obj.SetHeader(hdr) return nil, fmt.Errorf("could not write object header in Get forwarder: %w", err)
}
payload = make([]byte, 0, hdr.GetPayloadLength())
case *objectV2.GetObjectPartChunk: case *objectV2.GetObjectPartChunk:
if !headWas { if !headWas {
return nil, errWrongMessageSeq return nil, errWrongMessageSeq
} }
payload = append(payload, v.GetChunk()...) if err = streamWrapper.WriteChunk(v.GetChunk()); err != nil {
return nil, fmt.Errorf("could not write object chunk in Get forwarder: %w", err)
}
case *objectV2.SplitInfo: case *objectV2.SplitInfo:
si := object.NewSplitInfoFromV2(v) si := object.NewSplitInfoFromV2(v)
return nil, object.NewSplitInfoError(si) return nil, object.NewSplitInfoError(si)
} }
} }
obj.SetPayload(payload) return nil, nil
// convert the object
return object.NewFromV2(obj), nil
})) }))
} }
@ -203,9 +204,11 @@ func (s *Service) toRangePrm(req *objectV2.GetRangeRequest, stream objectSvc.Get
p := new(getsvc.RangePrm) p := new(getsvc.RangePrm)
p.SetCommonParameters(commonPrm) p.SetCommonParameters(commonPrm)
streamWrapper := &streamObjectRangeWriter{stream}
p.WithAddress(addr) p.WithAddress(addr)
p.WithRawFlag(body.GetRaw()) p.WithRawFlag(body.GetRaw())
p.SetChunkWriter(&streamObjectRangeWriter{stream}) p.SetChunkWriter(streamWrapper)
p.SetRange(object.NewRangeFromV2(body.GetRange())) p.SetRange(object.NewRangeFromV2(body.GetRange()))
if !commonPrm.LocalOnly() { if !commonPrm.LocalOnly() {
@ -250,9 +253,6 @@ func (s *Service) toRangePrm(req *objectV2.GetRangeRequest, stream objectSvc.Get
return nil, fmt.Errorf("could not create Get payload range stream: %w", err) return nil, fmt.Errorf("could not create Get payload range stream: %w", err)
} }
// allocate memory only after receiving a successful response
var payload []byte
resp := new(objectV2.GetRangeResponse) resp := new(objectV2.GetRangeResponse)
for { for {
@ -284,11 +284,9 @@ func (s *Service) toRangePrm(req *objectV2.GetRangeRequest, stream objectSvc.Get
case nil: case nil:
return nil, fmt.Errorf("unexpected range type %T", v) return nil, fmt.Errorf("unexpected range type %T", v)
case *objectV2.GetRangePartChunk: case *objectV2.GetRangePartChunk:
if payload == nil { if err = streamWrapper.WriteChunk(v.GetChunk()); err != nil {
payload = make([]byte, 0, body.GetRange().GetLength()) return nil, fmt.Errorf("could not write object chunk in GetRange forwarder: %w", err)
} }
payload = append(payload, v.GetChunk()...)
case *objectV2.SplitInfo: case *objectV2.SplitInfo:
si := object.NewSplitInfoFromV2(v) si := object.NewSplitInfoFromV2(v)
@ -296,10 +294,7 @@ func (s *Service) toRangePrm(req *objectV2.GetRangeRequest, stream objectSvc.Get
} }
} }
obj := object.New() return nil, nil
obj.SetPayload(payload)
return obj, nil
})) }))
} }