-
Notifications
You must be signed in to change notification settings - Fork 3.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
ccl/backupccl: add memory monitor to external SST iterators in restore
Previously, there was no limit on the amount of memory that can be used while constructing edternal SST iterators during restore. This patch adds a memory monitor to limit the amount of memory that can be used to construct external SST iterators. If a restore processor fails to acquire enough memory to open the next file for a restore span, it will send the iterator for all of the open files it has accumulated so far, and wait until it can acquire the memory to resume constructing the iterator for the remaining files. The memory limit can be controlled via the new cluster setting bulkio.restore.per_processor_memory_limit. Regardless of the setting, however, the amount of memory used will not exceed COCKROACH_RESTORE_MEM_FRACTION * max SQL memory. The new environment variable COCKROACH_RESTORE_MEM_FRACTION defaults to 0.5. Release note: None
- Loading branch information
Rui Hu
committed
Apr 3, 2023
1 parent
0745cd4
commit fca9148
Showing
20 changed files
with
945 additions
and
93 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,12 +1,38 @@ | ||
| load("//build/bazelutil/unused_checker:unused.bzl", "get_x_data") | ||
| load("@io_bazel_rules_go//go:def.bzl", "go_library") | ||
| load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test") | ||
|
|
||
| go_library( | ||
| name = "backuputils", | ||
| srcs = ["utils.go"], | ||
| srcs = [ | ||
| "memory_backed_quota_pool.go", | ||
| "utils.go", | ||
| ], | ||
| importpath = "github.com/cockroachdb/cockroach/pkg/ccl/backupccl/backuputils", | ||
| visibility = ["//visibility:public"], | ||
| deps = ["//pkg/cloud"], | ||
| deps = [ | ||
| "//pkg/cloud", | ||
| "//pkg/util/mon", | ||
| "//pkg/util/quotapool", | ||
| "//pkg/util/syncutil", | ||
| "@com_github_cockroachdb_errors//:errors", | ||
| "@com_github_cockroachdb_redact//:redact", | ||
| ], | ||
| ) | ||
|
|
||
| go_test( | ||
| name = "backuputils_test", | ||
| srcs = ["memory_backed_quota_pool_test.go"], | ||
| args = ["-test.timeout=295s"], | ||
| embed = [":backuputils"], | ||
| tags = ["ccl_test"], | ||
| deps = [ | ||
| "//pkg/settings/cluster", | ||
| "//pkg/util/leaktest", | ||
| "//pkg/util/mon", | ||
| "//pkg/util/quotapool", | ||
| "//pkg/util/syncutil", | ||
| "@com_github_cockroachdb_errors//:errors", | ||
| ], | ||
| ) | ||
|
|
||
| get_x_data(name = "get_x_data") |
127 changes: 127 additions & 0 deletions
127
pkg/ccl/backupccl/backuputils/memory_backed_quota_pool.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| // Copyright 2023 The Cockroach Authors. | ||
| // | ||
| // Licensed as a CockroachDB Enterprise file under the Cockroach Community | ||
| // License (the "License"); you may not use this file except in compliance with | ||
| // the License. You may obtain a copy of the License at | ||
| // | ||
| // https://github.com/cockroachdb/cockroach/blob/master/licenses/CCL.txt | ||
|
|
||
| package backuputils | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/cockroachdb/cockroach/pkg/util/mon" | ||
| "github.com/cockroachdb/cockroach/pkg/util/quotapool" | ||
| "github.com/cockroachdb/cockroach/pkg/util/syncutil" | ||
| "github.com/cockroachdb/errors" | ||
| "github.com/cockroachdb/redact" | ||
| ) | ||
|
|
||
| // MemoryBackedQuotaPool is an IntPool backed up by a memory monitor. | ||
| // Users of MemoryBackedQuotaPool can acquire capacity from the IntPool, | ||
| // but the capacity of the IntPool can only be increased by acquiring | ||
| // the corresponding amount of memory from the backing memory monitor. | ||
| type MemoryBackedQuotaPool struct { | ||
| mon *mon.BytesMonitor | ||
| mem *mon.BoundAccount | ||
|
|
||
| // capacityMu synchronizes operations related to the capacity | ||
| // of quotaPool. | ||
| capacityMu syncutil.Mutex | ||
| quotaPool *quotapool.IntPool | ||
| } | ||
|
|
||
| // NewMemoryBackedQuotaPool creates a MemoryBackedQuotaPool from a | ||
| // parent monitor m with a limit. | ||
| func NewMemoryBackedQuotaPool( | ||
| ctx context.Context, m *mon.BytesMonitor, name redact.RedactableString, limit int64, | ||
| ) *MemoryBackedQuotaPool { | ||
| q := MemoryBackedQuotaPool{ | ||
| quotaPool: quotapool.NewIntPool(fmt.Sprintf("%s-pool", name), 0), | ||
| } | ||
|
|
||
| if m != nil { | ||
| q.mon = mon.NewMonitorInheritWithLimit(name, limit, m) | ||
| q.mon.StartNoReserved(ctx, m) | ||
| mem := q.mon.MakeBoundAccount() | ||
| q.mem = &mem | ||
| } | ||
| return &q | ||
| } | ||
|
|
||
| // TryAcquireMaybeIncreaseCapacity tries to acquire size from the pool. | ||
| // On success, a non-nil alloc is returned and Release() must be called on | ||
| // it to return the quota to the pool. | ||
| // If the acquire fails because of not enough quota, it will repeatedly | ||
| // attempt to increase the capacity of the pool until the acquire succeeds. | ||
| // If the capacity increase fails, then the function will return | ||
| // with the error quotapool.ErrNotEnoughQuota. | ||
| // | ||
| // Safe for concurrent use. | ||
| func (q *MemoryBackedQuotaPool) TryAcquireMaybeIncreaseCapacity( | ||
| ctx context.Context, size uint64, | ||
| ) (*quotapool.IntAlloc, error) { | ||
| for { | ||
| alloc, err := q.quotaPool.TryAcquire(ctx, size) | ||
| if err == nil || !errors.Is(err, quotapool.ErrNotEnoughQuota) { | ||
| return alloc, err | ||
| } | ||
|
|
||
| // Not enough quota, attempt to grow the memory to increase quota pool | ||
| // capacity | ||
| if !q.IncreaseCapacity(ctx, size) { | ||
| return nil, quotapool.ErrNotEnoughQuota | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Acquire acquires size from the pool. On success, a non-nil alloc is | ||
| // returned and Release() must be called on it to return the quota to the pool. | ||
| // | ||
| // Safe for concurrent use. | ||
| func (q *MemoryBackedQuotaPool) Acquire( | ||
| ctx context.Context, size uint64, | ||
| ) (*quotapool.IntAlloc, error) { | ||
| return q.quotaPool.Acquire(ctx, size) | ||
| } | ||
|
|
||
| // Release will release allocs back to the pool. | ||
| func (q *MemoryBackedQuotaPool) Release(allocs ...*quotapool.IntAlloc) { | ||
| q.quotaPool.Release(allocs...) | ||
| } | ||
|
|
||
| // IncreaseCapacity will attempt to increase the capacity of the pool. | ||
| // Returns true if the increase succeeds and false otherwise. | ||
| // | ||
| // Safe for concurrent use. | ||
| func (q *MemoryBackedQuotaPool) IncreaseCapacity(ctx context.Context, size uint64) bool { | ||
| if err := q.mem.Grow(ctx, int64(size)); err != nil { | ||
| return false | ||
| } | ||
|
|
||
| q.capacityMu.Lock() | ||
| defer q.capacityMu.Unlock() | ||
| q.quotaPool.UpdateCapacity(q.quotaPool.Capacity() + size) | ||
| return true | ||
| } | ||
|
|
||
| // Capacity returns the capacity of the pool. | ||
| func (q *MemoryBackedQuotaPool) Capacity() uint64 { | ||
| q.capacityMu.Lock() | ||
| defer q.capacityMu.Unlock() | ||
| return q.quotaPool.Capacity() | ||
| } | ||
|
|
||
| // Close closes the pool and returns the reserved memory to the backing | ||
| // memory monitor. | ||
| func (q *MemoryBackedQuotaPool) Close(ctx context.Context) { | ||
| if q.mem != nil { | ||
| q.mem.Close(ctx) | ||
| } | ||
|
|
||
| if q.mon != nil { | ||
| q.mon.Stop(ctx) | ||
| } | ||
| } |
Oops, something went wrong.