From df9c764322dd36ea3140f58d7832fc2038d03d72 Mon Sep 17 00:00:00 2001 From: Wu Huocheng Date: Fri, 8 Nov 2024 15:39:14 +0800 Subject: [PATCH] @formatter:off fix(rdma): fix the merge codes mistake. Merge the rdma branch into release V3.4.0. Signed-off-by: Wu Huocheng --- build/build.sh | 17 +- client/fuse.go | 1 - datanode/data_partition_repair.go | 67 +++++--- datanode/partition_raft.go | 2 - datanode/repl/packet.go | 121 +++++++++++++- datanode/repl/repl_protocol.go | 125 +++++++++++---- datanode/server.go | 4 +- datanode/wrap_operator.go | 1 + datanode/wrap_prepare.go | 1 - depends/tiglabs/raft/proto/codec.go | 141 ----------------- depends/tiglabs/raft/transport_heartbeat.go | 64 ++------ depends/tiglabs/raft/transport_multi.go | 23 +-- depends/tiglabs/raft/transport_replicate.go | 85 +++------- depends/tiglabs/raft/transport_sender.go | 66 +------- depends/tiglabs/raft/util/conn.go | 84 +--------- depends/tiglabs/raft/util/io.go | 1 - metanode/manager.go | 2 +- metanode/manager_proxy.go | 5 +- metanode/manager_resp.go | 8 +- metanode/server.go | 4 +- proto/packet.go | 167 +++++++++++++++++--- sdk/data/stream/extent_handler.go | 144 +++++++++-------- sdk/data/stream/packet.go | 4 +- sdk/data/stream/stream_conn.go | 68 +++----- sdk/data/stream/stream_writer.go | 24 +-- sdk/data/wrapper/data_partition.go | 2 +- sdk/meta/conn.go | 2 +- util/unit.go | 1 + 28 files changed, 588 insertions(+), 646 deletions(-) diff --git a/build/build.sh b/build/build.sh index fc852c723..7fdc4336f 100755 --- a/build/build.sh +++ b/build/build.sh @@ -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 } diff --git a/client/fuse.go b/client/fuse.go index 7c008c1d7..13c1fda87 100644 --- a/client/fuse.go +++ b/client/fuse.go @@ -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" diff --git a/datanode/data_partition_repair.go b/datanode/data_partition_repair.go index 118d5f7a8..a23e7daad 100644 --- a/datanode/data_partition_repair.go +++ b/datanode/data_partition_repair.go @@ -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.", diff --git a/datanode/partition_raft.go b/datanode/partition_raft.go index b372b4f8c..afa804cb2 100644 --- a/datanode/partition_raft.go +++ b/datanode/partition_raft.go @@ -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 } diff --git a/datanode/repl/packet.go b/datanode/repl/packet.go index 1f90aad4b..e75630e69 100644 --- a/datanode/repl/packet.go +++ b/datanode/repl/packet.go @@ -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 +} diff --git a/datanode/repl/repl_protocol.go b/datanode/repl/repl_protocol.go index 013e4b06c..15df7089c 100644 --- a/datanode/repl/repl_protocol.go +++ b/datanode/repl/repl_protocol.go @@ -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 } diff --git a/datanode/server.go b/datanode/server.go index 50f127113..06ab737a6 100644 --- a/datanode/server.go +++ b/datanode/server.go @@ -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)) diff --git a/datanode/wrap_operator.go b/datanode/wrap_operator.go index 609b44bfe..6b3bf31e3 100644 --- a/datanode/wrap_operator.go +++ b/datanode/wrap_operator.go @@ -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" ) diff --git a/datanode/wrap_prepare.go b/datanode/wrap_prepare.go index 64403fa5b..a4869b1b4 100644 --- a/datanode/wrap_prepare.go +++ b/datanode/wrap_prepare.go @@ -17,7 +17,6 @@ package datanode import ( "encoding/json" "fmt" - "github.com/cubefs/cubefs/util/log" "hash/crc32" "sync/atomic" diff --git a/depends/tiglabs/raft/proto/codec.go b/depends/tiglabs/raft/proto/codec.go index d0a3aa915..112bc042e 100644 --- a/depends/tiglabs/raft/proto/codec.go +++ b/depends/tiglabs/raft/proto/codec.go @@ -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] diff --git a/depends/tiglabs/raft/transport_heartbeat.go b/depends/tiglabs/raft/transport_heartbeat.go index 37f5ad990..5032c69a8 100644 --- a/depends/tiglabs/raft/transport_heartbeat.go +++ b/depends/tiglabs/raft/transport_heartbeat.go @@ -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) } } } - }) } diff --git a/depends/tiglabs/raft/transport_multi.go b/depends/tiglabs/raft/transport_multi.go index 40dfd6b02..abfc55e15 100644 --- a/depends/tiglabs/raft/transport_multi.go +++ b/depends/tiglabs/raft/transport_multi.go @@ -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 -} diff --git a/depends/tiglabs/raft/transport_replicate.go b/depends/tiglabs/raft/transport_replicate.go index 00decd92b..c2188f6ae 100644 --- a/depends/tiglabs/raft/transport_replicate.go +++ b/depends/tiglabs/raft/transport_replicate.go @@ -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) } } } diff --git a/depends/tiglabs/raft/transport_sender.go b/depends/tiglabs/raft/transport_sender.go index 7c0cfb951..4c3ed6d85 100644 --- a/depends/tiglabs/raft/transport_sender.go +++ b/depends/tiglabs/raft/transport_sender.go @@ -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 -} diff --git a/depends/tiglabs/raft/util/conn.go b/depends/tiglabs/raft/util/conn.go index da80fdd9b..46197d0fb 100644 --- a/depends/tiglabs/raft/util/conn.go +++ b/depends/tiglabs/raft/util/conn.go @@ -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() } diff --git a/depends/tiglabs/raft/util/io.go b/depends/tiglabs/raft/util/io.go index 71214a185..c04c544a3 100644 --- a/depends/tiglabs/raft/util/io.go +++ b/depends/tiglabs/raft/util/io.go @@ -14,7 +14,6 @@ package util -import "C" import ( "bufio" "errors" diff --git a/metanode/manager.go b/metanode/manager.go index bfb4fbbd9..2762eb39b 100644 --- a/metanode/manager.go +++ b/metanode/manager.go @@ -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" ) diff --git a/metanode/manager_proxy.go b/metanode/manager_proxy.go index 4b68cc78f..3ce8bd6e2 100644 --- a/metanode/manager_proxy.go +++ b/metanode/manager_proxy.go @@ -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 diff --git a/metanode/manager_resp.go b/metanode/manager_resp.go index ee230b5b8..d3d9c4a31 100644 --- a/metanode/manager_resp.go +++ b/metanode/manager_resp.go @@ -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]", diff --git a/metanode/server.go b/metanode/server.go index e11cb52b4..7d4176331 100644 --- a/metanode/server.go +++ b/metanode/server.go @@ -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 } diff --git a/proto/packet.go b/proto/packet.go index dacd4f927..a35574990 100644 --- a/proto/packet.go +++ b/proto/packet.go @@ -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) +} diff --git a/sdk/data/stream/extent_handler.go b/sdk/data/stream/extent_handler.go index 01bc8da9e..d55ebd91f 100644 --- a/sdk/data/stream/extent_handler.go +++ b/sdk/data/stream/extent_handler.go @@ -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) diff --git a/sdk/data/stream/packet.go b/sdk/data/stream/packet.go index 7807c7987..f091e2ae7 100644 --- a/sdk/data/stream/packet.go +++ b/sdk/data/stream/packet.go @@ -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 diff --git a/sdk/data/stream/stream_conn.go b/sdk/data/stream/stream_conn.go index 082a7d44c..452319b3e 100644 --- a/sdk/data/stream/stream_conn.go +++ b/sdk/data/stream/stream_conn.go @@ -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 } diff --git a/sdk/data/stream/stream_writer.go b/sdk/data/stream/stream_writer.go index 9f6fd650a..b8049df99 100644 --- a/sdk/data/stream/stream_writer.go +++ b/sdk/data/stream/stream_writer.go @@ -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) +} diff --git a/sdk/data/wrapper/data_partition.go b/sdk/data/wrapper/data_partition.go index 2743c53a8..c53b8adc1 100644 --- a/sdk/data/wrapper/data_partition.go +++ b/sdk/data/wrapper/data_partition.go @@ -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. diff --git a/sdk/meta/conn.go b/sdk/meta/conn.go index 17b0aaa4f..c481b6181 100644 --- a/sdk/meta/conn.go +++ b/sdk/meta/conn.go @@ -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) } diff --git a/util/unit.go b/util/unit.go index 144b2af41..0bc149389 100644 --- a/util/unit.go +++ b/util/unit.go @@ -52,6 +52,7 @@ const ( SyscallTryMaxTimes = 3 PacketHeaderVerSize = 65 RdmaPacketHeaderSize = 121 + RdmaPacketArgSize = 64 ) const (