Signed-off-by: Lei Liu <lei01.liu@horizon.ai>
@ -15,12 +15,12 @@ func NewMemorySequencer() (m *MemorySequencer) {
return
}
func (m *MemorySequencer) NextFileId(count uint64) (uint64, uint64) {
func (m *MemorySequencer) NextFileId(count uint64) uint64 {
m.sequenceLock.Lock()
defer m.sequenceLock.Unlock()
ret := m.counter
m.counter += uint64(count)
return ret, count
m.counter += count
return ret
func (m *MemorySequencer) SetMax(seenValue uint64) {
@ -1,7 +1,7 @@
package sequence
type Sequencer interface {
NextFileId(count uint64) (uint64, uint64)
NextFileId(count uint64) uint64
SetMax(uint64)
Peek() uint64
@ -65,11 +65,17 @@ func (ms *MasterServer) findVolumeLocation(collection, vid string) operation.Loo
var err error
if ms.Topo.IsLeader() {
volumeId, newVolumeIdErr := needle.NewVolumeId(vid)
if newVolumeIdErr != nil {
err = fmt.Errorf("Unknown volume id %s", vid)
} else {
machines := ms.Topo.Lookup(collection, volumeId)
for _, loc := range machines {
locations = append(locations, operation.Location{Url: loc.Url(), PublicUrl: loc.PublicUrl})
err = newVolumeIdErr
if locations == nil {
err = fmt.Errorf("volume id %s not found", vid)
machines, getVidLocationsErr := ms.MasterClient.GetVidLocations(vid)
@ -125,7 +125,7 @@ func (t *Topology) PickForWrite(count uint64, option *VolumeGrowOption) (string,
if datanodes.Length() == 0 {
return "", 0, nil, fmt.Errorf("no writable volumes available for collection:%s replication:%s ttl:%s", option.Collection, option.ReplicaPlacement.String(), option.Ttl.String())
fileId, count := t.Sequence.NextFileId(count)
fileId := t.Sequence.NextFileId(count)
return needle.NewFileId(*vid, fileId, rand.Uint32()).String(), count, datanodes.Head(), nil