gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into mq

Browse Source
This commit is contained in:
chrislu
2024-06-09 12:14:57 -07:00
parent 25b2850e7d dbfbabac55
commit b5d36541d9
7 changed files with 62 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
weed/filer/filer_deletion.go
View File

@@ -2,7 +2,6 @@ package filer
import (
"github.com/seaweedfs/seaweedfs/weed/storage"
"math"
"strings"
"time"
@@ -138,47 +137,19 @@ func (f *Filer) DeleteChunksNotRecursive(chunks []*filer_pb.FileChunk) {
}
func (f *Filer) deleteChunksIfNotNew(oldEntry, newEntry *Entry) {
if oldEntry == nil {
return
var oldChunks, newChunks []*filer_pb.FileChunk
if oldEntry != nil {
oldChunks = oldEntry.GetChunks()
}
if newEntry == nil {
f.DeleteChunks(oldEntry.GetChunks())
return
if newEntry != nil {
newChunks = newEntry.GetChunks()
}
var toDelete []*filer_pb.FileChunk
newChunkIds := make(map[string]bool)
newDataChunks, newManifestChunks, err := ResolveChunkManifest(f.MasterClient.GetLookupFileIdFunction(),
newEntry.GetChunks(), 0, math.MaxInt64)
if err != nil {
glog.Errorf("Failed to resolve new entry chunks when delete old entry chunks. new: %s, old: %s",
newEntry.GetChunks(), oldEntry.Chunks)
return
}
for _, newChunk := range newDataChunks {
newChunkIds[newChunk.GetFileIdString()] = true
}
for _, newChunk := range newManifestChunks {
newChunkIds[newChunk.GetFileIdString()] = true
}
oldDataChunks, oldManifestChunks, err := ResolveChunkManifest(f.MasterClient.GetLookupFileIdFunction(),
oldEntry.GetChunks(), 0, math.MaxInt64)
toDelete, err := MinusChunks(f.MasterClient.GetLookupFileIdFunction(), oldChunks, newChunks)
if err != nil {
glog.Errorf("Failed to resolve old entry chunks when delete old entry chunks. new: %s, old: %s",
newEntry.GetChunks(), oldEntry.GetChunks())
newChunks, oldChunks)
return
}
for _, oldChunk := range oldDataChunks {
if _, found := newChunkIds[oldChunk.GetFileIdString()]; !found {
toDelete = append(toDelete, oldChunk)
}
}
for _, oldChunk := range oldManifestChunks {
if _, found := newChunkIds[oldChunk.GetFileIdString()]; !found {
toDelete = append(toDelete, oldChunk)
}
}
f.DeleteChunksNotRecursive(toDelete)
}

8
weed/server/master_grpc_server.go
View File

@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"github.com/google/uuid"
"github.com/seaweedfs/seaweedfs/weed/cluster"
"net"
"sort"
@@ -260,7 +261,12 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ
return ms.informNewLeader(stream)
}
peerAddress := pb.ServerAddress(req.ClientAddress)
clientAddress := req.ClientAddress
// Ensure that the clientAddress is unique.
if clientAddress == "" {
clientAddress = uuid.New().String()
}
peerAddress := pb.ServerAddress(clientAddress)
// buffer by 1 so we don't end up getting stuck writing to stopChan forever
stopChan := make(chan bool, 1)