mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-27 07:20:44 +01:00
filer: shut down test filers' meta-log buffers, fixing the 32-bit OOM (#12104)
* filer: shut down test filers' meta-log buffers, fixing the 32-bit OOM Each log_buffer.NewLogBuffer preallocates five 8 MiB windows and spawns loopFlush/loopInterval goroutines that keep them reachable. The weed/filer tests built ~70 filers and aggregators that almost never called ShutdownLogBuffer, pinning ~2.8 GB plus live churn, and the linux/386 job ran out of its ~4 GiB address space inside newSealedBuffers. - MetaAggregator.Shutdown now retires MetaLogBuffer too: the "aggr" buffer's loops previously outlived the aggregator. - newTestFiler and newTestFilerWithStubStore register t.Cleanup shutdowns; the one-off filer/aggregator literals defer theirs. * filer: wait for the aggregated meta-log buffer to stop in Shutdown ShutdownLogBuffer only signals the stop; loopFlush and its pooled windows exit asynchronously. Match Filer.Shutdown's local-buffer sequence so the aggregator returns with the buffer's memory already released. * filer: cancel test filers' flush context before buffer shutdown The stub-store helpers build real NewFiler instances, so the queued shutdown flush retries appendToFile every ~737ms under flushCtx, which nothing ever cancelled — the buffer and its pool stayed alive past each test. Cancelling first lets the flush drop the window and WaitForShutdown confirms loopFlush exited. * filer: wait for test log buffers to drain after shutdown The stub-flush buffers cannot hang, but without WaitForShutdown their loopFlush goroutines release the pooled windows asynchronously — the memory this PR means to free is still counted as live while the next test allocates.
This commit is contained in:
1 parent
aff18b2336
commit
6b379badbb
8 files changed
+55
-12
No files matched your search
@@ -125,6 +125,10 @@ func newFilerWithFakeMaster(t *testing.T) (*Filer, *hookedStore, *collectionDele
|
||||
LocalMetaLogBuffer: log_buffer.NewLogBuffer("test", time.Minute,
|
||||
func(*log_buffer.LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, func() {}),
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
f.LocalMetaLogBuffer.ShutdownLogBuffer()
|
||||
f.LocalMetaLogBuffer.WaitForShutdown()
|
||||
})
|
||||
return f, store, master
|
||||
}
|
||||
|
||||
|
||||
@@ -103,15 +103,26 @@ func TestEnsureEntryInodeHardLinkFitsSignedLong(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func newTestFilerWithStubStore() (*Filer, *stubFilerStore) {
|
||||
func newTestFilerWithStubStore(t *testing.T) (*Filer, *stubFilerStore) {
|
||||
t.Helper()
|
||||
store := newStubFilerStore()
|
||||
f := NewFiler(pb.ServerDiscovery{}, nil, "", "", "", "", "", 255, nil)
|
||||
f.Store = NewFilerStoreWrapper(store)
|
||||
// A live LogBuffer keeps its flush loops and their multi-buffer pool
|
||||
// reachable; without this the package's filers pin gigabytes on 32-bit.
|
||||
// There is no master to flush to, so cancel the flush context first:
|
||||
// otherwise the queued window retries forever and the buffer never exits.
|
||||
t.Cleanup(func() {
|
||||
close(f.deletionQuit)
|
||||
f.flushCancel()
|
||||
f.LocalMetaLogBuffer.ShutdownLogBuffer()
|
||||
f.LocalMetaLogBuffer.WaitForShutdown()
|
||||
})
|
||||
return f, store
|
||||
}
|
||||
|
||||
func TestCreateEntryAssignsInodeWhenMissing(t *testing.T) {
|
||||
f, store := newTestFilerWithStubStore()
|
||||
f, store := newTestFilerWithStubStore(t)
|
||||
|
||||
entry := &Entry{
|
||||
FullPath: util.FullPath("/dir/file.txt"),
|
||||
@@ -131,7 +142,7 @@ func TestCreateEntryAssignsInodeWhenMissing(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestCreateEntryAssignsInodesToAutoCreatedParents(t *testing.T) {
|
||||
f, store := newTestFilerWithStubStore()
|
||||
f, store := newTestFilerWithStubStore(t)
|
||||
|
||||
entry := &Entry{
|
||||
FullPath: util.FullPath("/a/b/c.txt"),
|
||||
@@ -152,7 +163,7 @@ func TestCreateEntryAssignsInodesToAutoCreatedParents(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestCreateEntryOExclPreservesExistingEntry(t *testing.T) {
|
||||
f, store := newTestFilerWithStubStore()
|
||||
f, store := newTestFilerWithStubStore(t)
|
||||
|
||||
original := &Entry{
|
||||
FullPath: util.FullPath("/buckets/my-bucket"),
|
||||
@@ -177,7 +188,7 @@ func TestCreateEntryOExclPreservesExistingEntry(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestCreateEntryOExclFailsOnLookupError(t *testing.T) {
|
||||
f, store := newTestFilerWithStubStore()
|
||||
f, store := newTestFilerWithStubStore(t)
|
||||
|
||||
original := &Entry{
|
||||
FullPath: util.FullPath("/buckets/my-bucket"),
|
||||
@@ -203,7 +214,7 @@ func TestCreateEntryOExclFailsOnLookupError(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestUpdateEntryPreservesExistingInode(t *testing.T) {
|
||||
f, store := newTestFilerWithStubStore()
|
||||
f, store := newTestFilerWithStubStore(t)
|
||||
|
||||
original := &Entry{
|
||||
FullPath: util.FullPath("/doc.txt"),
|
||||
@@ -240,7 +251,7 @@ func TestUpdateEntryPreservesExistingInode(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestUpdateEntryBackfillsMissingLegacyInode(t *testing.T) {
|
||||
f, store := newTestFilerWithStubStore()
|
||||
f, store := newTestFilerWithStubStore(t)
|
||||
|
||||
original := &Entry{
|
||||
FullPath: util.FullPath("/legacy.txt"),
|
||||
@@ -270,7 +281,7 @@ func TestUpdateEntryBackfillsMissingLegacyInode(t *testing.T) {
|
||||
// missing entry: on upsert stores proceeding would replace the entry without
|
||||
// loading the old chunks, orphaning them from cleanup.
|
||||
func TestCreateEntryFailsWhenLookupErrors(t *testing.T) {
|
||||
f, store := newTestFilerWithStubStore()
|
||||
f, store := newTestFilerWithStubStore(t)
|
||||
path := util.FullPath("/dir/file.txt")
|
||||
require.NoError(t, store.InsertEntry(context.Background(), &Entry{
|
||||
FullPath: path,
|
||||
|
||||
@@ -329,6 +329,12 @@ func newTestFiler(t *testing.T, store *stubFilerStore, rs *FilerRemoteStorage) *
|
||||
LocalMetaLogBuffer: log_buffer.NewLogBuffer("test", time.Minute,
|
||||
func(*log_buffer.LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, func() {}),
|
||||
}
|
||||
// A live LogBuffer keeps its flush loops and their multi-buffer pool
|
||||
// reachable; without this the package's filers pin gigabytes on 32-bit.
|
||||
t.Cleanup(func() {
|
||||
f.LocalMetaLogBuffer.ShutdownLogBuffer()
|
||||
f.LocalMetaLogBuffer.WaitForShutdown()
|
||||
})
|
||||
return f
|
||||
}
|
||||
|
||||
|
||||
@@ -16,7 +16,7 @@ import (
|
||||
// TestPromoteToPrefixObject covers a key written before the keys nested under it:
|
||||
// the file becomes the directory they live in, and has to stay an object of its own.
|
||||
func TestPromoteToPrefixObject(t *testing.T) {
|
||||
f, store := newTestFilerWithStubStore()
|
||||
f, store := newTestFilerWithStubStore(t)
|
||||
ctx := context.Background()
|
||||
|
||||
object := &Entry{
|
||||
@@ -49,7 +49,7 @@ func TestPromoteToPrefixObject(t *testing.T) {
|
||||
// TestExpiredDirectoryIsNotDeletedOnRead pins the other half: a TTL an older build
|
||||
// left on a promoted directory must not take the keys under it with it.
|
||||
func TestExpiredDirectoryIsNotDeletedOnRead(t *testing.T) {
|
||||
f, store := newTestFilerWithStubStore()
|
||||
f, store := newTestFilerWithStubStore(t)
|
||||
ctx := context.Background()
|
||||
|
||||
dirPath := util.FullPath("/buckets/bkt/a/foo")
|
||||
@@ -80,7 +80,7 @@ func TestExpiredDirectoryIsNotDeletedOnRead(t *testing.T) {
|
||||
// entries become invisible and their metadata row is removed without a
|
||||
// cache-specific transaction or reverse Volume index.
|
||||
func TestExpiredFileIsDeletedOnRead(t *testing.T) {
|
||||
f, store := newTestFilerWithStubStore()
|
||||
f, store := newTestFilerWithStubStore(t)
|
||||
ctx := context.Background()
|
||||
|
||||
filePath := util.FullPath("/buckets/lmcache/expired")
|
||||
@@ -106,7 +106,7 @@ func TestExpiredFileIsDeletedOnRead(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestListDirectoryEntriesContextCanceled(t *testing.T) {
|
||||
f, store := newTestFilerWithStubStore()
|
||||
f, store := newTestFilerWithStubStore(t)
|
||||
ctx := context.Background()
|
||||
|
||||
require.NoError(t, store.InsertEntry(ctx, &Entry{
|
||||
|
||||
@@ -389,6 +389,13 @@ func (ma *MetaAggregator) Shutdown() {
|
||||
for {
|
||||
select {
|
||||
case <-done:
|
||||
// The peers have stopped writing; retire the aggregated buffer so
|
||||
// its flush loops and sealed windows are released with the
|
||||
// aggregator instead of outliving it.
|
||||
if ma.MetaLogBuffer != nil {
|
||||
ma.MetaLogBuffer.ShutdownLogBuffer()
|
||||
ma.MetaLogBuffer.WaitForShutdown()
|
||||
}
|
||||
return
|
||||
case <-ticker.C:
|
||||
glog.Warningf("meta aggregator shutdown: peer subscriptions still stopping")
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
func TestOnPeerUpdateRepeatedAdd(t *testing.T) {
|
||||
peer := pb.ServerAddress("127.0.0.1:1")
|
||||
ma := NewMetaAggregator(nil, pb.ServerAddress("127.0.0.1:2"), nil)
|
||||
defer ma.Shutdown()
|
||||
|
||||
add := &master_pb.ClusterNodeUpdate{Address: string(peer), IsAdd: true}
|
||||
ma.OnPeerUpdate(add, time.Now())
|
||||
|
||||
@@ -103,6 +103,12 @@ func TestViperReadsMetaLogOverrides(t *testing.T) {
|
||||
}()
|
||||
|
||||
f := NewFiler(pb.ServerDiscovery{}, nil, "", "", "", "", "", 255, nil)
|
||||
defer func() {
|
||||
close(f.deletionQuit)
|
||||
f.flushCancel()
|
||||
f.LocalMetaLogBuffer.ShutdownLogBuffer()
|
||||
f.LocalMetaLogBuffer.WaitForShutdown()
|
||||
}()
|
||||
if got := f.metaLogTargetCollection; got != "filer-meta" {
|
||||
t.Fatalf("NewFiler did not read the collection override: %q", got)
|
||||
}
|
||||
|
||||
@@ -22,6 +22,10 @@ func TestNotifyUpdateEventRecordsRequestMetadataEvent(t *testing.T) {
|
||||
nil,
|
||||
),
|
||||
}
|
||||
defer func() {
|
||||
f.LocalMetaLogBuffer.ShutdownLogBuffer()
|
||||
f.LocalMetaLogBuffer.WaitForShutdown()
|
||||
}()
|
||||
|
||||
ctx, sink := WithMetadataEventSink(context.Background())
|
||||
f.NotifyUpdateEvent(ctx, &Entry{FullPath: util.FullPath("/dir/file.txt")}, nil, true, false, []int32{7})
|
||||
@@ -56,6 +60,10 @@ func TestNotifyUpdateEventReloadsLocalFilerConfiguration(t *testing.T) {
|
||||
nil,
|
||||
),
|
||||
}
|
||||
defer func() {
|
||||
f.LocalMetaLogBuffer.ShutdownLogBuffer()
|
||||
f.LocalMetaLogBuffer.WaitForShutdown()
|
||||
}()
|
||||
|
||||
updatedConf := NewFilerConf()
|
||||
if err := updatedConf.SetLocationConf(&filer_pb.FilerConf_PathConf{
|
||||
|
||||
Reference in new issue
Block a user