gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Refactor: remove unnecessary interfaces and methods from raft store

Browse Source 
Signed-off-by: Mofei Zhang <mofei2816@gmail.com>
This commit is contained in:
Mofei Zhang 2020-02-10 15:55:26 +08:00
parent 29ce3f6c5c
commit 3e2dcc58e3
No known key found for this signature in database
GPG Key ID: 7C19725C197AE672
13 changed files with 34 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
master/metadata_fsm.go
View File

@ -78,7 +78,7 @@ func (mf *MetadataFsm) restore() {
func (mf *MetadataFsm) restoreApplied() {
value, err := mf.Get(applied)
value, err := mf.store.Get(applied)
if err != nil {
panic(fmt.Sprintf("Failed to restore applied err:%v", err.Error()))
}
@ -125,7 +125,7 @@ func (mf *MetadataFsm) Apply(command []byte, index uint64) (resp interface{}, er
panic(err)
}
default:
if err = mf.batchPut(cmdMap); err != nil {
if err = mf.store.BatchPut(cmdMap, true); err != nil {
panic(err)
}
}
@ -198,25 +198,6 @@ func (mf *MetadataFsm) HandleLeaderChange(leader uint64) {
}
}
// Put implements the interface of raft.StateMachine
func (mf *MetadataFsm) Put(key, val interface{}) (interface{}, error) {
return mf.store.Put(key, val, true)
}
func (mf *MetadataFsm) batchPut(cmdMap map[string][]byte) (err error) {
return mf.store.BatchPut(cmdMap, true)
}
// Get implements the interface of raft.StateMachine
func (mf *MetadataFsm) Get(key interface{}) (interface{}, error) {
return mf.store.Get(key)
}
// Del implements the interface of raft.StateMachine
func (mf *MetadataFsm) Del(key interface{}) (interface{}, error) {
return mf.store.Del(key, true)
}
func (mf *MetadataFsm) delKeyAndPutIndex(key string, cmdMap map[string][]byte) (err error) {
return mf.store.DeleteKeyAndPutIndex(key, cmdMap, true)
}

4
metanode/partition.go
View File

@ -548,7 +548,7 @@ func (mp *metaPartition) UpdatePartition(req *UpdatePartitionReq,
resp.Result = err.Error()
return
}
r, err := mp.Put(opFSMUpdatePartition, reqData)
r, err := mp.submit(opFSMUpdatePartition, reqData)
if err != nil {
resp.Status = proto.TaskFailed
resp.Result = err.Error()
@ -566,7 +566,7 @@ func (mp *metaPartition) UpdatePartition(req *UpdatePartitionReq,
}
func (mp *metaPartition) DecommissionPartition(req []byte) (err error) {
_, err = mp.Put(opFSMDecommissionPartition, req)
_, err = mp.submit(opFSMDecommissionPartition, req)
return
}

9
metanode/partition_delete_extents.go
View File

@ -19,14 +19,15 @@ import (
"container/list"
"encoding/binary"
"fmt"
"github.com/chubaofs/chubaofs/proto"
"github.com/chubaofs/chubaofs/util/log"
"io"
"io/ioutil"
"os"
"path"
"strings"
"time"
"github.com/chubaofs/chubaofs/proto"
"github.com/chubaofs/chubaofs/util/log"
)
const (
@ -198,7 +199,7 @@ func (mp *metaPartition) deleteExtentsFromList(fileList *list.List) {
status := mp.raftPartition.Status()
if status.State == "StateLeader" && !status.
RestoringSnapshot {
if _, err = mp.Put(opFSMInternalDelExtentFile,
if _, err = mp.submit(opFSMInternalDelExtentFile,
[]byte(fileName)); err != nil {
log.LogErrorf(
"[deleteExtentsFromList] partitionId=%d,"+
@ -243,7 +244,7 @@ func (mp *metaPartition) deleteExtentsFromList(fileList *list.List) {
}
buff.Reset()
buff.WriteString(fmt.Sprintf("%s %d", fileName, cursor))
if _, err = mp.Put(opFSMInternalDelExtentCursor, buff.Bytes()); err != nil {
if _, err = mp.submit(opFSMInternalDelExtentCursor, buff.Bytes()); err != nil {
log.LogWarnf("[deleteExtentsFromList] partitionId=%d, %s",
mp.config.PartitionId, err.Error())
}

18
metanode/partition_fsm.go
View File

@ -347,11 +347,11 @@ func (mp *metaPartition) HandleLeaderChange(leader uint64) {
}
// Put puts the given key-value pair (operation key and operation request) into the raft store.
func (mp *metaPartition) Put(key, val interface{}) (resp interface{}, err error) {
func (mp *metaPartition) submit(op uint32, data []byte) (resp interface{}, err error) {
snap := NewMetaItem(0, nil, nil)
snap.Op = key.(uint32)
if val != nil {
snap.V = val.([]byte)
snap.Op = op
if data != nil {
snap.V = data
}
cmd, err := snap.MarshalJson()
if err != nil {
@ -363,16 +363,6 @@ func (mp *metaPartition) Put(key, val interface{}) (resp interface{}, err error)
return
}
// Get has not been implemented yet.
func (mp *metaPartition) Get(key interface{}) (interface{}, error) {
return nil, nil
}
// Del has not been implemented yet.
func (mp *metaPartition) Del(key interface{}) (interface{}, error) {
return nil, nil
}
func (mp *metaPartition) uploadApplyID(applyId uint64) {
atomic.StoreUint64(&mp.applyID, applyId)
}

2
metanode/partition_fsmop.go
View File

@ -48,7 +48,7 @@ func (mp *metaPartition) initInode(ino *Inode) {
log.LogFatalf("[initInode] marshal: %s", err.Error())
}
// put first root inode
resp, err := mp.Put(opFSMCreateInode, data)
resp, err := mp.submit(opFSMCreateInode, data)
if err != nil {
log.LogFatalf("[initInode] raft sync: %s", err.Error())
}

7
metanode/partition_op_dentry.go
View File

@ -17,6 +17,7 @@ package metanode
import (
"encoding/json"
"fmt"
"github.com/chubaofs/chubaofs/proto"
)
@ -38,7 +39,7 @@ func (mp *metaPartition) CreateDentry(req *CreateDentryReq, p *Packet) (err erro
if err != nil {
return
}
resp, err := mp.Put(opFSMCreateDentry, val)
resp, err := mp.submit(opFSMCreateDentry, val)
if err != nil {
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
return
@ -58,7 +59,7 @@ func (mp *metaPartition) DeleteDentry(req *DeleteDentryReq, p *Packet) (err erro
p.ResultCode = proto.OpErr
return
}
r, err := mp.Put(opFSMDeleteDentry, val)
r, err := mp.submit(opFSMDeleteDentry, val)
if err != nil {
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
return
@ -95,7 +96,7 @@ func (mp *metaPartition) UpdateDentry(req *UpdateDentryReq, p *Packet) (err erro
p.ResultCode = proto.OpErr
return
}
resp, err := mp.Put(opFSMUpdateDentry, val)
resp, err := mp.submit(opFSMUpdateDentry, val)
if err != nil {
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
return

3
metanode/partition_op_extend.go
View File

@ -16,6 +16,7 @@ package metanode
import (
"encoding/json"
"github.com/chubaofs/chubaofs/proto"
)
@ -126,6 +127,6 @@ func (mp *metaPartition) putExtend(op uint32, extend *Extend) (resp interface{},
if marshaled, err = extend.Bytes(); err != nil {
return
}
resp, err = mp.Put(op, marshaled)
resp, err = mp.submit(op, marshaled)
return
}

6
metanode/partition_op_extent.go
View File

@ -32,7 +32,7 @@ func (mp *metaPartition) ExtentAppend(req *proto.AppendExtentKeyRequest, p *Pack
p.PacketErrorWithBody(proto.OpErr, nil)
return
}
resp, err := mp.Put(opFSMExtentsAdd, val)
resp, err := mp.submit(opFSMExtentsAdd, val)
if err != nil {
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
return
@ -80,7 +80,7 @@ func (mp *metaPartition) ExtentsTruncate(req *ExtentsTruncateReq,
p.PacketErrorWithBody(proto.OpErr, nil)
return
}
resp, err := mp.Put(opFSMExtentTruncate, val)
resp, err := mp.submit(opFSMExtentTruncate, val)
if err != nil {
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
return
@ -108,7 +108,7 @@ func (mp *metaPartition) BatchExtentAppend(req *proto.AppendExtentKeysRequest, p
p.PacketErrorWithBody(proto.OpErr, nil)
return
}
resp, err := mp.Put(opFSMExtentsAdd, val)
resp, err := mp.submit(opFSMExtentsAdd, val)
if err != nil {
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
return

12
metanode/partition_op_inode.go
View File

@ -60,7 +60,7 @@ func (mp *metaPartition) CreateInode(req *CreateInoReq, p *Packet) (err error) {
p.PacketErrorWithBody(proto.OpErr, []byte(err.Error()))
return
}
resp, err := mp.Put(opFSMCreateInode, val)
resp, err := mp.submit(opFSMCreateInode, val)
if err != nil {
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
return
@ -94,7 +94,7 @@ func (mp *metaPartition) UnlinkInode(req *UnlinkInoReq, p *Packet) (err error) {
p.PacketErrorWithBody(proto.OpErr, nil)
return
}
r, err := mp.Put(opFSMUnlinkInode, val)
r, err := mp.submit(opFSMUnlinkInode, val)
if err != nil {
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
return
@ -171,7 +171,7 @@ func (mp *metaPartition) CreateInodeLink(req *LinkInodeReq, p *Packet) (err erro
p.PacketErrorWithBody(proto.OpErr, []byte(err.Error()))
return
}
resp, err := mp.Put(opFSMCreateLinkInode, val)
resp, err := mp.submit(opFSMCreateLinkInode, val)
if err != nil {
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
return
@ -204,7 +204,7 @@ func (mp *metaPartition) EvictInode(req *EvictInodeReq, p *Packet) (err error) {
p.PacketErrorWithBody(proto.OpErr, []byte(err.Error()))
return
}
resp, err := mp.Put(opFSMEvictInode, val)
resp, err := mp.submit(opFSMEvictInode, val)
if err != nil {
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
return
@ -216,7 +216,7 @@ func (mp *metaPartition) EvictInode(req *EvictInodeReq, p *Packet) (err error) {
// SetAttr set the inode attributes.
func (mp *metaPartition) SetAttr(reqData []byte, p *Packet) (err error) {
_, err = mp.Put(opFSMSetAttr, reqData)
_, err = mp.submit(opFSMSetAttr, reqData)
if err != nil {
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
return
@ -237,7 +237,7 @@ func (mp *metaPartition) DeleteInode(req *proto.DeleteInodeRequest, p *Packet) (
p.ResultCode = proto.OpErr
return
}
_, err = mp.Put(opFSMInternalDeleteInode, encoded)
_, err = mp.submit(opFSMInternalDeleteInode, encoded)
if err != nil {
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
return

2
metanode/partition_op_multipart.go
View File

@ -212,6 +212,6 @@ func (mp *metaPartition) putMultipart(op uint32, multipart *Multipart) (resp int
if encoded, err = multipart.Bytes(); err != nil {
return
}
resp, err = mp.Put(op, encoded)
resp, err = mp.submit(op, encoded)
return
}

2
metanode/partition_store_ticket.go
View File

@ -110,7 +110,7 @@ func (mp *metaPartition) startSchedule(curIndex uint64) {
timer.Reset(intervalToPersistData)
continue
}
if _, err := mp.Put(opFSMStoreTick, nil); err != nil {
if _, err := mp.submit(opFSMStoreTick, nil); err != nil {
log.LogErrorf("[startSchedule] raft submit: %s", err.Error())
if _, ok := mp.IsLeader(); ok {
timer.Reset(intervalToPersistData)

8
raftstore/partition.go
View File

@ -15,9 +15,10 @@
package raftstore
import (
"os"
"github.com/tiglabs/raft"
"github.com/tiglabs/raft/proto"
"os"
)
// PartitionStatus is a type alias of raft.Status
@ -26,10 +27,7 @@ type PartitionStatus = raft.Status
// PartitionFsm wraps necessary methods include both FSM implementation
// and data storage operation for raft store partition.
// It extends from raft StateMachine and Store.
type PartitionFsm interface {
raft.StateMachine
Store
}
type PartitionFsm = raft.StateMachine
// Partition wraps necessary methods for raft store partition operation.
// Partition is a shard for multi-raft in RaftSore. RaftStore is based on multi-raft which

22
raftstore/store.go
View File

@ -1,22 +0,0 @@
// Copyright 2018 The Chubao Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package raftstore
// Store is the interface that defines the abstract and necessary methods for storage operation.
type Store interface {
Put(key, val interface{}) (interface{}, error)
Get(key interface{}) (interface{}, error)
Del(key interface{}) (interface{}, error)
}