diff --git a/pkg/appview/storage/proxy_blob_store.go b/pkg/appview/storage/proxy_blob_store.go index 9a30b90..c1e3c12 100644 --- a/pkg/appview/storage/proxy_blob_store.go +++ b/pkg/appview/storage/proxy_blob_store.go @@ -706,7 +706,10 @@ type inFlightPart struct { // Docker PATCH body was not being drained, and while the buffer filled S3 sat // idle, so the wall clock for a layer was receive time plus send time. The // price is a second buffer, so a large upload's peak memory is -// 2 * maxBufferSize (32MB), not one buffer's worth. +// 2 * maxBufferSize (32MB), not one buffer's worth. That second buffer is +// taken only when the budget can spare it without waiting; a writer that +// cannot have one falls back to the serial upload, which is slower but needs +// no memory the writer does not already hold. type ProxyBlobWriter struct { store *ProxyBlobStore options distribution.CreateOptions @@ -726,10 +729,10 @@ type ProxyBlobWriter struct { // is what makes part numbers and ETags ordered for free. flight *inFlightPart - // spare is the second buffer while it is idle: handed back by a finished - // part upload with its capacity intact, waiting to become the next part. - // nil until the first hand-off, so a blob that never flushes never - // allocates it. + // spare is the other buffer while it is idle: handed back by a finished + // part upload with its capacity intact, waiting to be filled again. nil + // until a part has landed, so a blob that never flushes never allocates a + // second buffer, and so does a writer whose parts go up one at a time. spare *bytes.Buffer // flightErr is the failure of a background part upload, kept so it can be @@ -822,7 +825,8 @@ func (w *ProxyBlobWriter) projectedCap(n int) int64 { // otherBufferCap is the backing array held by the writer's second buffer: the // one currently being uploaded, or the one a finished upload handed back. Zero -// until the first hand-off, because until then there is no second buffer. +// whenever the writer has only one buffer, which is every blob that never +// flushes and every writer the budget could not spare a second buffer for. // // Reading Cap on a buffer that is in flight is safe: runPart only reads out of // it (Bytes), and the Reset that hands it back is done under w.mu. @@ -841,15 +845,22 @@ func (w *ProxyBlobWriter) otherBufferCap() int64 { // budgetDelta is the extra budget the writer must acquire before an n byte // write can land. // -// What is charged is the backing arrays, not the bytes written: the buffer -// being filled, plus the second buffer from the moment one exists at all. The -// second is charged as it grows, through this same path, because it is created -// empty at hand-off and only reaches full size as writes land in it. +// What is charged is the backing arrays, not the bytes written. This path only +// ever charges for the buffer being filled, and only while that is the writer's +// first one: a second buffer is charged in full at hand-off (see +// tryTakeSecondBuffer), so once one exists the two capacities together are +// already covered by w.charged and the delta here is zero for the rest of the +// upload. // -// So a large upload now costs 2 * maxBufferSize, 32MB, not 16MB: that is the -// price of holding one part in flight while the next one fills. A blob that -// never fills a buffer never causes a hand-off, never allocates a second -// buffer, and is charged only for what its one buffer grew to. +// That split is deliberate. This is the blocking charge, and a write must only +// ever block for memory the writer genuinely cannot proceed without. One full +// buffer is enough to finish any blob, so waiting for the first one is honest +// backpressure; waiting for the second was a wedge, because every writer +// mid-growth held memory that only a Commit could release and no writer could +// reach Commit. +// +// A blob that never fills a buffer never causes a hand-off, never allocates a +// second buffer, and is charged only for what its one buffer grew to. // // Nothing is released between parts, because bytes.Buffer.Reset keeps the // array: releasing there would report memory as free while the writer still @@ -939,7 +950,9 @@ func (w *ProxyBlobWriter) reapIfIdle(now time.Time, ttl time.Duration) (time.Dur // is handed to a background part upload and this call carries straight on into // a second buffer. At most one part is ever in flight: when the second buffer // fills too, the write blocks until the running part lands, takes its buffer -// back, and hands the second one off. +// back, and hands the second one off. A writer the budget could not give a +// second buffer to uploads the part where it stands and carries on in the same +// buffer. // // Never let the buffer pass the threshold. Appending a whole chunk and // checking afterwards let the last chunk before a flush land a few bytes past @@ -1068,8 +1081,16 @@ func (w *ProxyBlobWriter) growBuffer(n int) { w.buffer.Grow(maxBufferSize - w.buffer.Len()) } -// handOffBuffer gives the full buffer to a background part upload and leaves -// the writer a second, empty buffer to keep filling. +// handOffBuffer gives the full buffer to a part upload and leaves the writer an +// empty one to keep filling. +// +// The part goes up in the background whenever the writer has somewhere to carry +// on writing: the buffer a finished part handed back, or a fresh one when the +// budget for it is free right now. When it is not, the part goes up on this +// goroutine and the same buffer comes back, which is slower but always +// possible. That asymmetry is the safety property the budget rests on: a writer +// holding one full buffer can always finish the blob, so it can always reach +// the Commit that returns its budget. // // Only one part is ever in flight, so if a previous one is still running this // waits for it first and reuses the buffer it hands back. That wait is where a @@ -1093,13 +1114,14 @@ func (w *ProxyBlobWriter) handOffBuffer() error { // The buffer the finished part handed back, capacity intact // (bytes.Buffer.Reset keeps the backing array), so a large upload allocates - // its two buffers once and then just swaps them. nil only on the very first - // hand-off: a blob that never fills a buffer never gets here, and so never - // allocates a second one. + // its two buffers once and then just swaps them. nil only while the writer + // has never had a second buffer: a blob that never fills a buffer never + // gets here, and one whose second buffer the budget could not afford + // uploads its parts one at a time out of the first. next := w.spare w.spare = nil if next == nil { - next = &bytes.Buffer{} + next = w.tryTakeSecondBuffer() } f := &inFlightPart{ @@ -1108,13 +1130,76 @@ func (w *ProxyBlobWriter) handOffBuffer() error { done: make(chan struct{}), } w.partNumber++ - w.buffer = next w.flight = f + if next == nil { + // No budget to spare, so there is nothing to overlap the upload with: + // run it here and take the same buffer back. The next hand-off tries + // again, so a writer starts pipelining as soon as memory frees up. + return w.runPartInline(f) + } + + w.buffer = next go w.runPart(f) return nil } +// tryTakeSecondBuffer returns a second buffer for the writer to carry on in, or +// nil when the budget cannot spare one right now. Never waits: this is the +// charge the writer must be able to do without. +// +// The full maxBufferSize is charged and allocated up front rather than grown +// into. A writer only reaches here having already filled one whole buffer, so +// this one will be filled too unless the blob ends within the next part, and +// charging it whole keeps w.charged exactly equal to the backing arrays the +// writer holds. It also means no later write in this upload has to ask the +// budget for anything. +// +// A buffer's worth of budget is left free on purpose. Pipelining is a bonus, +// and spending the last of the budget on it would leave an arriving writer +// unable to fill even its first buffer, which is the one charge that has to be +// waited for. +// +// Callers hold w.mu. +func (w *ProxyBlobWriter) tryTakeSecondBuffer() *bytes.Buffer { + if !uploadBudget.tryAcquire(maxBufferSize, maxBufferSize) { + return nil + } + w.charged += maxBufferSize + + next := &bytes.Buffer{} + next.Grow(maxBufferSize) + return next +} + +// runPartInline uploads the part on this goroutine and takes its buffer back, +// which is what a writer with only one buffer has to do. +// +// The writer is left exactly as a background part leaves it, minus the overlap: +// the part is recorded (or its failure is), and the buffer comes back empty +// with its capacity intact to be filled again. +// +// Callers hold w.mu. The lock is dropped for the upload, which it must be: +// runPart takes it to record the outcome and hand the buffer back. w.flight is +// set for the whole of it, so the sweeper leaves the writer alone and a Cancel +// waits for the part rather than aborting the session underneath it. +func (w *ProxyBlobWriter) runPartInline(f *inFlightPart) error { + w.mu.Unlock() + w.runPart(f) + w.mu.Lock() + + // runPart handed the buffer back as the spare. It is the only one this + // writer has, so it goes straight back to being the one being filled. + w.buffer = w.spare + w.spare = nil + + if w.closed { + // The part failed, or a Cancel arrived while it was going up. + return w.closedErr() + } + return nil +} + // drainFlight waits for the in-flight part, if there is one, and reports // whether it failed. The failure is sticky, so every later caller sees the // cause rather than just a closed writer. diff --git a/pkg/appview/storage/upload_budget.go b/pkg/appview/storage/upload_budget.go index 28b2e4f..78bdc64 100644 --- a/pkg/appview/storage/upload_budget.go +++ b/pkg/appview/storage/upload_budget.go @@ -12,22 +12,30 @@ import ( const ( // maxWriterBuffers is how many buffers one writer can hold at its peak: the // one being filled plus the one being uploaded. A blob small enough never - // to flush holds one and never allocates the second. + // to flush holds one and never allocates the second, and so does a large + // one whose second buffer the budget could not afford. maxWriterBuffers = 2 - // maxWriterFootprint is a writer's peak buffer memory, 32MB. It is what a - // large upload costs now that a part is uploaded in the background while - // the next buffer fills; before that overlap a writer peaked at one buffer. + // maxWriterFootprint is a writer's peak buffer memory, 32MB: what a large + // upload costs when it gets to upload a part in the background while the + // next buffer fills. It is a ceiling, not a price of admission. Only the + // first buffer is ever waited for; the second is taken only when the budget + // can spare it, and a writer that cannot have one uploads its parts + // synchronously out of the single buffer it already holds. maxWriterFootprint = maxWriterBuffers * maxBufferSize // defaultUploadBufferBudget is the process-wide ceiling on bytes held in // blob upload buffers. Every in-flight push holds up to maxWriterFootprint, // Docker pushes up to five layers at once per client, and nothing used to // stop N clients from multiplying that out until the AppView was killed by - // the OOM reaper. 512MB is 16 large uploads at their peak (or 32 that never - // grew a second buffer), which is far more concurrency than a single - // AppView instance sees in practice while still fitting comfortably in the - // smallest deployment. + // the OOM reaper. + // + // What the number buys is progress for budget/maxBufferSize large uploads, + // 32 of them at 512MB, because one full buffer is all a writer needs to + // finish. Pipelining is what the rest of the memory is spent on when it is + // there: up to 16 of those 32 can also hold a part in flight. That is far + // more concurrency than a single AppView instance sees in practice while + // still fitting comfortably in the smallest deployment. defaultUploadBufferBudget = 512 * 1024 * 1024 // 512MB // defaultUploadIdleTimeout is how long a writer may go without a Write or @@ -84,9 +92,9 @@ func newBufferBudget(limit int64) *bufferBudget { } // acquire blocks until n bytes of budget are available, the context is done, or -// the wait cap expires. A writer never charges more than maxWriterFootprint in -// total, and the budget is never smaller than that, so a single writer can -// always eventually be satisfied. +// the wait cap expires. Only a writer's first buffer is ever acquired this way, +// and that buffer never exceeds maxBufferSize, which the budget is never +// smaller than, so a single writer can always eventually be satisfied. func (b *bufferBudget) acquire(ctx context.Context, n int64) error { if n <= 0 { return nil @@ -101,6 +109,35 @@ func (b *bufferBudget) acquire(ctx context.Context, n int64) error { return nil } +// tryAcquire takes n bytes of budget if they are free right now and keepFree +// bytes would still be free afterwards, and reports whether it got them. It +// never blocks and never queues, which is what makes it safe for a charge a +// writer must be able to do without: waiting for anything past the first buffer +// is what wedged the budget. +// +// keepFree is asked for as part of the same weight and handed straight back, +// because the semaphore has no way to ask whether that much more is free +// without taking it. Only n ends up held. Between the two calls the budget +// looks fuller than it is, which can make a concurrent tryAcquire decline; that +// costs a writer its pipelining for one part and nothing more. +// +// x/sync/semaphore's TryAcquire also fails outright while anyone is queued on +// Acquire, so an opportunistic second buffer can never jump ahead of a writer +// that is waiting for its first. +func (b *bufferBudget) tryAcquire(n, keepFree int64) bool { + if n <= 0 { + return true + } + if !b.sem.TryAcquire(n + keepFree) { + return false + } + if keepFree > 0 { + b.sem.Release(keepFree) + } + b.held.Add(n) + return true +} + // release returns n bytes to the budget. Callers must release exactly what they // acquired and no more; ProxyBlobWriter does that by tracking a single charged // figure and zeroing it as it releases. @@ -131,11 +168,13 @@ var ( // semaphore outright rather than resizing it, which is only safe while nothing // holds budget. // -// A budget below maxWriterFootprint is raised to it. Anything less could never -// be acquired by even a single large writer, so it would not throttle uploads, -// it would stall every one of them until the wait cap expired. The floor is two -// buffers rather than one because a writer at its peak holds two: the one being -// filled and the one being uploaded. +// A budget below maxWriterFootprint is raised to it. One buffer is all a writer +// needs to finish, so the floor could be maxBufferSize and still be correct; +// two is the smallest figure that leaves the process room to do anything but +// serialise every push behind one buffer. At the floor itself that is two large +// uploads each filling a buffer rather than one going twice as fast: a second +// buffer is only ever taken while a buffer's worth of budget would still be +// free afterwards, which at 32MB it never is. func ConfigureUploads(budgetBytes int64, idleTimeout time.Duration) { if budgetBytes < maxWriterFootprint { slog.Warn("Upload buffer budget below one writer's peak, raising to the minimum", diff --git a/pkg/appview/storage/upload_budget_test.go b/pkg/appview/storage/upload_budget_test.go index 8da392d..acb8fab 100644 --- a/pkg/appview/storage/upload_budget_test.go +++ b/pkg/appview/storage/upload_budget_test.go @@ -3,7 +3,9 @@ package storage import ( "context" "errors" + "fmt" "sync" + "sync/atomic" "testing" "time" @@ -314,6 +316,259 @@ func TestUploadBudget_WriteLargerThanTheWholeBudgetFailsFast(t *testing.T) { } } +// TestUploadBudget_ThreeWritersShareABudgetForTwo pins the invariant that keeps +// the budget from wedging: a writer never waits on the budget for anything past +// its first buffer. +// +// The wedge it regresses: every writer charged its second buffer as that buffer +// grew, so writers that were all mid-growth could hold the entire budget while +// none of them had reached its peak. Nothing is released before Commit, no +// writer could reach Commit without more budget, and there was none, so every +// Write sat out the five minute wait cap and failed. Three writers with room +// for two is exactly that shape: 48MB of first buffers out of 64MB, then all +// three carrying on at once. +func TestUploadBudget_ThreeWritersShareABudgetForTwo(t *testing.T) { + s3Server := newMockS3Server(t, true) + defer s3Server.Close() + + holdServer := newMockHoldServer(t, s3Server.URL) + defer holdServer.Close() + + isolateUploads(t) + + const writers = 3 + withBudget(t, 2*maxWriterFootprint) // room for two writers at their peak + + store := createTestProxyBlobStore(t, holdServer.URL) + + // Two buffers each, so every writer flushes a part and then keeps writing, + // which is the only way to hold a second buffer at all. + data := generateTestData(2 * maxBufferSize) + dgst := digest.FromBytes(data) + + // Nobody goes past its first buffer until all three have one. That is the + // state the budget used to run out in, and it is the one a regression would + // hang in. + var firstBuffers sync.WaitGroup + firstBuffers.Add(writers) + done := make(chan error, writers) + + for i := range writers { + go func(i int) { + writer, err := store.Create(context.Background()) + if err != nil { + firstBuffers.Done() + done <- fmt.Errorf("writer %d: Create: %w", i, err) + return + } + + // Streamed rather than handed over whole, the way a PATCH body + // arrives, so the writers interleave their growth. + write := func(b []byte) error { + const chunk = 4 << 20 + for off := 0; off < len(b); off += chunk { + if _, err := writer.Write(b[off:min(off+chunk, len(b))]); err != nil { + return fmt.Errorf("writer %d: Write: %w", i, err) + } + } + return nil + } + + err = write(data[:maxBufferSize]) + firstBuffers.Done() + if err != nil { + writer.Cancel(context.Background()) + done <- err + return + } + firstBuffers.Wait() + + if err := write(data[maxBufferSize:]); err != nil { + writer.Cancel(context.Background()) + done <- err + return + } + + if _, err := writer.Commit(context.Background(), distribution.Descriptor{ + Digest: dgst, + Size: int64(len(data)), + }); err != nil { + done <- fmt.Errorf("writer %d: Commit: %w", i, err) + return + } + done <- nil + }(i) + } + + // Generous on purpose: the mock servers answer in microseconds, so the only + // way to reach this is a writer stuck on the budget. It exists to fail + // rather than sit through the acquire's own five minute cap. + deadline := time.After(60 * time.Second) + for range writers { + select { + case err := <-done: + if err != nil { + t.Fatalf("A writer failed while three shared a budget for two: %v", err) + } + case <-deadline: + t.Fatalf("Writers wedged on the budget: %d of %d bytes held with nothing able to release them", + bytesHeld(), uploadBudget.limit) + } + } + + if held := bytesHeld(); held != 0 { + t.Errorf("Expected the whole budget back once every writer committed, got %d bytes still held", held) + } +} + +// TestUploadBudget_PipelinesOnlyWhenTheBudgetIsFree pins both sides of the +// opportunistic second buffer: a writer that can have one keeps filling while +// its part goes up, and a writer that cannot still finishes the blob out of the +// one buffer it holds. +func TestUploadBudget_PipelinesOnlyWhenTheBudgetIsFree(t *testing.T) { + t.Run("free budget buys a second buffer", func(t *testing.T) { + s3Server := newMockS3Server(t, true) + defer s3Server.Close() + + holdServer := newMockHoldServer(t, s3Server.URL) + defer holdServer.Close() + + isolateUploads(t) + withBudget(t, defaultUploadBufferBudget) + + // Hold the first part on its way to S3, so the only way the write below + // can finish is in a second buffer. + release := make(chan struct{}) + s3Server.setPartHook(func(int) error { + <-release + return nil + }) + + store := createTestProxyBlobStore(t, holdServer.URL) + writer, err := store.Create(context.Background()) + if err != nil { + t.Fatalf("Create() failed: %v", err) + } + pbw := writer.(*ProxyBlobWriter) + + data := generateTestData(maxBufferSize + 4<<20) + wrote := make(chan error, 1) + go func() { + _, werr := writer.Write(data) + wrote <- werr + }() + + select { + case err := <-wrote: + if err != nil { + t.Fatalf("Write() failed: %v", err) + } + case <-time.After(30 * time.Second): + close(release) + t.Fatal("Write waited for the part it handed off instead of carrying on in a second buffer") + } + + // The part is still going up, so both buffers are the writer's and both + // are charged. + if held := bytesHeld(); held != maxWriterFootprint { + t.Errorf("Expected both buffers charged while a part is in flight, %d, got %d", maxWriterFootprint, held) + } + pbw.mu.Lock() + pipelined := pbw.flight != nil + pbw.mu.Unlock() + if !pipelined { + t.Error("Expected the part to still be in flight behind the second buffer") + } + + close(release) + s3Server.setPartHook(nil) + + if _, err := writer.Commit(context.Background(), distribution.Descriptor{ + Digest: digest.FromBytes(data), + Size: int64(len(data)), + }); err != nil { + t.Fatalf("Commit() failed: %v", err) + } + if held := bytesHeld(); held != 0 { + t.Errorf("Expected the budget back after Commit, got %d bytes still held", held) + } + }) + + t.Run("a budget of one buffer still finishes the blob", func(t *testing.T) { + s3Server := newMockS3Server(t, true) + defer s3Server.Close() + + holdServer := newMockHoldServer(t, s3Server.URL) + defer holdServer.Close() + + isolateUploads(t) + withBudget(t, maxBufferSize) // exactly one buffer: no room to pipeline + + store := createTestProxyBlobStore(t, holdServer.URL) + writer, err := store.Create(context.Background()) + if err != nil { + t.Fatalf("Create() failed: %v", err) + } + pbw := writer.(*ProxyBlobWriter) + + // Sample what the budget reports throughout: the writer must never hold + // more than the single buffer it was able to charge. + var peak atomic.Int64 + stop := make(chan struct{}) + var sampler sync.WaitGroup + sampler.Add(1) + go func() { + defer sampler.Done() + for { + select { + case <-stop: + return + default: + if held := bytesHeld(); held > peak.Load() { + peak.Store(held) + } + } + } + }() + + data := generateTestData(3 * maxBufferSize) + for off := 0; off < len(data); off += 1 << 20 { + if _, err := writer.Write(data[off:min(off+(1<<20), len(data))]); err != nil { + close(stop) + sampler.Wait() + t.Fatalf("Write() failed with a budget of one buffer: %v", err) + } + } + + if other := pbw.otherBufferCap(); other != 0 { + t.Errorf("Expected a writer under pressure to hold one buffer, found a second of %d bytes", other) + } + + desc, err := writer.Commit(context.Background(), distribution.Descriptor{ + Digest: digest.FromBytes(data), + Size: int64(len(data)), + }) + close(stop) + sampler.Wait() + if err != nil { + t.Fatalf("Commit() failed with a budget of one buffer: %v", err) + } + if desc.Size != int64(len(data)) { + t.Errorf("Committed size %d, expected %d", desc.Size, len(data)) + } + + if peak.Load() > maxBufferSize { + t.Errorf("Expected never to hold more than one buffer, peaked at %d", peak.Load()) + } + if parts := s3Server.finishedParts(); len(parts) != 3 { + t.Errorf("Expected the blob to go up as 3 parts, got %v", parts) + } + if held := bytesHeld(); held != 0 { + t.Errorf("Expected the budget back after Commit, got %d bytes still held", held) + } + }) +} + // TestSweep_ReapsIdleWritersAndLeavesLiveOnes pins the sweep: an abandoned // upload loses its buffer, its budget, its hold-side multipart session and its // slot in the map, while an upload that is merely slow is untouched.