cubefs

1
// Copyright 2020 The CubeFS Authors.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
12
// implied. See the License for the specific language governing
13
// permissions and limitations under the License.
14

15
package wrapper
16

17
import (
18
	"errors"
19
	"strings"
20

21
	"github.com/cubefs/cubefs/util/log"
22
)
23

24
// This type defines the constructor used to create and initialize the selector.
25
type DataPartitionSelectorConstructor = func(param string) (DataPartitionSelector, error)
26

27
// DataPartitionSelector is the interface defines the methods necessary to implement
28
// a selector for data partition selecting.
29
type DataPartitionSelector interface {
30
	// Name return name of current selector instance.
31
	Name() string
32

33
	// Refresh refreshes current selector instance by specified data partitions.
34
	Refresh(partitions []*DataPartition) error
35

36
	// Select returns an data partition picked by selector.
37
	Select(excludes map[string]struct{}) (*DataPartition, error)
38

39
	// RemoveDP removes specified data partition.
40
	RemoveDP(partitionID uint64)
41

42
	// Count return number of data partitions held by selector.
43
	Count() int
44
}
45

46
var (
47
	dataPartitionSelectorConstructors = make(map[string]DataPartitionSelectorConstructor)
48

49
	ErrDuplicatedDataPartitionSelectorConstructor = errors.New("duplicated data partition selector constructor")
50
	ErrDataPartitionSelectorConstructorNotExist   = errors.New("data partition selector constructor not exist")
51
)
52

53
// RegisterDataPartitionSelector registers a selector constructor.
54
// Users can register their own defined selector through this method.
55
func RegisterDataPartitionSelector(name string, constructor DataPartitionSelectorConstructor) error {
56
	clearName := strings.TrimSpace(strings.ToLower(name))
57
	if _, exist := dataPartitionSelectorConstructors[clearName]; exist {
58
		return ErrDuplicatedDataPartitionSelectorConstructor
59
	}
60
	dataPartitionSelectorConstructors[clearName] = constructor
61
	return nil
62
}
63

64
func newDataPartitionSelector(name string, param string) (newDpSelector DataPartitionSelector, err error) {
65
	clearName := strings.TrimSpace(strings.ToLower(name))
66
	constructor, exist := dataPartitionSelectorConstructors[clearName]
67
	if !exist {
68
		return nil, ErrDataPartitionSelectorConstructorNotExist
69
	}
70
	return constructor(param)
71
}
72

73
func (w *Wrapper) initDpSelector() (err error) {
74
	w.dpSelectorChanged = false
75
	selectorName := w.dpSelectorName
76
	if strings.TrimSpace(selectorName) == "" {
77
		log.LogInfof("initDpSelector: can not find dp selector[%v], use default selector", w.dpSelectorName)
78
		selectorName = DefaultRandomSelectorName
79
	}
80
	var selector DataPartitionSelector
81
	if selector, err = newDataPartitionSelector(selectorName, w.dpSelectorParm); err != nil {
82
		log.LogErrorf("initDpSelector: dpSelector[%v] init failed caused by [%v], use default selector", w.dpSelectorName,
83
			err)
84
		return
85
	}
86
	w.dpSelector = selector
87
	return
88
}
89

90
func (w *Wrapper) refreshDpSelector(partitions []*DataPartition) {
91
	w.Lock.RLock()
92
	dpSelector := w.dpSelector
93
	dpSelectorChanged := w.dpSelectorChanged
94
	w.Lock.RUnlock()
95

96
	if dpSelectorChanged {
97
		selectorName := w.dpSelectorName
98
		if strings.TrimSpace(selectorName) == "" {
99
			log.LogWarnf("refreshDpSelector: can not find dp selector[%v], use default selector", w.dpSelectorName)
100
			selectorName = DefaultRandomSelectorName
101
		}
102
		newDpSelector, err := newDataPartitionSelector(selectorName, w.dpSelectorParm)
103
		if err != nil {
104
			log.LogErrorf("refreshDpSelector: change dpSelector to [%v %v] failed caused by [%v],"+
105
				" use last valid selector. Please change dpSelector config through master.",
106
				w.dpSelectorName, w.dpSelectorParm, err)
107
		} else {
108
			w.Lock.Lock()
109
			log.LogInfof("refreshDpSelector: change dpSelector to [%v %v]", w.dpSelectorName, w.dpSelectorParm)
110
			w.dpSelector = newDpSelector
111
			w.dpSelectorChanged = false
112
			dpSelector = newDpSelector
113
			w.Lock.Unlock()
114
		}
115
	}
116

117
	_ = dpSelector.Refresh(partitions)
118
}
119

120
// getDataPartitionForWrite returns an available data partition for write.
121
func (w *Wrapper) GetDataPartitionForWrite(exclude map[string]struct{}) (*DataPartition, error) {
122
	w.Lock.RLock()
123
	dpSelector := w.dpSelector
124
	w.Lock.RUnlock()
125

126
	return dpSelector.Select(exclude)
127
}
128

129
func (w *Wrapper) RemoveDataPartitionForWrite(partitionID uint64) {
130
	w.Lock.RLock()
131
	dpSelector := w.dpSelector
132
	w.Lock.RUnlock()
133

134
	if dpSelector.Count() <= 1 {
135
		return
136
	}
137

138
	dpSelector.RemoveDP(partitionID)
139
}
140