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

@formatter:off

Browse Source 
fix(rdma): fix the merge codes mistake.

Merge the rdma branch into release V3.4.0.

Signed-off-by: Wu Huocheng <wuhuocheng@oppo.com>
This commit is contained in:
Wu Huocheng 2024-11-08 15:39:14 +08:00
parent 471fc5336a
commit df9c764322
28 changed files with 588 additions and 646 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
build/build.sh
View File

@ -13,7 +13,7 @@ cgo_ldflags="-L${BuildDependsLibPath} -lrocksdb -lz -lbz2 -lsnappy -llz4 -lzstd
if [ "${use_clang}" != "" ]; then
cgo_ldflags="-L${BuildDependsLibPath} -lrocksdb -lz -lbz2 -lsnappy -llz4 -lzstd -lc++"
fi
cgo_cflags="-I${BuildDependsIncludePath}"
cgo_cflags="-I${BuildDependsIncludePath} -std=gnu99"
MODFLAGS=""
gomod=${2:-"on"}
@ -63,6 +63,8 @@ if [ -e /sys/fs/cgroup/cpu ] ; then
fi
NPROC=${NPROC:-"1"}
GCC_LIBRARY_PATH="/lib /lib64 /usr/lib /usr/lib64 /usr/local/lib /usr/local/lib64"
case $(uname -s | tr 'A-Z' 'a-z') in
"linux"|"darwin")
;;
@ -310,16 +312,17 @@ build_blobstore() {
}
build_client() {
pre_build
pushd $SrcPath >/dev/null
echo -n "build cfs-client "
CGO_ENABLED=0 go build ${MODFLAGS} -gcflags=all=-trimpath=${SrcPath} -asmflags=all=-trimpath=${SrcPath} -ldflags="${LDFlags}" -o ${BuildBinPath}/cfs-client ${SrcPath}/client/*.go && echo "success" || echo "failed"
CGO_ENABLED=1 go build ${MODFLAGS} -gcflags=all=-trimpath=${SrcPath} -asmflags=all=-trimpath=${SrcPath} -ldflags="${LDFlags}" -o ${BuildBinPath}/cfs-client ${SrcPath}/client/*.go && echo "success" || echo "failed"
popd >/dev/null
}
build_authtool() {
pushd $SrcPath >/dev/null
echo -n "build cfs-authtool "
CGO_ENABLED=0 go build ${MODFLAGS} -gcflags=all=-trimpath=${SrcPath} -asmflags=all=-trimpath=${SrcPath} -ldflags="${LDFlags}" -o ${BuildBinPath}/cfs-authtool ${SrcPath}/authnode/authtool/*.go && echo "success" || echo "failed"
CGO_ENABLED=1 go build ${MODFLAGS} -gcflags=all=-trimpath=${SrcPath} -asmflags=all=-trimpath=${SrcPath} -ldflags="${LDFlags}" -o ${BuildBinPath}/cfs-authtool ${SrcPath}/authnode/authtool/*.go && echo "success" || echo "failed"
popd >/dev/null
}
@ -345,7 +348,7 @@ build_cfs_deploy() {
build_fsck() {
pushd $SrcPath >/dev/null
echo -n "build cfs-fsck "
CGO_ENABLED=0 go build ${MODFLAGS} -gcflags=all=-trimpath=${SrcPath} -asmflags=all=-trimpath=${SrcPath} -ldflags="${LDFlags}" -o ${BuildBinPath}/cfs-fsck ${SrcPath}/tool/fsck/*.go && echo "success" || echo "failed"
CGO_ENABLED=1 go build ${MODFLAGS} -gcflags=all=-trimpath=${SrcPath} -asmflags=all=-trimpath=${SrcPath} -ldflags="${LDFlags}" -o ${BuildBinPath}/cfs-fsck ${SrcPath}/tool/fsck/*.go && echo "success" || echo "failed"
popd >/dev/null
}
@ -399,20 +402,20 @@ build_libsdk() {
build_fdstore() {
pushd $SrcPath >/dev/null
echo -n "build fdstore "
CGO_ENABLED=0 go build ${MODFLAGS} -gcflags=all=-trimpath=${SrcPath} -asmflags=all=-trimpath=${SrcPath} -ldflags="${LDFlags}" -o ${BuildBinPath}/fdstore ${SrcPath}/client/fdstore/*.go && echo "success" || echo "failed"
CGO_ENABLED=1 go build ${MODFLAGS} -gcflags=all=-trimpath=${SrcPath} -asmflags=all=-trimpath=${SrcPath} -ldflags="${LDFlags}" -o ${BuildBinPath}/fdstore ${SrcPath}/client/fdstore/*.go && echo "success" || echo "failed"
popd >/dev/null
}
build_preload() {
pushd $SrcPath >/dev/null
echo -n "build cfs-preload "
CGO_ENABLED=0 go build ${MODFLAGS} -gcflags=all=-trimpath=${SrcPath} -asmflags=all=-trimpath=${SrcPath} -ldflags="${LDFlags}" -o ${BuildBinPath}/cfs-preload ${SrcPath}/tool/preload/*.go && echo "success" || echo "failed"
CGO_ENABLED=1 go build ${MODFLAGS} -gcflags=all=-trimpath=${SrcPath} -asmflags=all=-trimpath=${SrcPath} -ldflags="${LDFlags}" -o ${BuildBinPath}/cfs-preload ${SrcPath}/tool/preload/*.go && echo "success" || echo "failed"
}
build_bcache(){
pushd $SrcPath >/dev/null
echo -n "build cfs-blockcache "
CGO_ENABLED=0 go build ${MODFLAGS} -gcflags=all=-trimpath=${SrcPath} -asmflags=all=-trimpath=${SrcPath} -ldflags="${LDFlags}" -o ${BuildBinPath}/cfs-bcache ${SrcPath}/client/blockcache/*.go && echo "success" || echo "failed"
CGO_ENABLED=1 go build ${MODFLAGS} -gcflags=all=-trimpath=${SrcPath} -asmflags=all=-trimpath=${SrcPath} -ldflags="${LDFlags}" -o ${BuildBinPath}/cfs-bcache ${SrcPath}/client/blockcache/*.go && echo "success" || echo "failed"
popd >/dev/null
}

1
client/fuse.go
View File

@ -42,7 +42,6 @@ import (
"syscall"
"time"
"github.com/cubefs/cubefs/blockcache/bcache"
"github.com/cubefs/cubefs/client/blockcache/bcache"
cfs "github.com/cubefs/cubefs/client/fs"
"github.com/cubefs/cubefs/depends/bazil.org/fuse"

67
datanode/data_partition_repair.go
View File

@ -25,14 +25,14 @@ import (
"sync"
"time"
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/datanode/repl"
"github.com/cubefs/cubefs/datanode/storage"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/rdma"
)
// DataPartitionRepairTask defines the repair task for the data partition.
@ -583,28 +583,39 @@ func (dp *DataPartition) NormalExtentRepairRead(p repl.PacketInterface, connect
if needReplySize <= 0 {
break
}
var rdmaBuffer *rdma.RdmaBuffer
err = nil
reply := makeRspPacket(p.GetReqID(), p.GetPartitionID(), p.GetExtentID())
reply.SetStartT(p.GetStartT())
currReadSize := uint32(util.Min(int(needReplySize), int(dp.GetRepairBlockSize())))
if currReadSize == util.RepairReadBlockSize {
var data []byte
data, err = proto.Buffers.Get(util.RepairReadBlockSize)
if err != nil {
log.LogErrorf("[NormalExtentRepairRead] dp(%v) failed to get repair data, err(%v)", dp.partitionID, err)
if conn, ok := connect.(*rdma.Connection); ok {
rdmaBuffer, err = conn.GetConnTxDataBuffer(util.PacketHeaderSize + currReadSize)
if err == nil {
reply.SetData(rdmaBuffer.Data[util.PacketHeaderSize:])
reply.SetRdmaBuffer(rdmaBuffer)
} else {
return
}
reply.SetData(data)
} else if currReadSize == util.BlockSize {
var data []byte
data, err = proto.Buffers.Get(util.BlockSize)
if err != nil {
log.LogErrorf("[NormalExtentRepairRead] dp(%v) failed to get repair data, err(%v)", dp.partitionID, err)
return
}
reply.SetData(data)
} else {
reply.SetData(make([]byte, currReadSize))
if currReadSize == util.RepairReadBlockSize {
var data []byte
data, err = proto.Buffers.Get(util.RepairReadBlockSize)
if err != nil {
log.LogErrorf("[NormalExtentRepairRead] dp(%v) failed to get repair data, err(%v)", dp.partitionID, err)
return
}
reply.SetData(data)
} else if currReadSize == util.BlockSize {
var data []byte
data, err = proto.Buffers.Get(util.BlockSize)
if err != nil {
log.LogErrorf("[NormalExtentRepairRead] dp(%v) failed to get repair data, err(%v)", dp.partitionID, err)
return
}
reply.SetData(data)
} else {
reply.SetData(make([]byte, currReadSize))
}
}
if !shallDegrade {
partitionIOMetric = exporter.NewTPCnt(MetricPartitionIOName)
@ -637,13 +648,27 @@ func (dp *DataPartition) NormalExtentRepairRead(p repl.PacketInterface, connect
reply.SetResultCode(proto.OpOk)
reply.SetOpCode(p.GetOpcode())
p.SetResultCode(proto.OpOk)
if err = reply.WriteToConn(connect); err != nil {
return
if conn, ok := connect.(*rdma.Connection); ok {
if err = reply.WriteTxBufferToRdmaConn(conn, reply.GetRdmaBuffer()); err != nil {
conn.ReleaseConnTxDataBuffer(reply.GetRdmaBuffer())
return
}
} else {
if err = reply.WriteToConn(connect); err != nil {
if currReadSize == util.ReadBlockSize {
proto.Buffers.Put(reply.GetData())
}
return
}
}
needReplySize -= currReadSize
offset += int64(currReadSize)
if currReadSize == util.ReadBlockSize {
proto.Buffers.Put(reply.GetData())
if conn, ok := connect.(*rdma.Connection); ok {
conn.ReleaseConnTxDataBuffer(reply.GetRdmaBuffer())
} else {
if currReadSize == util.ReadBlockSize {
proto.Buffers.Put(reply.GetData())
}
}
if connect.RemoteAddr() != nil {
logContent := fmt.Sprintf("action[operatePacket] %v.",

2
datanode/partition_raft.go
View File

@ -28,7 +28,6 @@ import (
"time"
"github.com/cubefs/cubefs/datanode/repl"
"github.com/cubefs/cubefs/depends/tiglabs/raft"
raftproto "github.com/cubefs/cubefs/depends/tiglabs/raft/proto"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/raftstore"
@ -564,7 +563,6 @@ func (s *DataNode) parseRaftConfig(cfg *config.Config) (err error) {
log.LogDebugf("[parseRaftConfig] load raftDir(%v).", s.raftDir)
log.LogDebugf("[parseRaftConfig] load raftHearbeat(%v).", s.raftHeartbeat)
log.LogDebugf("[parseRaftConfig] load raftReplica(%v).", s.raftReplica)
log.LogDebugf("[parseRaftConfig] load enableRaftRdma(%v).", raft.IsRdma)
return
}

121
datanode/repl/packet.go
View File

@ -15,6 +15,7 @@
package repl
import (
"encoding/binary"
"fmt"
"io"
"net"
@ -28,6 +29,7 @@ import (
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/rdma"
)
var (
@ -66,6 +68,9 @@ type PacketInterface interface {
PacketOkReply()
SetArglen(len uint32)
SetArg(data []byte)
SetRdmaBuffer(buff *rdma.RdmaBuffer)
GetRdmaBuffer() *rdma.RdmaBuffer
WriteTxBufferToRdmaConn(conn *rdma.Connection, rdmaBuffer *rdma.RdmaBuffer) (err error)
}
type (
@ -150,7 +155,7 @@ func (p *Packet) AfterTp() (ok bool) {
return
}
func (p *Packet) clean() {
func (p *Packet) clean(c ...net.Conn) {
if p.Data == nil && p.OrgBuffer == nil {
return
}
@ -158,9 +163,19 @@ func (p *Packet) clean() {
p.TpObject = nil
p.Data = nil
p.Arg = nil
if p.OrgBuffer != nil && len(p.OrgBuffer) == util.BlockSize && p.IsNormalWriteOperation() {
proto.Buffers.Put(p.OrgBuffer)
p.OrgBuffer = nil
if p.IsRdma {
if p.RdmaBuffer != nil {
if len(c) != 1 {
return
}
conn := c[0].(*rdma.Connection)
conn.ReleaseConnRxDataBuffer(p.RdmaBuffer)
}
} else {
if p.OrgBuffer != nil && len(p.OrgBuffer) == util.BlockSize && p.IsNormalWriteOperation() {
proto.Buffers.Put(p.OrgBuffer)
p.OrgBuffer = nil
}
}
}
@ -177,6 +192,7 @@ func copyPacket(src *Packet, dst *FollowerPacket) {
dst.ExtentOffset = src.ExtentOffset
dst.ReqID = src.ReqID
dst.Data = src.OrgBuffer
dst.RdmaBuffer = src.RdmaBuffer
}
func (p *Packet) BeforeTp(clusterID string) (ok bool) {
@ -492,6 +508,86 @@ func (p *Packet) ReadFull(c net.Conn, opcode uint8, readSize int) (err error) {
return
}
func (p *Packet) ReadFromConnFromCli(c net.Conn, deadlineTime int) (err error) {
if conn, ok := c.(*rdma.Connection); ok {
return p.ReadFromRdmaConnFromCliWithVer(conn, deadlineTime)
} else {
return p.ReadFromConnWithVer(c, deadlineTime)
}
}
func (p *Packet) ReadFromRdmaConnFromCliWithVer(conn *rdma.Connection, deadlineTime int) (err error) {
if deadlineTime != proto.NoReadDeadlineTime {
conn.SetReadDeadline(time.Now().Add(time.Duration(deadlineTime) * time.Second))
} else {
conn.SetReadDeadline(time.Time{})
}
var dataBuffer []byte
var rdmaBuffer *rdma.RdmaBuffer
var offset uint32
if rdmaBuffer, err = conn.GetRecvMsgBuffer(); err != nil {
return
}
dataBuffer = rdmaBuffer.Data
if err = p.UnmarshalHeader(dataBuffer[0:util.PacketHeaderSize]); err != nil {
conn.ReleaseConnRxDataBuffer(rdmaBuffer)
return
}
offset = util.PacketHeaderSize
if p.ExtentType&proto.MultiVersionFlag > 0 {
ver := dataBuffer[offset : offset+8]
p.VerSeq = binary.BigEndian.Uint64(ver)
offset += 8
}
if p.IsVersionList() {
cntByte := dataBuffer[offset : offset+2]
cnt := binary.BigEndian.Uint16(cntByte)
log.LogDebugf("action[ReadFromConnWithVer] op %s verseq %v, extType %d, cnt %d",
p.GetOpMsg(), p.VerSeq, p.ExtentType, cnt)
verData := dataBuffer[offset : offset+uint32(cnt)*proto.VerInfoCnt]
offset += 2 * uint32(cnt) * proto.VerInfoCnt
err = p.UnmarshalVersionSlice(int(cnt), verData)
if err != nil {
log.LogWarnf("ReadFromConnWithVer: unmarshal ver slice failed, err %s", err.Error())
conn.ReleaseConnRxDataBuffer(rdmaBuffer)
return err
}
}
if len(dataBuffer) == util.RdmaPacketHeaderSize+int(p.Size) { //header + args + data
if p.ArgLen > 0 {
//p.Arg = make([]byte, int(p.ArgLen))
//copy(p.Arg, dataBuffer[util.PacketHeaderSize:util.PacketHeaderSize+p.ArgLen])
p.Arg = dataBuffer[util.PacketHeaderSize : util.PacketHeaderSize+p.ArgLen]
}
offset += util.RdmaPacketArgSize
}
log.LogDebugf("read packet(%v) len dataBuffer(%v) addr(%p) offset(%v)", p, len(dataBuffer), &dataBuffer[0], offset)
if p.Size < 0 {
conn.ReleaseConnRxDataBuffer(rdmaBuffer)
return
}
size := p.Size
if p.IsReadOperation() && p.ResultCode == proto.OpInitResultCode {
size = 0
}
if len(dataBuffer) < int(offset+size) {
log.LogErrorf("data is out of range. buffer len(%d), offset(%d), size(%d)", len(dataBuffer), offset, size)
conn.ReleaseConnRxDataBuffer(rdmaBuffer)
return errors.New("data size is out of buffer")
}
p.Data = dataBuffer[offset : offset+size]
p.RdmaBuffer = rdmaBuffer
p.IsRdma = true
return
}
func (p *Packet) IsMasterCommand() bool {
switch p.Opcode {
case
@ -625,3 +721,20 @@ func (p *Packet) SetArglen(len uint32) {
func (p *Packet) SetArg(data []byte) {
p.Arg = data
}
func (p *Packet) SetRdmaBuffer(buff *rdma.RdmaBuffer) {
p.RdmaBuffer = buff
}
func (p *Packet) GetRdmaBuffer() *rdma.RdmaBuffer {
return p.RdmaBuffer
}
func (p *Packet) WriteTxBufferToRdmaConn(conn *rdma.Connection, rdmaBuffer *rdma.RdmaBuffer) (err error) {
p.MarshalHeader(rdmaBuffer.Data[0:util.PacketHeaderSize])
if _, err = conn.WriteBuffer(rdmaBuffer); err != nil {
return
}
return
}

125
datanode/repl/repl_protocol.go
View File

@ -26,9 +26,13 @@ import (
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/rdma"
)
var gConnPool = util.NewConnectPool()
var (
gConnPool = util.NewConnectPool()
RdmaConnPool *util.RdmaConnectPool
)
// ReplProtocol defines the struct of the replication protocol.
// 1. ServerConn reads a packet from the client socket, and analyzes the addresses of the followers.
@ -90,18 +94,36 @@ func (ft *FollowerTransport) serverWriteToFollower() {
for {
select {
case p := <-ft.sendCh:
if err := p.WriteToConn(ft.conn); err != nil {
p.PackErrorBody(ActionSendToFollowers, err.Error())
p.respCh <- fmt.Errorf(string(p.Data[:p.Size]))
log.LogErrorf("serverWriteToFollower ft.addr(%v), req(%s), err (%v)", ft.addr, p.String(), err.Error())
ft.conn.Close()
continue
log.LogDebugf("writeToFollowerRdmaConn start: followerPacket:%v", p)
if conn, ok := ft.conn.(*rdma.Connection); ok {
if err := p.WriteExternalToRdmaConn(conn, p.RdmaBuffer, int(util.RdmaPacketHeaderSize+p.Size)); err != nil {
p.PackErrorBody(ActionSendToFollowers, err.Error())
p.respCh <- fmt.Errorf(string(p.Data[:p.Size]))
log.LogErrorf("serverWriteToFollower ft.addr(%v) packet(%v), err (%v)", ft.addr, p, err.Error())
RdmaConnPool.PutRdmaConn(conn, true)
continue
}
} else {
if err := p.WriteToConn(ft.conn); err != nil {
p.PackErrorBody(ActionSendToFollowers, err.Error())
p.respCh <- fmt.Errorf(string(p.Data[:p.Size]))
log.LogErrorf("serverWriteToFollower ft.addr(%v), err (%v)", ft.addr, err.Error())
ft.conn.Close()
continue
}
}
log.LogDebugf("writeToFollowerRdmaConn end: followerPacket:%v", p)
ft.recvCh <- p
case <-ft.exitCh:
ft.exitedMu.Lock()
if atomic.AddInt32(&ft.isclosed, -1) == FollowerTransportExited {
ft.conn.Close()
if conn, ok := ft.conn.(*rdma.Connection); ok {
RdmaConnPool.PutRdmaConn(conn, true)
} else {
ft.conn.Close()
}
atomic.StoreInt32(&ft.isclosed, FollowerTransportExited)
}
ft.exitedMu.Unlock()
@ -118,7 +140,11 @@ func (ft *FollowerTransport) serverReadFromFollower() {
case <-ft.exitCh:
ft.exitedMu.Lock()
if atomic.AddInt32(&ft.isclosed, -1) == FollowerTransportExited {
ft.conn.Close()
if conn, ok := ft.conn.(*rdma.Connection); ok {
RdmaConnPool.PutRdmaConn(conn, true)
} else {
ft.conn.Close()
}
atomic.StoreInt32(&ft.isclosed, FollowerTransportExited)
}
ft.exitedMu.Unlock()
@ -131,10 +157,20 @@ func (ft *FollowerTransport) serverReadFromFollower() {
func (ft *FollowerTransport) readFollowerResult(request *FollowerPacket) (err error) {
reply := NewPacket()
defer func() {
reply.clean()
request.respCh <- err
if err != nil {
ft.conn.Close()
if conn, ok := ft.conn.(*rdma.Connection); ok {
reply.clean()
conn.ReleaseConnExternalDataBuffer(request.RdmaBuffer)
request.respCh <- err
if err != nil {
log.LogErrorf("serverWriteToFollower ft.addr(%v), err (%v)", ft.addr, err.Error())
RdmaConnPool.PutRdmaConn(conn, true)
}
} else {
reply.clean()
request.respCh <- err
if err != nil {
ft.conn.Close()
}
}
}()
@ -146,11 +182,19 @@ func (ft *FollowerTransport) readFollowerResult(request *FollowerPacket) (err er
if request.IsBatchDeleteExtents() || request.IsBatchLockNormalExtents() || request.IsBatchUnlockNormalExtents() {
timeOut = proto.BatchDeleteExtentReadDeadLineTime
}
if err = reply.ReadFromConnWithVer(ft.conn, timeOut); err != nil {
log.LogErrorf("readFollowerResult ft.addr(%v), err(%v)", ft.addr, err.Error())
return
log.LogDebugf("RecvRespFromRDMAConn start: reply:%v", reply)
if conn, ok := ft.conn.(*rdma.Connection); ok {
if reply.ReadFromRdmaConnWithVer(conn, timeOut); err != nil {
log.LogErrorf("readFollowerResult ft.addr(%v), err(%v)", ft.addr, err.Error())
return
}
} else {
if err = reply.ReadFromConnWithVer(ft.conn, timeOut); err != nil {
log.LogErrorf("readFollowerResult ft.addr(%v), err(%v)", ft.addr, err.Error())
return
}
}
log.LogDebugf("RecvRespFromRDMAConn end: reply:%v", reply)
if reply.GetReqID() != request.ReqID || reply.GetPartitionID() != request.PartitionID ||
reply.ExtentOffset != request.ExtentOffset || reply.CRC != request.CRC || reply.ExtentID != request.ExtentID {
err = fmt.Errorf(ActionCheckReply+" request(%v), reply(%v) ", request.GetUniqueLogId(),
@ -205,8 +249,7 @@ func (ft *FollowerTransport) Write(p *FollowerPacket) {
}
func NewReplProtocol(inConn net.Conn, prepareFunc func(p *Packet) error,
operatorFunc func(p *Packet, c net.Conn) error, postFunc func(p *Packet) error,
) *ReplProtocol {
operatorFunc func(p *Packet, c net.Conn) error, postFunc func(p *Packet) error) *ReplProtocol {
rp := new(ReplProtocol)
rp.packetList = list.New()
rp.ackCh = make(chan struct{}, RequestChanSize)
@ -285,7 +328,7 @@ func (rp *ReplProtocol) hasError() bool {
func (rp *ReplProtocol) readPkgAndPrepare() (err error) {
request := NewPacket()
if err = request.ReadFromConnWithVer(rp.sourceConn, proto.NoReadDeadlineTime); err != nil {
if err = request.ReadFromConnFromCli(rp.sourceConn, proto.NoReadDeadlineTime); err != nil {
return
}
// log.LogDebugf("action[readPkgAndPrepare] packet(%v) op %v from remote(%v) conn(%v) ",
@ -301,7 +344,13 @@ func (rp *ReplProtocol) readPkgAndPrepare() (err error) {
}
err = rp.putToBeProcess(request)
if err != nil {
if request.IsRdma {
conn := rp.sourceConn.(*rdma.Connection)
conn.ReleaseConnRxDataBuffer(request.RdmaBuffer)
}
}
log.LogDebugf("read pkg(%v) and prepare exit", request)
return
}
@ -418,7 +467,7 @@ func (rp *ReplProtocol) checkLocalResultAndReciveAllFollowerResponse() {
func (rp *ReplProtocol) writeResponse(reply *Packet) {
var err error
defer func() {
reply.clean()
reply.clean(rp.sourceConn)
}()
log.LogDebugf("writeResponse.opcode %v reply %v conn(%v)", reply.Opcode, reply.GetUniqueLogId(), rp.sourceConn.RemoteAddr().String())
if reply.IsErrPacket() {
@ -428,6 +477,9 @@ func (rp *ReplProtocol) writeResponse(reply *Packet) {
log.LogInfof(err.Error())
} else {
log.LogErrorf(err.Error())
if conn, ok := rp.sourceConn.(*rdma.Connection); ok {
conn.CloseWaitFlag = true
}
}
rp.Stop()
}
@ -441,11 +493,24 @@ func (rp *ReplProtocol) writeResponse(reply *Packet) {
return
}
if err = reply.WriteToConn(rp.sourceConn); err != nil {
err = fmt.Errorf(reply.LogMessage(ActionWriteToClient, fmt.Sprintf("local(%v)->remote(%v)", rp.sourceConn.LocalAddr().String(),
rp.sourceConn.RemoteAddr().String()), reply.StartT, err))
log.LogErrorf(err.Error())
rp.Stop()
if conn, ok := rp.sourceConn.(*rdma.Connection); ok {
log.LogDebugf("send resp to rdma conn: packet(%v)", reply)
if err = reply.WriteToRdmaConn(conn); err != nil {
err = fmt.Errorf(reply.LogMessage(ActionWriteToClient, fmt.Sprintf("local(%v)->remote(%v)", rp.sourceConn.LocalAddr().String(),
rp.sourceConn.RemoteAddr().String()), reply.StartT, err))
log.LogErrorf(err.Error())
rp.Stop()
}
log.LogDebugf("send resp to rdma conn: time[%v]", time.Now())
} else {
log.LogDebugf("send resp to tcp conn: packet(%v)", reply)
if err = reply.WriteToConn(rp.sourceConn); err != nil {
err = fmt.Errorf(reply.LogMessage(ActionWriteToClient, fmt.Sprintf("local(%v)->remote(%v)", rp.sourceConn.LocalAddr().String(),
rp.sourceConn.RemoteAddr().String()), reply.StartT, err))
log.LogErrorf(err.Error())
rp.Stop()
}
log.LogDebugf("send resp to tcp conn: time[%v]", time.Now())
}
log.LogDebugf(reply.LogMessage(ActionWriteToClient,
rp.sourceConn.RemoteAddr().String(), reply.StartT, err))
@ -499,7 +564,11 @@ func (rp *ReplProtocol) allocateFollowersConns(p *Packet, index int) (transport
}
} else {
conn, err = gConnPool.GetConnect(addr)
if _, ok := rp.sourceConn.(*rdma.Connection); ok {
conn, err = RdmaConnPool.GetRdmaConn(addr)
} else {
conn, err = gConnPool.GetConnect(addr)
}
if err != nil {
return
}

4
datanode/server.go
View File

@ -34,7 +34,6 @@ import (
"github.com/cubefs/cubefs/cmd/common"
"github.com/cubefs/cubefs/datanode/repl"
"github.com/cubefs/cubefs/depends/tiglabs/raft"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/raftstore"
masterSDK "github.com/cubefs/cubefs/sdk/master"
@ -372,8 +371,7 @@ func (s *DataNode) parseConfig(cfg *config.Config) (err error) {
port = cfg.GetString(proto.ListenPort)
isRdma = cfg.GetBoolWithDefault("enableRdma", false)
raft.IsRdma = cfg.GetBoolWithDefault("enableRaftRdma", false)
if isRdma || raft.IsRdma {
if isRdma {
LocalRdmaIP = cfg.GetString("rdmaIP")
rdmaServerPort = cfg.GetString("rdmaPort")
util.Config.MemBlockNum = int(cfg.GetInt64WithDefault("rdmaMemBlockNum", 4*8*1024))

1
datanode/wrap_operator.go
View File

@ -39,6 +39,7 @@ import (
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/rdma"
"github.com/cubefs/cubefs/util/strutil"
)

1
datanode/wrap_prepare.go
View File

@ -17,7 +17,6 @@ package datanode
import (
"encoding/json"
"fmt"
"github.com/cubefs/cubefs/util/log"
"hash/crc32"
"sync/atomic"

141
depends/tiglabs/raft/proto/codec.go
View File

@ -16,7 +16,6 @@ package proto
import (
"encoding/binary"
"fmt"
"github.com/cubefs/cubefs/util/rdma"
"io"
"sort"
@ -105,17 +104,6 @@ func (m *SnapshotMeta) Encode(w io.Writer) error {
return nil
}
func (m *SnapshotMeta) EncodeBuffer(buf []byte) {
binary.BigEndian.PutUint64(buf, m.Index)
binary.BigEndian.PutUint64(buf[8:], m.Term)
binary.BigEndian.PutUint32(buf[16:], uint32(len(m.Peers)))
start := snapmeta_header
for _, p := range m.Peers {
p.Encode(buf[start:])
start += peer_size
}
}
func (m *SnapshotMeta) Decode(datas []byte) {
m.Index = binary.BigEndian.Uint64(datas)
m.Term = binary.BigEndian.Uint64(datas[8:])
@ -133,13 +121,6 @@ func (e *Entry) Size() uint64 {
return entry_header + uint64(len(e.Data))
}
func (e *Entry) EncodeBuffer(buf []byte) {
buf[0] = byte(e.Type)
binary.BigEndian.PutUint64(buf[1:], e.Term)
binary.BigEndian.PutUint64(buf[9:], e.Index)
copy(buf[entry_header:], e.Data)
}
func (e *Entry) Encode(w io.Writer) error {
buf := getByteSlice()
defer returnByteSlice(buf)
@ -168,16 +149,6 @@ func (e *Entry) Decode(datas []byte) {
}
}
func (e *Entry) DecodeByRdma(datas []byte) {
e.Type = EntryType(datas[0])
e.Term = binary.BigEndian.Uint64(datas[1:])
e.Index = binary.BigEndian.Uint64(datas[9:])
if uint64(len(datas)) > entry_header {
e.Data = make([]byte, len(datas[entry_header:]))
copy(e.Data, datas[entry_header:])
}
}
// Message codec
func (m *Message) Size() uint64 {
if m.Type == ReqMsgSnapShot {
@ -252,61 +223,6 @@ func (m *Message) Encode(w io.Writer) error {
return nil
}
func (m *Message) EncodeByRdma(c *util.ConnTimeout) (err error) {
var rdmaBuffer *rdma.RdmaBuffer
var buf []byte
buffSize := int(m.Size() + 4)
if rdmaBuffer, err = c.GetDataBuffer(uint32(buffSize)); err != nil {
return err
}
buf = rdmaBuffer.Data
binary.BigEndian.PutUint32(buf, uint32(m.Size()))
buf[4] = version1
buf[5] = byte(m.Type)
if m.ForceVote {
buf[6] = 1
} else {
buf[6] = 0
}
if m.Reject {
buf[7] = 1
} else {
buf[7] = 0
}
binary.BigEndian.PutUint64(buf[8:], m.RejectIndex)
binary.BigEndian.PutUint64(buf[16:], m.ID)
binary.BigEndian.PutUint64(buf[24:], m.From)
binary.BigEndian.PutUint64(buf[32:], m.To)
binary.BigEndian.PutUint64(buf[40:], m.Term)
binary.BigEndian.PutUint64(buf[48:], m.LogTerm)
binary.BigEndian.PutUint64(buf[56:], m.Index)
binary.BigEndian.PutUint64(buf[64:], m.Commit)
start := int(message_header + 4)
if m.Type == ReqMsgSnapShot {
m.SnapshotMeta.EncodeBuffer(buf[start:])
err = c.AddWriteRequest(rdmaBuffer)
return err
}
binary.BigEndian.PutUint32(buf[start:], uint32(len(m.Entries)))
start += 4
if len(m.Entries) > 0 {
for _, e := range m.Entries {
binary.BigEndian.PutUint32(buf[start:], uint32(e.Size()))
start += 4
e.EncodeBuffer(buf[start:])
start += int(e.Size())
}
}
if len(m.Context) > 0 {
copy(buf[start:], m.Context)
}
err = c.AddWriteRequest(rdmaBuffer)
return err
}
func (m *Message) Decode(r *util.BufferReader) error {
var (
datas []byte
@ -365,63 +281,6 @@ func (m *Message) Decode(r *util.BufferReader) error {
return nil
}
func (m *Message) DecodeByRdma(c *util.ConnTimeout) error {
var (
rdmaBuffer *rdma.RdmaBuffer
datas []byte
err error
)
if rdmaBuffer, err = c.ReadByRdma(); err != nil {
return err
}
datas = rdmaBuffer.Data[:4]
cnt := int(binary.BigEndian.Uint32(datas))
if cnt > 256*1024*1024 {
return fmt.Errorf("msg len is too big, please check, %d", cnt)
}
datas = rdmaBuffer.Data[4:]
ver := datas[0]
if ver == version1 {
m.Type = MsgType(datas[1])
m.ForceVote = (datas[2] == 1)
m.Reject = (datas[3] == 1)
m.RejectIndex = binary.BigEndian.Uint64(datas[4:])
m.ID = binary.BigEndian.Uint64(datas[12:])
m.From = binary.BigEndian.Uint64(datas[20:])
m.To = binary.BigEndian.Uint64(datas[28:])
m.Term = binary.BigEndian.Uint64(datas[36:])
m.LogTerm = binary.BigEndian.Uint64(datas[44:])
m.Index = binary.BigEndian.Uint64(datas[52:])
m.Commit = binary.BigEndian.Uint64(datas[60:])
if m.Type == ReqMsgSnapShot {
m.SnapshotMeta.Decode(datas[message_header:])
} else {
size := binary.BigEndian.Uint32(datas[message_header:])
start := message_header + 4
if size > 0 {
for i := uint32(0); i < size; i++ {
esize := binary.BigEndian.Uint32(datas[start:])
start = start + 4
end := start + uint64(esize)
entry := new(Entry)
entry.DecodeByRdma(datas[start:end])
m.Entries = append(m.Entries, entry)
start = end
}
}
if start < uint64(len(datas)) {
m.Context = make([]byte, len(datas[start:]))
copy(m.Context, datas[start:])
}
}
}
if err = c.ReleaseRxByRdma(rdmaBuffer); err != nil {
return err
}
return nil
}
func EncodeHBConext(ctx HeartbeatContext) (buf []byte) {
sort.Slice(ctx, func(i, j int) bool {
return ctx[i] < ctx[j]

64
depends/tiglabs/raft/transport_heartbeat.go
View File

@ -16,10 +16,7 @@ package raft
import (
"fmt"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/rdma"
"net"
"strings"
"sync"
//"fmt"
@ -43,25 +40,10 @@ func newHeartbeatTransport(raftServer *RaftServer, config *TransportConfig) (*he
listener net.Listener
err error
)
if IsRdma {
heartbeatAddrSplits := strings.Split(config.HeartbeatAddr, ":")
if len(heartbeatAddrSplits) != 2 {
err = errors.New("illegal heartbeat address")
return nil, err
}
ip, port := heartbeatAddrSplits[0], heartbeatAddrSplits[1]
if ip == "" {
ip = "127.0.0.1"
}
if listener, err = rdma.NewRdmaServer(ip, port); err != nil {
return nil, err
}
} else {
if listener, err = net.Listen("tcp", config.HeartbeatAddr); err != nil {
return nil, err
}
}
if listener, err = net.Listen("tcp", config.HeartbeatAddr); err != nil {
return nil, err
}
t := &heartbeatTransport{
config: config,
raftServer: raftServer,
@ -109,39 +91,21 @@ func (t *heartbeatTransport) handleConn(conn *util.ConnTimeout) {
util.RunWorker(func() {
defer conn.Close()
if conn.IsRdma() {
for {
select {
case <-t.stopc:
bufRd := util.NewBufferReader(conn, 16*KB)
for {
select {
case <-t.stopc:
return
default:
if msg, err := reciveMessage(bufRd); err != nil {
logger.Error(fmt.Sprintf("[heartbeatTransport] recive message from conn error, %s", err.Error()))
return
default:
if msg, err := reciveMessageByRdma(conn); err != nil {
logger.Error(fmt.Sprintf("[heartbeatTransport] recive message from rdma conn error, %s", err.Error()))
return
} else {
//logger.Debug(fmt.Sprintf("Recive %v from (%v)", msg.ToString(), conn.RemoteAddr()))
t.raftServer.reciveMessage(msg)
}
}
}
} else {
bufRd := util.NewBufferReader(conn, 16*KB)
for {
select {
case <-t.stopc:
return
default:
if msg, err := reciveMessage(bufRd); err != nil {
logger.Error(fmt.Sprintf("[heartbeatTransport] recive message from conn error, %s", err.Error()))
return
} else {
//logger.Debug(fmt.Sprintf("Recive %v from (%v)", msg.ToString(), conn.RemoteAddr()))
t.raftServer.reciveMessage(msg)
}
} else {
//logger.Debug(fmt.Sprintf("Recive %v from (%v)", msg.ToString(), conn.RemoteAddr()))
t.raftServer.reciveMessage(msg)
}
}
}
})
}

23
depends/tiglabs/raft/transport_multi.go
View File

@ -16,12 +16,7 @@ package raft
import (
"github.com/cubefs/cubefs/depends/tiglabs/raft/proto"
raftUtil "github.com/cubefs/cubefs/depends/tiglabs/raft/util"
"github.com/cubefs/cubefs/util"
)
var (
IsRdma bool
"github.com/cubefs/cubefs/depends/tiglabs/raft/util"
)
type MultiTransport struct {
@ -32,12 +27,6 @@ type MultiTransport struct {
func NewMultiTransport(raft *RaftServer, config *TransportConfig) (Transport, error) {
mt := new(MultiTransport)
if IsRdma {
if err := util.InitRdmaEnv(); err != nil {
return nil, err
}
}
if ht, err := newHeartbeatTransport(raft, config); err != nil {
return nil, err
} else {
@ -72,18 +61,10 @@ func (t *MultiTransport) SendSnapshot(m *proto.Message, rs *snapshotStatus) {
t.replicate.sendSnapshot(m, rs)
}
func reciveMessage(r *raftUtil.BufferReader) (msg *proto.Message, err error) {
func reciveMessage(r *util.BufferReader) (msg *proto.Message, err error) {
msg = proto.GetMessage()
if err = msg.Decode(r); err != nil {
proto.ReturnMessage(msg)
}
return
}
func reciveMessageByRdma(conn *raftUtil.ConnTimeout) (msg *proto.Message, err error) {
msg = proto.GetMessage()
if err = msg.DecodeByRdma(conn); err != nil {
proto.ReturnMessage(msg)
}
return
}

85
depends/tiglabs/raft/transport_replicate.go
View File

@ -17,12 +17,9 @@ package raft
import (
"encoding/binary"
"fmt"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/rdma"
"io"
"net"
"runtime"
"strings"
"sync"
"sync/atomic"
"time"
@ -48,23 +45,8 @@ func newReplicateTransport(raftServer *RaftServer, config *TransportConfig) (*re
err error
)
if IsRdma {
replicateRdmaAddrSplits := strings.Split(config.ReplicateAddr, ":")
if len(replicateRdmaAddrSplits) != 2 {
err = errors.New("illegal replica rdma address")
return nil, err
}
ip, port := replicateRdmaAddrSplits[0], replicateRdmaAddrSplits[1]
if ip == "" {
ip = "127.0.0.1"
}
if listener, err = rdma.NewRdmaServer(ip, port); err != nil {
return nil, err
}
} else {
if listener, err = net.Listen("tcp", config.ReplicateAddr); err != nil {
return nil, err
}
if listener, err = net.Listen("tcp", config.ReplicateAddr); err != nil {
return nil, err
}
t := &replicateTransport{
config: config,
@ -223,55 +205,28 @@ func (t *replicateTransport) handleConn(conn *util.ConnTimeout) {
defer conn.Close()
loopCount := 0
if conn.IsRdma() {
for {
loopCount = loopCount + 1
if loopCount > 16 {
loopCount = 0
runtime.Gosched()
}
select {
case <-t.stopc:
return
default:
if msg, err := reciveMessageByRdma(conn); err != nil {
return
} else {
//logger.Debug(fmt.Sprintf("Recive %v from (%v)", msg.ToString(), conn.RemoteAddr()))
if msg.Type == proto.ReqMsgSnapShot {
err = errors.NewErrorf("rdma mode does not support processing snapshot")
return
} else {
t.raftServer.reciveMessage(msg)
}
}
}
bufRd := util.NewBufferReader(conn, 16*KB)
for {
loopCount = loopCount + 1
if loopCount > 16 {
loopCount = 0
runtime.Gosched()
}
} else {
bufRd := util.NewBufferReader(conn, 16*KB)
for {
loopCount = loopCount + 1
if loopCount > 16 {
loopCount = 0
runtime.Gosched()
}
select {
case <-t.stopc:
select {
case <-t.stopc:
return
default:
if msg, err := reciveMessage(bufRd); err != nil {
return
default:
if msg, err := reciveMessage(bufRd); err != nil {
return
} else {
//logger.Debug(fmt.Sprintf("Recive %v from (%v)", msg.ToString(), conn.RemoteAddr()))
if msg.Type == proto.ReqMsgSnapShot {
if err := t.handleSnapshot(msg, conn, bufRd); err != nil {
return
}
} else {
t.raftServer.reciveMessage(msg)
} else {
//logger.Debug(fmt.Sprintf("Recive %v from (%v)", msg.ToString(), conn.RemoteAddr()))
if msg.Type == proto.ReqMsgSnapShot {
if err := t.handleSnapshot(msg, conn, bufRd); err != nil {
return
}
} else {
t.raftServer.reciveMessage(msg)
}
}
}

66
depends/tiglabs/raft/transport_sender.go
View File

@ -86,15 +86,8 @@ func (s *transportSender) stop() {
func (s *transportSender) loopSend(recvc chan *proto.Message) {
util.RunWorkerUtilStop(func() {
var conn *util.ConnTimeout
var bufWr *util.BufferWriter
if IsRdma {
conn = getRdmaConn(s.nodeID, s.senderType, s.resolver, 0, 5*time.Second)
} else {
conn = getConn(s.nodeID, s.senderType, s.resolver, 0, 2*time.Second)
bufWr = util.NewBufferWriter(conn, 16*KB)
}
conn := getConn(s.nodeID, s.senderType, s.resolver, 0, 2*time.Second)
bufWr := util.NewBufferWriter(conn, 16*KB)
defer func() {
if conn != nil {
@ -117,11 +110,7 @@ func (s *transportSender) loopSend(recvc chan *proto.Message) {
case msg := <-recvc:
if conn == nil {
if IsRdma {
conn = getRdmaConn(s.nodeID, s.senderType, s.resolver, 0, 0*time.Second)
} else {
conn = getConn(s.nodeID, s.senderType, s.resolver, 0, 2*time.Second)
}
conn = getConn(s.nodeID, s.senderType, s.resolver, 0, 2*time.Second)
if conn == nil {
proto.ReturnMessage(msg)
// reset chan
@ -137,15 +126,9 @@ func (s *transportSender) loopSend(recvc chan *proto.Message) {
time.Sleep(50 * time.Millisecond)
continue
}
if !IsRdma {
bufWr.Reset(conn)
}
}
if IsRdma {
err = msg.EncodeByRdma(conn)
} else {
err = msg.Encode(bufWr)
bufWr.Reset(conn)
}
err = msg.Encode(bufWr)
proto.ReturnMessage(msg)
if err != nil {
goto flush
@ -155,11 +138,7 @@ func (s *transportSender) loopSend(recvc chan *proto.Message) {
for i := 0; i < 16; i++ {
select {
case msg := <-recvc:
if IsRdma {
err = msg.EncodeByRdma(conn)
} else {
err = msg.Encode(bufWr)
}
err = msg.Encode(bufWr)
//logger.Debug(fmt.Sprintf("SendMesg %v to (%v) ", msg.ToString(), conn.RemoteAddr()))
proto.ReturnMessage(msg)
if err != nil {
@ -177,11 +156,7 @@ func (s *transportSender) loopSend(recvc chan *proto.Message) {
flush:
// flush write
if err == nil {
if IsRdma {
err = conn.Flush()
} else {
err = bufWr.Flush()
}
err = bufWr.Flush()
}
if err != nil {
logger.Error("[Transport]send message[%s] to %v[%s] error:[%v].", s.senderType, s.nodeID, conn.RemoteAddr(), err)
@ -212,30 +187,3 @@ func getConn(nodeID uint64, socketType SocketType, resolver SocketResolver, rdTi
}
return
}
func getRdmaConn(nodeID uint64, socketType SocketType, resolver SocketResolver, rdTime, wrTime time.Duration) (conn *util.ConnTimeout) {
var (
addr string
err error
)
if addr, err = resolver.NodeAddress(nodeID, socketType); err == nil {
if conn, err = util.DialRdmaTimeout(addr, 2*time.Second); err == nil {
if rdTime.Nanoseconds() > 0 {
err = conn.GetRdmaConn().SetReadDeadline(time.Now().Add(rdTime))
}
if err == nil {
if wrTime.Nanoseconds() > 0 {
err = conn.GetRdmaConn().SetWriteDeadline(time.Now().Add(wrTime))
}
}
}
}
if err != nil {
conn = nil
if logger.IsEnableDebug() {
logger.Debug("[Transport] get rdma connection[%s] to %v[%s] failed,error is: %s", socketType, nodeID, addr, err)
}
}
return
}

84
depends/tiglabs/raft/util/conn.go
View File

@ -15,10 +15,7 @@
package util
import (
"errors"
"github.com/cubefs/cubefs/util/rdma"
"net"
"strings"
"time"
)
@ -27,7 +24,6 @@ type ConnTimeout struct {
conn net.Conn
readTime time.Duration
writeTime time.Duration
isRdma bool
}
func DialTimeout(addr string, connTime time.Duration) (*ConnTimeout, error) {
@ -39,22 +35,7 @@ func DialTimeout(addr string, connTime time.Duration) (*ConnTimeout, error) {
conn.(*net.TCPConn).SetNoDelay(true)
conn.(*net.TCPConn).SetLinger(0)
conn.(*net.TCPConn).SetKeepAlive(true)
return &ConnTimeout{conn: conn, addr: addr, isRdma: false}, nil
}
func DialRdmaTimeout(addr string, connTime time.Duration) (*ConnTimeout, error) {
str := strings.Split(addr, ":")
targetIp := str[0]
targetPort := str[1]
conn := &rdma.Connection{}
conn.TargetIp = targetIp
conn.TargetPort = targetPort
err := conn.DialTimeout(targetIp, targetPort, false, connTime)
if err != nil {
return nil, err
}
conn.SetLoopExchange()
return &ConnTimeout{conn: conn, addr: addr, isRdma: true}, nil
return &ConnTimeout{conn: conn, addr: addr}, nil
}
func NewConnTimeout(conn net.Conn) *ConnTimeout {
@ -62,16 +43,10 @@ func NewConnTimeout(conn net.Conn) *ConnTimeout {
return nil
}
if c, ok := conn.(*rdma.Connection); ok {
c.SetLoopExchange()
return &ConnTimeout{conn: c, addr: c.RemoteAddr().String(), isRdma: true}
} else {
conn.(*net.TCPConn).SetNoDelay(true)
conn.(*net.TCPConn).SetLinger(0)
conn.(*net.TCPConn).SetKeepAlive(true)
return &ConnTimeout{conn: conn, addr: conn.RemoteAddr().String(), isRdma: false}
}
conn.(*net.TCPConn).SetNoDelay(true)
conn.(*net.TCPConn).SetLinger(0)
conn.(*net.TCPConn).SetKeepAlive(true)
return &ConnTimeout{conn: conn, addr: conn.RemoteAddr().String()}
}
func (c *ConnTimeout) SetReadTimeout(timeout time.Duration) {
@ -94,21 +69,6 @@ func (c *ConnTimeout) Read(p []byte) (n int, err error) {
return
}
func (c *ConnTimeout) ReadByRdma() (rdmaBuffer *rdma.RdmaBuffer, err error) {
conn, _ := c.conn.(*rdma.Connection)
return conn.GetRecvMsgBuffer()
}
func (c *ConnTimeout) ReleaseRxByRdma(rdmaBuffer *rdma.RdmaBuffer) (err error) {
conn, ok := c.conn.(*rdma.Connection)
if !ok {
return errors.New("release rx data buffer failed: rdma conn type conversion error")
}
conn.ReleaseConnRxDataBuffer(rdmaBuffer)
return
}
func (c *ConnTimeout) Write(p []byte) (n int, err error) {
if c.writeTime.Nanoseconds() > 0 {
err = c.conn.SetWriteDeadline(time.Now().Add(c.writeTime))
@ -121,44 +81,10 @@ func (c *ConnTimeout) Write(p []byte) (n int, err error) {
return
}
func (c *ConnTimeout) GetDataBuffer(len uint32) (*rdma.RdmaBuffer, error) {
conn, ok := c.conn.(*rdma.Connection)
if !ok {
return nil, errors.New("get data buffer failed: rdma conn type conversion error")
}
return conn.GetConnTxDataBuffer(len)
}
func (c *ConnTimeout) AddWriteRequest(rdmaBuffer *rdma.RdmaBuffer) error {
conn, ok := c.conn.(*rdma.Connection)
if !ok {
return errors.New("add write sge failed: rdma conn type conversion error")
}
return conn.AddWriteRequest(rdmaBuffer)
}
func (c *ConnTimeout) Flush() error {
conn, ok := c.conn.(*rdma.Connection)
if !ok {
return errors.New("flush age failed: rdma conn type conversion error")
}
return conn.FlushWriteRequest()
}
func (c *ConnTimeout) RemoteAddr() string {
return c.addr
}
func (c *ConnTimeout) GetRdmaConn() *rdma.Connection {
if c.isRdma {
return c.conn.(*rdma.Connection)
}
return nil
}
func (c *ConnTimeout) IsRdma() bool {
return c.isRdma
}
func (c *ConnTimeout) Close() error {
return c.conn.Close()
}

1
depends/tiglabs/raft/util/io.go
View File

@ -14,7 +14,6 @@
package util
import "C"
import (
"bufio"
"errors"

2
metanode/manager.go
View File

@ -17,7 +17,6 @@ package metanode
import (
"encoding/json"
"fmt"
"github.com/cubefs/cubefs/util/rdma"
syslog "log"
"net"
"os"
@ -37,6 +36,7 @@ import (
"github.com/cubefs/cubefs/util/exporter"
"github.com/cubefs/cubefs/util/loadutil"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/rdma"
"github.com/cubefs/cubefs/util/strutil"
)

5
metanode/manager_proxy.go
View File

@ -15,14 +15,13 @@
package metanode
import (
"github.com/cubefs/cubefs/util/rdma"
"net"
"strings"
"github.com/cubefs/cubefs/datanode/storage"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/rdma"
)
const (
@ -193,7 +192,7 @@ func (m *metadataManager) serveRdmaProxy(conn *rdma.Connection, mp MetaPartition
}
// read connection from the master
if err = p.ReadFromRdmaConn(mConn, proto.NoReadDeadlineTime); err != nil {
if err = p.ReadFromRdmaConnWithVer(mConn, proto.NoReadDeadlineTime); err != nil {
p.PacketErrorWithBody(proto.OpErr, []byte(err.Error()))
m.rdmaConnPool.PutRdmaConn(mConn, ForceClosedConnect)
goto end

8
metanode/manager_resp.go
View File

@ -15,12 +15,12 @@
package metanode
import (
"github.com/cubefs/cubefs/util/rdma"
"net"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/rdma"
)
// Reply operation results to the master.
@ -60,7 +60,11 @@ func (m *metadataManager) respondToClientWithVer(conn net.Conn, p *Packet) (err
if p.VerSeq > 0 {
p.ExtentType |= proto.MultiVersionFlag
}
err = p.WriteToConn(conn)
if c, ok := conn.(*rdma.Connection); ok {
err = p.WriteToRdmaConn(c)
} else {
err = p.WriteToConn(conn)
}
if err != nil {
log.LogErrorf("response to client[%s], "+
"request[%s], response packet[%s]",

4
metanode/server.go
View File

@ -16,7 +16,6 @@ package metanode
import (
"fmt"
"github.com/cubefs/cubefs/util/rdma"
"io"
"net"
@ -25,6 +24,7 @@ import (
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/rdma"
)
var (
@ -165,7 +165,7 @@ func (m *MetaNode) serveRdmaConn(conn *rdma.Connection, stopC chan uint8) {
default:
}
p := &Packet{}
if err := p.ReadFromRdmaConn(conn, proto.NoReadDeadlineTime); err != nil {
if err := p.ReadFromRdmaConnWithVer(conn, proto.NoReadDeadlineTime); err != nil {
log.LogError("rdma serve MetaNode: ", err.Error())
return
}

167
proto/packet.go
View File

@ -320,6 +320,8 @@ const (
DecommissionedCreateDataPartition = 1
)
const VerInfoCnt = 17
// Packet defines the packet structure.
type Packet struct {
Magic uint8
@ -796,12 +798,10 @@ func (p *Packet) UnmarshalHeader(in []byte) error {
return nil
}
const verInfoCnt = 17
func (p *Packet) MarshalVersionSlice() (data []byte, err error) {
items := p.VerList
cnt := len(items)
buff := bytes.NewBuffer(make([]byte, 0, 2*cnt*verInfoCnt))
buff := bytes.NewBuffer(make([]byte, 0, p.GetVersionListSize()))
if err := binary.Write(buff, binary.BigEndian, uint16(cnt)); err != nil {
return nil, err
}
@ -920,12 +920,21 @@ func (p *Packet) WriteToConn(c net.Conn) (err error) {
return
}
func (p *Packet) WriteExternalToRdmaConn(conn *rdma.Connection, rdmaBuffer *rdma.RdmaBuffer, len int) (err error) {
conn.SetWriteDeadline(time.Now().Add(WriteDeadlineTime * time.Second))
func (p *Packet) WriteExternalToRdmaConn(conn *rdma.Connection, rdmaBuffer *rdma.RdmaBuffer, size int) (err error) {
p.MarshalHeader(rdmaBuffer.Data[0:util.PacketHeaderSize])
verSize := 0
if p.IsVersionList() {
d, err1 := p.MarshalVersionSlice()
if err1 != nil {
log.LogErrorf("MarshalVersionSlice: marshal version ifo failed, err %s", err1.Error())
return err1
}
copy(rdmaBuffer.Data[util.PacketHeaderSize:], d)
verSize = int(p.GetVersionListSize())
}
size += verSize
if _, err = conn.WriteExternalBuffer(rdmaBuffer, len); err != nil {
if _, err = conn.WriteExternalBuffer(rdmaBuffer, size); err != nil {
return
}
return
@ -951,25 +960,63 @@ func (p *Packet) WriteToRdmaConn(conn *rdma.Connection) (err error) {
}
}()
headSize := util.PacketHeaderSize
if p.Opcode == OpRandomWriteVer || p.ExtentType&MultiVersionFlag > 0 {
headSize = util.PacketHeaderVerSize
}
conn.SetWriteDeadline(time.Now().Add(WriteDeadlineTime * time.Second))
if p.ArgLen > 0 {
if rdmaBuffer, err = conn.GetConnTxDataBuffer(util.RdmaPacketHeaderSize + p.Size); err != nil {
return
}
var verSize uint32
if p.IsVersionList() {
verSize = p.GetVersionListSize()
} else {
if rdmaBuffer, err = conn.GetConnTxDataBuffer(util.PacketHeaderSize + p.Size); err != nil {
return
}
verSize = 0
}
if p.ArgLen > 0 {
rdmaBuffer, err = conn.GetConnTxDataBuffer(uint32(headSize) + verSize + util.RdmaPacketArgSize + p.Size)
} else {
rdmaBuffer, err = conn.GetConnTxDataBuffer(uint32(headSize) + verSize + p.Size)
}
if err != nil {
return
}
dataBuffer = rdmaBuffer.Data
p.MarshalHeader(dataBuffer[:util.PacketHeaderSize])
if p.Opcode == OpRandomWriteVer || p.ExtentType&MultiVersionFlag > 0 {
p.MarshalHeader(dataBuffer[:util.PacketHeaderVerSize])
offset = util.PacketHeaderVerSize
} else {
p.MarshalHeader(dataBuffer[:util.PacketHeaderSize])
offset = util.PacketHeaderSize
}
offset = util.PacketHeaderSize
if p.IsVersionList() {
items := p.VerList
cnt := len(items)
var buff bytes.Buffer
if err := binary.Write(&buff, binary.BigEndian, uint16(cnt)); err != nil {
return err
}
for _, v := range items {
if err := binary.Write(&buff, binary.BigEndian, v.Ver); err != nil {
return err
}
if err := binary.Write(&buff, binary.BigEndian, v.DelTime); err != nil {
return err
}
if err := binary.Write(&buff, binary.BigEndian, v.Status); err != nil {
return err
}
}
copy(dataBuffer[offset:], buff.Bytes())
}
offset = uint32(headSize) + verSize
if p.ArgLen > 0 {
copy(dataBuffer[offset:offset+p.ArgLen], p.Arg)
offset = util.RdmaPacketHeaderSize
offset += util.RdmaPacketArgSize
}
if p.Data != nil && p.Size != 0 {
@ -1000,7 +1047,7 @@ func (p *Packet) IsReadOperation() bool {
p.Opcode == OpSnapshotExtentRepairRead
}
func (p *Packet) ReadFromRdmaConn(c *rdma.Connection, timeoutSec int) (err error) {
func (p *Packet) ReadFromRdmaConnWithVer(c *rdma.Connection, timeoutSec int) (err error) {
if timeoutSec != NoReadDeadlineTime {
c.SetReadDeadline(time.Now().Add(time.Second * time.Duration(timeoutSec)))
} else {
@ -1024,13 +1071,84 @@ func (p *Packet) ReadFromRdmaConn(c *rdma.Connection, timeoutSec int) (err error
if err = p.UnmarshalHeader(dataBuffer[0:util.PacketHeaderSize]); err != nil {
return
}
offset = util.PacketHeaderSize
if p.ExtentType&MultiVersionFlag > 0 {
ver := dataBuffer[offset : offset+8]
p.VerSeq = binary.BigEndian.Uint64(ver)
offset += 8
}
if p.IsVersionList() {
cntByte := dataBuffer[offset : offset+2]
cnt := binary.BigEndian.Uint16(cntByte)
log.LogDebugf("action[ReadFromConnWithVer] op %s verseq %v, extType %d, cnt %d",
p.GetOpMsg(), p.VerSeq, p.ExtentType, cnt)
verData := dataBuffer[offset : offset+uint32(cnt)*VerInfoCnt]
offset += 2 * uint32(cnt) * VerInfoCnt
err = p.UnmarshalVersionSlice(int(cnt), verData)
if err != nil {
log.LogWarnf("ReadFromConnWithVer: unmarshal ver slice failed, err %s", err.Error())
return err
}
}
if len(dataBuffer) == int(offset+util.RdmaPacketArgSize+p.Size) {
if p.ArgLen > 0 {
p.Arg = make([]byte, int(p.ArgLen))
copy(p.Arg, dataBuffer[offset:offset+p.ArgLen])
}
offset += util.RdmaPacketArgSize
}
if p.Size < 0 {
return syscall.EBADMSG
}
size := p.Size
if (p.Opcode == OpRead || p.Opcode == OpStreamRead || p.Opcode == OpExtentRepairRead || p.Opcode == OpStreamFollowerRead) && p.ResultCode == OpInitResultCode {
size = 0
}
if len(dataBuffer) < int(offset+size) {
log.LogErrorf("data is out of range. buffer len(%d), offset(%d), size(%d)", len(dataBuffer), offset, size)
return errors.New("data size is out of buffer")
}
p.Data = make([]byte, size)
copy(p.Data, dataBuffer[offset:offset+size])
p.IsRdma = true
return
}
/*
func (p *Packet) ReadFromRdmaConn(c *rdma.Connection, timeoutSec int) (err error) {
//if timeoutSec != NoReadDeadlineTime { //rdma todo
// c.SetReadDeadline(time.Now().Add(time.Second * time.Duration(timeoutSec)))
//} else {
// c.SetReadDeadline(time.Time{})
//}
var dataBuffer []byte
var offset uint32
if dataBuffer, err = c.GetRecvMsgBuffer(); err != nil {
return
}
defer func() {
c.ReleaseConnRxDataBuffer(dataBuffer) //rdma todo
}()
if err = p.UnmarshalHeader(dataBuffer[:util.PacketHeaderSize]); err != nil {
return
}
offset = util.PacketHeaderSize
if p.ArgLen > 0 {
p.Arg = make([]byte, int(p.ArgLen))
copy(p.Arg, dataBuffer[util.PacketHeaderSize:util.PacketHeaderSize+p.ArgLen])
offset = util.RdmaPacketHeaderSize
copy(p.Arg, dataBuffer[offset:offset+p.ArgLen])
offset += p.ArgLen
}
if p.Size < 0 {
@ -1042,9 +1160,10 @@ func (p *Packet) ReadFromRdmaConn(c *rdma.Connection, timeoutSec int) (err error
}
p.Data = make([]byte, size)
copy(p.Data, dataBuffer[offset:offset+size])
p.IsRdma = true
return
}
*/
// ReadFromConn reads the data from the given connection.
// Recognize the version bit and parse out version,
@ -1088,7 +1207,7 @@ func (p *Packet) ReadFromConnWithVer(c net.Conn, timeoutSec int) (err error) {
cnt := binary.BigEndian.Uint16(cntByte)
log.LogDebugf("action[ReadFromConnWithVer] op %s verseq %v, extType %d, cnt %d",
p.GetOpMsg(), p.VerSeq, p.ExtentType, cnt)
verData := make([]byte, cnt*verInfoCnt)
verData := make([]byte, cnt*VerInfoCnt)
if _, err = io.ReadFull(c, verData); err != nil {
log.LogWarnf("ReadFromConnWithVer: read ver slice from conn failed, err %s", err.Error())
return err
@ -1393,3 +1512,7 @@ func (p *Packet) IsBatchLockNormalExtents() bool {
func (p *Packet) IsBatchUnlockNormalExtents() bool {
return p.Opcode == OpBatchUnlockNormalExtent
}
func (p *Packet) GetVersionListSize() uint32 {
return uint32(2 * len(p.VerList) * VerInfoCnt)
}

144
sdk/data/stream/extent_handler.go
View File

@ -348,19 +348,22 @@ func (eh *ExtentHandler) processReply(packet *Packet) {
if err := eh.recoverPacket(packet); err != nil {
eh.discardPacket(packet)
log.LogErrorf("processReply discard packet: handler is in recovery status, inflight(%v) eh(%v) packet(%v) err(%v)", atomic.LoadInt32(&eh.inflight), eh, packet, err)
return
}
log.LogDebugf("processReply recover packet: handler is in recovery status, inflight(%v) from eh(%v) to recoverHandler(%v) packet(%v)", atomic.LoadInt32(&eh.inflight), eh, eh.recoverHandler, packet)
return
}
var reply *Packet
var verUpdate bool
reply := NewReply(packet.ReqID, packet.PartitionID, packet.ExtentID)
var err error
var allReply []*Packet
if IsRdma {
errs := make([]error, len(eh.rdmaConn))
allReply = make([]*Packet, len(eh.rdmaConn))
for index, conn := range eh.rdmaConn {
reply = NewReply(packet.ReqID, packet.PartitionID, packet.ExtentID)
errs[index] = reply.ReadFromRdmaConn(conn, proto.ReadDeadlineTime*3)
errs[index] = reply.ReadFromRdmaConnWithVer(conn, proto.ReadDeadlineTime*3)
allReply[index] = reply
}
for i := 0; i < len(eh.rdmaConn); i++ {
@ -370,68 +373,24 @@ func (eh *ExtentHandler) processReply(packet *Packet) {
return
}
if allReply[i].ResultCode != proto.OpOk {
errmsg := fmt.Sprintf("reply NOK: (%v) Reply(%v)", eh.dp.Hosts[i], allReply[i])
eh.processReplyError(packet, errmsg)
return
}
if !packet.isValidWriteReply(allReply[i]) {
errmsg := fmt.Sprintf("request and reply does not match: (%v) reply(%v)", eh.dp.Hosts[i], allReply[i])
eh.processReplyError(packet, errmsg)
return
}
if allReply[i].CRC != packet.CRC {
errmsg := fmt.Sprintf("inconsistent CRC: reqCRC(%v) replyCRC(%v) (%v) reply(%v) ", packet.CRC, allReply[i].CRC, eh.dp.Hosts[i], allReply[i])
eh.processReplyError(packet, errmsg)
if err, verUpdate = processSingleReply(eh, packet, allReply[i], errs[i]); err != nil {
return
}
}
} else {
err = reply.ReadFromConnWithVer(eh.conn, proto.ReadDeadlineTime)
log.LogDebugf("tcp conn recv reply: %v, err: %v", reply, err)
log.LogDebugf("tcp conn recv reply end: time[%v]", time.Now())
reply = NewReply(packet.ReqID, packet.PartitionID, packet.ExtentID)
err := reply.ReadFromConnWithVer(eh.conn, proto.ReadDeadlineTime)
err = reply.ReadFromConnWithVer(eh.conn, proto.ReadDeadlineTime)
log.LogDebugf("tcp conn recv reply: %v, err: %v", reply, err)
log.LogDebugf("tcp conn recv reply end: time[%v]", time.Now())
if err != nil {
eh.processReplyError(packet, err.Error())
return
}
if reply.VerSeq > atomic.LoadUint64(&eh.stream.verSeq) || (eh.key != nil && reply.VerSeq > eh.key.GetSeq()) {
log.LogDebugf("processReply.UpdateLatestVer update verseq according to data rsp from version %v to %v", eh.stream.verSeq, reply.VerSeq)
if err = eh.stream.client.UpdateLatestVer(&proto.VolVersionInfoList{VerList: reply.VerList}); err != nil {
eh.processReplyError(packet, err.Error())
return
}
if err = eh.appendExtentKey(); err != nil {
eh.processReplyError(packet, err.Error())
return
}
eh.key = nil
verUpdate = true
}
// NOTE: if meet a io limited error
if reply.ResultCode == proto.OpLimitedIoErr {
log.LogWarnf("[processReply] eh(%v) packet(%v) reply(%v) try again", eh, packet, reply)
eh.meetLimitedIoError = true
time.Sleep(StreamSendSleepInterval)
}
if reply.ResultCode != proto.OpOk {
if reply.ResultCode != proto.ErrCodeVersionOpError {
errmsg := fmt.Sprintf("reply NOK: reply(%v)", reply)
log.LogDebugf("processReply packet (%v) errmsg (%v)", packet, errmsg)
eh.processReplyError(packet, errmsg)
return
}
// todo(leonchang) need check safety
log.LogWarnf("processReply: get reply, eh(%v) packet(%v) reply(%v)", eh, packet, reply)
eh.stream.GetExtentsForceRefresh()
}
if reply.CRC != packet.CRC {
errmsg := fmt.Sprintf("inconsistent CRC: reqCRC(%v) replyCRC(%v) reply(%v) ", packet.CRC, reply.CRC, reply)
eh.processReplyError(packet, errmsg)
if err, verUpdate = processSingleReply(eh, packet, reply, err); err != nil {
return
}
}
@ -477,8 +436,55 @@ func (eh *ExtentHandler) processReply(packet *Packet) {
packet.Data = nil
eh.dirty = true
}
return
func processSingleReply(eh *ExtentHandler, packet *Packet, reply *Packet, err error) (error, bool) {
verUpdate := false
if reply.VerSeq > atomic.LoadUint64(&eh.stream.verSeq) || (eh.key != nil && reply.VerSeq > eh.key.GetSeq()) {
log.LogDebugf("processReply.UpdateLatestVer update verseq according to data rsp from version %v to %v", eh.stream.verSeq, reply.VerSeq)
if err := eh.stream.client.UpdateLatestVer(&proto.VolVersionInfoList{VerList: reply.VerList}); err != nil {
eh.processReplyError(packet, err.Error())
return err, verUpdate
}
if err := eh.appendExtentKey(); err != nil {
eh.processReplyError(packet, err.Error())
return err, verUpdate
}
eh.key = nil
verUpdate = true
}
if reply.ResultCode == proto.OpLimitedIoErr {
log.LogWarnf("[processReply] eh(%v) packet(%v) reply(%v) try again", eh, packet, reply)
eh.meetLimitedIoError = true
time.Sleep(StreamSendSleepInterval)
}
if reply.ResultCode != proto.OpOk {
if reply.ResultCode != proto.ErrCodeVersionOpError {
errmsg := fmt.Sprintf("reply NOK: reply(%v)", reply)
log.LogDebugf("processReply packet (%v) errmsg (%v)", packet, errmsg)
eh.processReplyError(packet, errmsg)
return errors.New(errmsg), verUpdate
}
log.LogWarnf("processReply: get reply, eh(%v) packet(%v) reply(%v)", eh, packet, reply)
eh.stream.GetExtentsForceRefresh()
}
if !packet.isValidWriteReply(reply) {
errmsg := fmt.Sprintf("request and reply does not match: reply(%v)", reply)
eh.processReplyError(packet, errmsg)
return errors.New(errmsg), verUpdate
}
if reply.CRC != packet.CRC {
errmsg := fmt.Sprintf("inconsistent CRC: reqCRC(%v) replyCRC(%v) reply(%v) ", packet.CRC, reply.CRC, reply)
eh.processReplyError(packet, errmsg)
return errors.New(errmsg), verUpdate
}
return nil, verUpdate
}
func (eh *ExtentHandler) processReplyError(packet *Packet, errmsg string) {
@ -527,14 +533,16 @@ func (eh *ExtentHandler) cleanup() (err error) {
status := eh.getStatus()
StreamConnPool.PutConnect(conn, status >= ExtentStatusRecovery)
}
if eh.rdmaConn != nil {
rdmaConn := eh.rdmaConn
eh.rdmaConn = nil
if status := eh.getStatus(); status >= ExtentStatusRecovery {
StreamRdmaConnPool.PutRdmaConn(rdmaConn, true)
} else {
StreamRdmaConnPool.PutRdmaConn(rdmaConn, false)
if len(eh.rdmaConn) != 0 {
for _, conn := range eh.rdmaConn {
if status := eh.getStatus(); status >= ExtentStatusRecovery {
StreamRdmaConnPool.PutRdmaConn(conn, true)
} else {
StreamRdmaConnPool.PutRdmaConn(conn, false)
}
}
eh.rdmaConn = nil
}
close(eh.stop)
})
@ -730,6 +738,12 @@ func (eh *ExtentHandler) allocateExtentTcp() (err error) {
extID, err = eh.createExtent(dp)
}
if err != nil {
// NOTE: try again
if strings.Contains(err.Error(), "Again") || strings.Contains(err.Error(), "LimitedIoErr") {
log.LogWarnf("[allocateExtent] eh(%v) try again", eh)
time.Sleep(time.Second * time.Duration(i+1))
continue
}
log.LogWarnf("allocateExtentTcp: exclude dp[%v] for write caused by create extent failed, eh(%v) err(%v) exclude(%v)",
dp, eh, err, exclude)
eh.stream.client.dataWrapper.RemoveDataPartitionForWrite(dp.PartitionID)
@ -821,9 +835,9 @@ func (eh *ExtentHandler) allocateExtentRdma() (err error) {
}
eh.mulCopy = false
//if eh.storeMode == proto.NormalExtentType && eh.stream.client.LimitManager.GetWriteSpeed() <= (300*util.MB) {
// eh.mulCopy = true
//}
if eh.storeMode == proto.NormalExtentType && eh.stream.client.LimitManager.GetWriteSpeed() <= (300*util.MB) {
eh.mulCopy = true
}
if !eh.mulCopy {
rdmaConn = make([]*rdma.Connection, 0, 1)

4
sdk/data/stream/packet.go
View File

@ -118,7 +118,7 @@ func NewOverwriteByAppendPacket(dp *wrapper.DataPartition, extentID uint64, exte
}
// NewOverwritePacket returns a new overwrite packet.
func NewOverwritePacket(dp *wrapper.DataPartition, extentID uint64, extentOffset int, inode uint64, fileOffset int, isSnap bool) *Packet {
func NewOverwritePacket(dp *wrapper.DataPartition, extentID uint64, extentOffset int, inode uint64, fileOffset int, isSnap bool, verSize uint32) *Packet {
p := new(Packet)
p.PartitionID = dp.PartitionID
p.Magic = proto.ProtoMagic
@ -137,7 +137,7 @@ func NewOverwritePacket(dp *wrapper.DataPartition, extentID uint64, extentOffset
p.inode = inode
p.KernelOffset = uint64(fileOffset)
if IsRdma {
rdmaBuffer, err := rdma.GetDataBuffer(util.BlockSize + util.PacketHeaderSize)
rdmaBuffer, err := rdma.GetDataBuffer(util.BlockSize + util.PacketHeaderSize + verSize)
if err == nil {
p.Data = rdmaBuffer.Data[util.PacketHeaderSize:]
p.RdmaBuffer = rdmaBuffer

68
sdk/data/stream/stream_conn.go
View File

@ -132,7 +132,7 @@ func (sc *StreamConn) Send(retry *bool, req *Packet, getReply GetReplyFunc, isRd
retryInterval := StreamSendSleepInterval
for i := 0; i < StreamSendMaxRetry; i++ {
err = sc.sendToPartition(req, retry, getReply, isRdma)
err = sc.sendToDataPartition(req, retry, getReply, isRdma)
if err == nil || err == proto.ErrCodeVersionOp || !*retry || err == TryOtherAddrError || strings.Contains(err.Error(), "OpForbidErr") {
return
}
@ -153,11 +153,11 @@ func (sc *StreamConn) Send(retry *bool, req *Packet, getReply GetReplyFunc, isRd
return errors.NewErrorf("StreamConn Send: retried %v times and still failed, sc(%v) reqPacket(%v), err %s", StreamSendMaxRetry, sc, req, err.Error())
}
func (sc *StreamConn) sendToPartition(req *Packet, retry *bool, getReply GetReplyFunc, isRdma bool) (err error) {
func (sc *StreamConn) sendToDataPartition(req *Packet, retry *bool, getReply GetReplyFunc, isRdma bool) (err error) {
if isRdma {
return sc.sendToPartitionByRdma(req, retry, getReply)
} else {
return sc.sendToDataPartition(req, retry, getReply)
return sc.sendToPartitionByTcp(req, retry, getReply)
}
}
@ -169,28 +169,16 @@ func (sc *StreamConn) sendToPartitionByRdma(req *Packet, retry *bool, getReply G
log.LogDebugf("req opcode %v, conn %v", req.Opcode, conn)
err = sc.sendToRdmaConn(conn, req, getReply)
if err == nil {
if req.Size != 0 && req.Data != nil { //OpRandomWrite
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
} else { //OpStreamRead || OpStreamFollowerRead
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
}
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
StreamRdmaConnPool.PutRdmaConn(conn, false)
return
}
if err != TryOtherAddrError || !*retry {
if req.Size != 0 && req.Data != nil { //OpRandomWrite
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
} else { //OpStreamRead || OpStreamFollowerRead
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
}
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
StreamRdmaConnPool.PutRdmaConn(conn, true)
return
}
if req.Size != 0 && req.Data != nil { //OpRandomWrite
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
} else { //OpStreamRead || OpStreamFollowerRead
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
}
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
StreamRdmaConnPool.PutRdmaConn(conn, true)
log.LogWarnf("sendToPartition: send to curr addr failed, addr(%v) reqPacket(%v) err(%v)", sc.currAddr, req, err)
} else {
@ -210,35 +198,23 @@ func (sc *StreamConn) sendToPartitionByRdma(req *Packet, retry *bool, getReply G
sc.dp.LeaderAddr = addr
err = sc.sendToRdmaConn(conn, req, getReply)
if err == nil {
if req.Size != 0 && req.Data != nil { //OpRandomWrite
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
} else { //OpStreamRead || OpStreamFollowerRead
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
}
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
StreamRdmaConnPool.PutRdmaConn(conn, false)
return
}
if err != TryOtherAddrError {
if req.Size != 0 && req.Data != nil { //OpRandomWrite
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
} else { //OpStreamRead || OpStreamFollowerRead
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
}
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
StreamRdmaConnPool.PutRdmaConn(conn, true)
return
}
if req.Size != 0 && req.Data != nil { //OpRandomWrite
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
} else { //OpStreamRead || OpStreamFollowerRead
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
}
conn.ReleaseConnExternalDataBuffer(req.RdmaBuffer)
StreamRdmaConnPool.PutRdmaConn(conn, true)
log.LogWarnf("sendToPartition: try addr(%v) failed! reqPacket(%v) err(%v)", rdmaAddr, req, err)
}
return errors.New(fmt.Sprintf("sendToPatition Failed: sc(%v) reqPacket(%v)", sc, req))
}
func (sc *StreamConn) sendToDataPartition(req *Packet, retry *bool, getReply GetReplyFunc) (err error) {
func (sc *StreamConn) sendToPartitionByTcp(req *Packet, retry *bool, getReply GetReplyFunc) (err error) {
conn, err := StreamConnPool.GetConnect(sc.currAddr)
if err == nil {
@ -249,22 +225,23 @@ func (sc *StreamConn) sendToDataPartition(req *Packet, retry *bool, getReply Get
StreamConnPool.PutConnect(conn, false)
return
}
log.LogWarnf("sendToDataPartition: send to curr addr failed, addr(%v) reqPacket(%v) err(%v)", sc.currAddr, req, err)
log.LogWarnf("sendToPartitionByTcp: send to curr addr failed, addr(%v) reqPacket(%v) err(%v)", sc.currAddr, req, err)
StreamConnPool.PutConnectEx(conn, err)
if err != TryOtherAddrError || !*retry {
return
}
} else {
log.LogWarnf("sendToDataPartition: get connection to curr addr failed, addr(%v) reqPacket(%v) err(%v)", sc.currAddr, req, err)
log.LogWarnf("sendToPartitionByTcp: get connection to curr addr failed, addr(%v) reqPacket(%v) err(%v)", sc.currAddr, req, err)
}
hosts := sortByStatus(sc.dp, true)
for _, addr := range hosts {
log.LogWarnf("sendToDataPartition: try addr(%v) reqPacket(%v)", addr, req)
log.LogWarnf("sendToPartitionByTcp: try addr(%v) reqPacket(%v)", addr, req)
conn, err = StreamConnPool.GetConnect(addr)
if err != nil {
log.LogWarnf("sendToDataPartition: failed to get connection to addr(%v) reqPacket(%v) err(%v)", addr, req, err)
log.LogWarnf("sendToPartitionByTcp: failed to get connection to addr(%v) reqPacket(%v) err(%v)", addr, req, err)
continue
}
sc.currAddr = addr
@ -278,21 +255,16 @@ func (sc *StreamConn) sendToDataPartition(req *Packet, retry *bool, getReply Get
if err != TryOtherAddrError {
return
}
log.LogWarnf("sendToDataPartition: try addr(%v) failed! reqPacket(%v) err(%v)", addr, req, err)
log.LogWarnf("sendToPartitionByTcp: try addr(%v) failed! reqPacket(%v) err(%v)", addr, req, err)
}
return errors.New(fmt.Sprintf("sendToPatition Failed: sc(%v) reqPacket(%v)", sc, req))
return errors.New(fmt.Sprintf("sendToPartitionByTcp Failed: sc(%v) reqPacket(%v)", sc, req))
}
func (sc *StreamConn) sendToRdmaConn(conn *rdma.Connection, req *Packet, getReply GetReplyFunc) (err error) {
rdmaAddr := wrapper.GetDpRdmaAddr(sc.currAddr)
for i := 0; i < StreamSendMaxRetry; i++ {
log.LogDebugf("sendToRdmaConn: send to addr(%v), reqPacket(%v)", rdmaAddr, req)
if req.Size != 0 && req.Data != nil { //OpRandomWrite
err = req.WriteExternalToRdmaConn(conn, req.RdmaBuffer, int(util.PacketHeaderSize+req.Size))
} else { //OpStreamRead || OpStreamFollowerRead
err = req.WriteExternalToRdmaConn(conn, req.RdmaBuffer, util.PacketHeaderSize)
}
err = req.WriteExternalToRdmaConn(conn, req.RdmaBuffer, int(util.PacketHeaderSize+req.Size))
if err != nil {
msg := fmt.Sprintf("sendToRdmaConn: failed to write to addr(%v) err(%v)", rdmaAddr, err)
log.LogWarn(msg)
@ -312,7 +284,7 @@ func (sc *StreamConn) sendToRdmaConn(conn *rdma.Connection, req *Packet, getRepl
time.Sleep(StreamSendSleepInterval)
}
log.LogDebugf("sendToConn exit: send to addr(%v) reqPacket(%v) err(%v)", rdmaAddr, req, err)
log.LogDebugf("sendToRdmaConn exit: send to addr(%v) reqPacket(%v) err(%v)", rdmaAddr, req, err)
return
}
@ -336,7 +308,7 @@ func (sc *StreamConn) sendToTcpConn(conn *net.TCPConn, req *Packet, getReply Get
}
// NOTE: if we meet a try again error
if err == LimitedIoError {
log.LogWarnf("sendToConn: found ulimit io error, dp %d, req %d, err %s", req.PartitionID, req.ReqID, err.Error())
log.LogWarnf("sendToTcpConn: found ulimit io error, dp %d, req %d, err %s", req.PartitionID, req.ReqID, err.Error())
i -= 1
}

24
sdk/data/stream/stream_writer.go
View File

@ -30,7 +30,6 @@ import (
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/rdma"
"github.com/cubefs/cubefs/util/stat"
)
const (
@ -493,7 +492,7 @@ func (s *Streamer) doDirectWriteByAppend(req *ExtentRequest, direct bool, op uin
var e error
if IsRdma {
c, _ := conn.(*rdma.Connection)
e = replyPacket.RecvRespFromRDMAConn(c, proto.ReadDeadlineTime)
e = replyPacket.ReadFromRdmaConnWithVer(c, proto.ReadDeadlineTime)
} else {
e = replyPacket.ReadFromConnWithVer(conn, proto.ReadDeadlineTime)
}
@ -633,10 +632,11 @@ func (s *Streamer) doOverwrite(req *ExtentRequest, direct bool) (total int, err
}
sc := NewStreamConn(dp, false, s.client.streamRetryTimeout)
verSize := s.GetVerListSize()
for total < size {
reqPacket := NewOverwritePacket(dp, req.ExtentKey.ExtentId, offset-ekFileOffset+total+ekExtOffset,
s.inode, offset, s.client.dataWrapper.IsSnapshotEnabled)
s.inode, offset, s.client.dataWrapper.IsSnapshotEnabled, verSize)
if s.client.dataWrapper.IsSnapshotEnabled {
reqPacket.VerSeq = s.verSeq
reqPacket.VerList = make([]*proto.VolVersionInfo, len(s.client.multiVerMgr.verList.VerList))
@ -665,7 +665,7 @@ func (s *Streamer) doOverwrite(req *ExtentRequest, direct bool) (total int, err
if IsRdma {
c, _ := conn.(*rdma.Connection)
e = replyPacket.ReadFromRdmaConn(c, proto.ReadDeadlineTime)
e = replyPacket.ReadFromRdmaConnWithVer(c, proto.ReadDeadlineTime)
} else {
e = replyPacket.ReadFromConnWithVer(conn, proto.ReadDeadlineTime)
}
@ -727,8 +727,6 @@ func (s *Streamer) doOverwrite(req *ExtentRequest, direct bool) (total int, err
return
}
<<<<<<< HEAD
<<<<<<< HEAD
func (s *Streamer) tryInitExtentHandlerByLastEk(offset, size int) (isLastEkVerNotEqual bool) {
storeMode := s.GetStoreMod(offset, size)
getEndEkFunc := func() *proto.ExtentKey {
@ -836,7 +834,6 @@ func (s *Streamer) doWriteAppend(req *ExtentRequest, direct bool) (writeSize int
}
func (s *Streamer) doWriteAppendEx(data []byte, offset, size int, direct bool, reUseEk bool) (total int, err error, status int32) {
bgTime := stat.BeginStat()
var (
ek *proto.ExtentKey
storeMode int
@ -847,8 +844,6 @@ func (s *Streamer) doWriteAppendEx(data []byte, offset, size int, direct bool, r
storeMode = s.GetStoreMod(offset, size)
log.LogDebugf("doWriteAppendEx enter: ino(%v) offset(%v) size(%v) storeMode(%v)", s.inode, offset, size, storeMode)
stat.EndStat("write(doWrite print log)", nil, bgTime, 1)
if proto.IsHot(s.client.volumeType) {
if reUseEk {
if isLastEkVerNotEqual := s.tryInitExtentHandlerByLastEk(offset, size); isLastEkVerNotEqual {
@ -859,10 +854,6 @@ func (s *Streamer) doWriteAppendEx(data []byte, offset, size int, direct bool, r
} else if s.handler != nil {
s.closeOpenHandler()
}
stat.EndStat("write(doWrite get ek)", nil, bgTime, 1)
bgTime1 := stat.BeginStat()
for i := 0; i < MaxNewHandlerRetry; i++ {
if s.handler == nil {
s.handler = NewExtentHandler(s, offset, storeMode, 0)
@ -917,9 +908,6 @@ func (s *Streamer) doWriteAppendEx(data []byte, offset, size int, direct bool, r
total = size
log.LogDebugf("doWrite exit: ino(%v) offset(%v) size(%v) ek(%v)", s.inode, offset, size, ek)
stat.EndStat("write(doWrite)", nil, bgTime, 1)
return
}
@ -1105,3 +1093,7 @@ func (s *Streamer) updateVer(verSeq uint64) (err error) {
func (s *Streamer) tinySizeLimit() int {
return util.DefaultTinySizeLimit
}
func (s *Streamer) GetVerListSize() uint32 {
return uint32(2 * len(s.client.multiVerMgr.verList.VerList) * proto.VerInfoCnt)
}

2
sdk/data/wrapper/data_partition.go
View File

@ -16,7 +16,6 @@ package wrapper
import (
"fmt"
"github.com/cubefs/cubefs/util/rdma"
"net"
"strings"
"sync"
@ -26,6 +25,7 @@ import (
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/rdma"
)
// DataPartition defines the wrapper of the data partition.

2
sdk/meta/conn.go
View File

@ -287,7 +287,7 @@ func (mc *MetaConn) sendByRdma(req *proto.Packet) (resp *proto.Packet, err error
return nil, errors.Trace(err, "Failed to write to rdma conn, req(%v)", req)
}
resp = proto.NewPacket()
err = resp.ReadFromRdmaConn(mc.rdmaConn, proto.ReadDeadlineTime)
err = resp.ReadFromRdmaConnWithVer(mc.rdmaConn, proto.ReadDeadlineTime)
if err != nil {
return nil, errors.Trace(err, "Failed to read from rdma conn, req(%v)", req)
}

1
util/unit.go
View File

@ -52,6 +52,7 @@ const (
SyscallTryMaxTimes = 3
PacketHeaderVerSize = 65
RdmaPacketHeaderSize = 121
RdmaPacketArgSize = 64
)
const (