mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-08 15:41:15 +02:00
Chris Lu
5f787a25c3
* master: survive a layout deleted twice Two volume servers dropping the last replica of volumes that share a layout both find it empty and both delete it. The loser's lookup misses, and the single-value type assertion on the result crashed the master before the caller could look at the found flag. Claude-Session: https://claude.ai/code/session_01WmX6Rchx298NQksHDXg7sk * master: remove a layout and read it back in one step DeleteVolumeLayout looked the layout up and then deleted it, so two deleters could each release the lookup ownership of the same layout, or one could find nothing to release at all. Have the map hand back what it removed. Claude-Session: https://claude.ai/code/session_01WmX6Rchx298NQksHDXg7sk
108 lines
2.9 KiB
Go
108 lines
2.9 KiB
Go
package topology
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/types"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
)
|
|
|
|
type Collection struct {
|
|
Name string
|
|
volumeSizeLimit uint64
|
|
replicationAsMin bool
|
|
storageType2VolumeLayout *util.ConcurrentReadMap
|
|
}
|
|
|
|
func NewCollection(name string, volumeSizeLimit uint64, replicationAsMin bool) *Collection {
|
|
c := &Collection{
|
|
Name: name,
|
|
volumeSizeLimit: volumeSizeLimit,
|
|
replicationAsMin: replicationAsMin,
|
|
}
|
|
c.storageType2VolumeLayout = util.NewConcurrentReadMap()
|
|
return c
|
|
}
|
|
|
|
func (c *Collection) String() string {
|
|
return fmt.Sprintf("Name:%s, volumeSizeLimit:%d, storageType2VolumeLayout:%v", c.Name, c.volumeSizeLimit, c.storageType2VolumeLayout)
|
|
}
|
|
|
|
func (c *Collection) GetOrCreateVolumeLayout(rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) *VolumeLayout {
|
|
keyString := rp.String()
|
|
if ttl != nil {
|
|
keyString += ttl.String()
|
|
}
|
|
if diskType != types.HardDriveType {
|
|
keyString += string(diskType)
|
|
}
|
|
vl := c.storageType2VolumeLayout.Get(keyString, func() interface{} {
|
|
return NewVolumeLayout(rp, ttl, diskType, c.volumeSizeLimit, c.replicationAsMin)
|
|
})
|
|
return vl.(*VolumeLayout)
|
|
}
|
|
|
|
func (c *Collection) GetVolumeLayout(rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) (*VolumeLayout, bool) {
|
|
keyString := rp.String()
|
|
if ttl != nil {
|
|
keyString += ttl.String()
|
|
}
|
|
if diskType != types.HardDriveType {
|
|
keyString += string(diskType)
|
|
}
|
|
vl, ok := c.storageType2VolumeLayout.Find(keyString)
|
|
if !ok {
|
|
return nil, false
|
|
}
|
|
return vl.(*VolumeLayout), true
|
|
}
|
|
|
|
func (c *Collection) GetAllVolumeLayouts() []*VolumeLayout {
|
|
var vls []*VolumeLayout
|
|
for _, vl := range c.storageType2VolumeLayout.Items() {
|
|
if vl != nil {
|
|
vls = append(vls, vl.(*VolumeLayout))
|
|
}
|
|
}
|
|
return vls
|
|
}
|
|
|
|
func (c *Collection) DeleteVolumeLayout(rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) {
|
|
keyString := rp.String()
|
|
if ttl != nil {
|
|
keyString += ttl.String()
|
|
}
|
|
if diskType != types.HardDriveType {
|
|
keyString += string(diskType)
|
|
}
|
|
// Unpublish first so a racing registration re-resolves into a fresh layout.
|
|
if vl, found := c.storageType2VolumeLayout.Delete(keyString); found {
|
|
vl.(*VolumeLayout).releaseLookupOwnership()
|
|
}
|
|
}
|
|
|
|
func (c *Collection) Lookup(vid needle.VolumeId) []*DataNode {
|
|
for _, vl := range c.storageType2VolumeLayout.Items() {
|
|
if vl != nil {
|
|
if list := vl.(*VolumeLayout).Lookup(vid); list != nil {
|
|
return list
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (c *Collection) ListVolumeServers() (nodes []*DataNode) {
|
|
for _, vl := range c.storageType2VolumeLayout.Items() {
|
|
if vl != nil {
|
|
if list := vl.(*VolumeLayout).ListVolumeServers(); list != nil {
|
|
nodes = append(nodes, list...)
|
|
}
|
|
}
|
|
}
|
|
return
|
|
}
|