proxy: break up writes from the remote writer to avoid grpc limits
The remote content writer proxy already has the capability to break up large files into multiple writes, but the current API doesn't recognize when it's about to exceed the limits and attempts to send the data over grpc in one message instead of breaking it into multiple messages. This changes the behavior of `Write` to automatically break up the size of the content based on the max send message size. Signed-off-by: Jonathan A. Sternberg <jonathan.sternberg@docker.com> (cherry picked from commit f25f36c334144d87233e06b0de90522ebd97e144)
This commit is contained in:
@@ -26,6 +26,7 @@ import (
|
||||
digest "github.com/opencontainers/go-digest"
|
||||
|
||||
"github.com/containerd/containerd/v2/core/content"
|
||||
"github.com/containerd/containerd/v2/defaults"
|
||||
"github.com/containerd/containerd/v2/pkg/protobuf"
|
||||
)
|
||||
|
||||
@@ -76,27 +77,37 @@ func (rw *remoteWriter) Digest() digest.Digest {
|
||||
}
|
||||
|
||||
func (rw *remoteWriter) Write(p []byte) (n int, err error) {
|
||||
offset := rw.offset
|
||||
const maxBufferSize = defaults.DefaultMaxSendMsgSize >> 1
|
||||
for i := 0; i < len(p); i += maxBufferSize {
|
||||
offset := rw.offset
|
||||
|
||||
resp, err := rw.send(&contentapi.WriteContentRequest{
|
||||
Action: contentapi.WriteAction_WRITE,
|
||||
Offset: offset,
|
||||
Data: p,
|
||||
})
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("failed to send write: %w", errgrpc.ToNative(err))
|
||||
}
|
||||
end := i + maxBufferSize
|
||||
if end > len(p) {
|
||||
end = len(p)
|
||||
}
|
||||
data := p[i:end]
|
||||
|
||||
n = int(resp.Offset - offset)
|
||||
if n < len(p) {
|
||||
err = io.ErrShortWrite
|
||||
}
|
||||
resp, err := rw.send(&contentapi.WriteContentRequest{
|
||||
Action: contentapi.WriteAction_WRITE,
|
||||
Offset: offset,
|
||||
Data: data,
|
||||
})
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("failed to send write: %w", errgrpc.ToNative(err))
|
||||
}
|
||||
|
||||
rw.offset += int64(n)
|
||||
if resp.Digest != "" {
|
||||
rw.digest = digest.Digest(resp.Digest)
|
||||
written := int(resp.Offset - offset)
|
||||
rw.offset += int64(written)
|
||||
if resp.Digest != "" {
|
||||
rw.digest = digest.Digest(resp.Digest)
|
||||
}
|
||||
n += written
|
||||
|
||||
if written < len(data) {
|
||||
return n, io.ErrShortWrite
|
||||
}
|
||||
}
|
||||
return
|
||||
return n, nil
|
||||
}
|
||||
|
||||
func (rw *remoteWriter) Commit(ctx context.Context, size int64, expected digest.Digest, opts ...content.Opt) (err error) {
|
||||
|
||||
Reference in New Issue
Block a user