Hi bro.Minh,
- In my understanding, tx probation timer only start when sender send
first message.
Then sender relies on chunkAck to know receiver support MDS FCTRL or
timeout as not support.
But from what you describe, sender got tx probation timer timeout
before sending first message?
Or after sending first message but sender cannot get any chunkAck somehow?
I am confused this point. Could you help explain?
- About the code, mistake set '0' twice for .acked_ in
TipcPortId::ReceiveReset()
Best Regards,
ThuanTr
-----Original Message-----
From: Minh Chau <minh.c...@dektech.com.au>
Sent: Friday, October 11, 2019 10:52 AM
To: hans.nordeb...@ericsson.com; gary....@dektech.com.au;
vu.m.ngu...@dektech.com.au; thuan.t...@dektech.com.au
Cc: opensaf-devel@lists.sourceforge.net; Minh Chau
<minh.c...@dektech.com.au>
Subject: [PATCH 1/1] mds: Add Reset message [#3090]
mds relies on data message sent from the peer to determine whether the
MDS_TIPC_FCTRL_ENABLED is set. The data message may not be sent right
after TIPC_PUBLISHED event, which can cause the tx probation timer timeout.
This patch add Reset message, which is sent right after the
TIPC_PUBLISHED to help mds determine the flow control supported at the peer
earlier.
---
src/mds/mds_main.c | 2 +-
src/mds/mds_tipc_fctrl_intf.cc | 27 ++++++++++++++++++++++
src/mds/mds_tipc_fctrl_msg.cc | 11 +++++++++
src/mds/mds_tipc_fctrl_msg.h | 18 +++++++++++++++
src/mds/mds_tipc_fctrl_portid.cc | 49
++++++++++++++++++++++++++++++----------
src/mds/mds_tipc_fctrl_portid.h | 2 ++
6 files changed, 96 insertions(+), 13 deletions(-)
diff --git a/src/mds/mds_main.c b/src/mds/mds_main.c index
8c9b1f1..c7d2f7b
100644
--- a/src/mds/mds_main.c
+++ b/src/mds/mds_main.c
@@ -408,7 +408,7 @@ uint32_t mds_lib_req(NCS_LIB_REQ_INFO *req)
if (tipc_mcast_enabled != false)
tipc_mcast_enabled = true;
- m_MDS_LOG_DBG(
+ m_MDS_LOG_NOTIFY(
"MDS: TIPC_MCAST_ENABLED: %d Set argument
\n",
tipc_mcast_enabled);
}
diff --git a/src/mds/mds_tipc_fctrl_intf.cc
b/src/mds/mds_tipc_fctrl_intf.cc index 6271890..e8c9121 100644
--- a/src/mds/mds_tipc_fctrl_intf.cc
+++ b/src/mds/mds_tipc_fctrl_intf.cc
@@ -39,6 +39,7 @@ using mds::DataMessage; using mds::ChunkAck; using
mds::HeaderMessage; using mds::Nack;
+using mds::Reset;
namespace {
// flow control enabled/disabled
@@ -124,12 +125,20 @@ uint32_t process_flow_event(const Event& evt) {
uint32_t rc = NCSCC_RC_SUCCESS;
TipcPortId *portid = portid_lookup(evt.id_);
if (portid == nullptr) {
+ // the null portid normally should not happen; however because the
+ // tipc_cb.Dsock and tipc_cb.BSRsock are separated; the data message
+ // sent from BSRsock may come before reception of TIPC_PUBLISHED
if (evt.type_ == Event::Type::kEvtRcvData) {
portid = new TipcPortId(evt.id_, data_sock_fd,
kChunkAckSize, sock_buf_size);
portid_map[TipcPortId::GetUniqueId(evt.id_)] = portid;
rc = portid->ReceiveData(evt.mseq_, evt.mfrag_,
evt.fseq_, evt.svc_id_);
+ } else if (evt.type_ == Event::Type::kEvtRcvReset) {
+ portid = new TipcPortId(evt.id_, data_sock_fd,
+ kChunkAckSize, sock_buf_size);
+ portid_map[TipcPortId::GetUniqueId(evt.id_)] = portid;
+ portid->ReceiveReset();
} else {
m_MDS_LOG_ERR("FCTRL: [me] <-- [node:%x, ref:%u], "
"RcvEvt[evt:%d], Error[PortId not found]", @@ -151,6
+160,9 @@ uint32_t process_flow_event(const Event& evt) {
portid->ReceiveNack(evt.mseq_, evt.mfrag_,
evt.fseq_);
}
+ if (evt.type_ == Event::Type::kEvtRcvReset) {
+ portid->ReceiveReset();
+ }
}
return rc;
}
@@ -489,6 +501,16 @@ uint32_t mds_tipc_fctrl_rcv_data(uint8_t *buffer,
uint16_t len,
m_MDS_LOG_ERR("FCTRL: Failed to send msg to mbx_events,
Error[%s]",
strerror(errno));
}
+ } else if (header.msg_type_ == Reset::kResetMsgType) {
+ // no need to decode reset message
+ // the decoding reset message type is done in header decoding
+ // send to the event thread
+ if (m_NCS_IPC_SEND(&mbx_events,
+ new Event(Event::Type::kEvtRcvReset, id, 0, 0, 0, 0),
+ NCS_IPC_PRIORITY_HIGH) != NCSCC_RC_SUCCESS) {
+ m_MDS_LOG_ERR("FCTRL: Failed to send msg to mbx_events,
Error[%s]",
+ strerror(errno));
+ }
} else {
m_MDS_LOG_ERR("FCTRL: [me] <-- [node:%x, ref:%u], "
"[msg_type:%u], Error[not supported message type]", @@
-516,6
+538,11 @@ uint32_t mds_tipc_fctrl_rcv_data(uint8_t *buffer, uint16_t
+len,
portid_map_mutex.unlock();
return rc;
}
+ } else {
+ m_MDS_LOG_DBG("FCTRL: [me] <-- [node:%x, ref:%u], "
+ "Receive non-flow-control data message, "
+ "header.pro_ver:%u",
+ id.node, id.ref, header.pro_ver_);
}
return NCSCC_RC_SUCCESS;
}
diff --git a/src/mds/mds_tipc_fctrl_msg.cc
b/src/mds/mds_tipc_fctrl_msg.cc index 932120f..4aba3fb 100644
--- a/src/mds/mds_tipc_fctrl_msg.cc
+++ b/src/mds/mds_tipc_fctrl_msg.cc
@@ -178,4 +178,15 @@ void Nack::Decode(uint8_t *msg) {
nacked_fseq_ = ncs_decode_16bit(&ptr); }
+
+void Reset::Encode(uint8_t *msg) {
+ uint8_t *ptr;
+ // encode protocol identifier
+ ptr = &msg[Reset::FieldIndex::kProtocolIdentifier];
+ ncs_encode_32bit(&ptr, MDS_PROT_FCTRL_ID);
+ // encode message type
+ ptr = &msg[Reset::FieldIndex::kFlowControlMessageType];
+ ncs_encode_8bit(&ptr, kResetMsgType); }
+
} // end namespace mds
diff --git a/src/mds/mds_tipc_fctrl_msg.h
b/src/mds/mds_tipc_fctrl_msg.h index e1db200..e94fb9d 100644
--- a/src/mds/mds_tipc_fctrl_msg.h
+++ b/src/mds/mds_tipc_fctrl_msg.h
@@ -45,6 +45,7 @@ class Event {
kEvtDropData, // event reported from tipc that a message is
not
// delivered
kEvtRcvNack, // event that received nack message
+ kEvtRcvReset, // event that received reset message
kEvtTmrAll,
kEvtTmrTxProb, // event that tx probation timer expired for once
kEvtTmrChunkAck, // event to send the chunk ack @@ -172,6
+173,23 @@ class Nack: public BaseMessage {
void Decode(uint8_t *msg) override;
};
+class Reset: public BaseMessage {
+ public:
+ enum FieldIndex {
+ kProtocolIdentifier = 11,
+ kFlowControlMessageType = 15,
+ };
+ static const uint8_t kResetMsgType = 3;
+ static const uint16_t kResetMsgLength = 16;
+
+ uint8_t msg_type_{0};
+
+ Reset() { msg_type_ = kResetMsgType; }
+ virtual ~Reset() {}
+ void Encode(uint8_t *msg) override; };
+
+
} // end namespace mds
#endif // MDS_MDS_TIPC_FCTRL_MSG_H_ diff --git
a/src/mds/mds_tipc_fctrl_portid.cc
b/src/mds/mds_tipc_fctrl_portid.cc
index 24a7e2a..f195b01 100644
--- a/src/mds/mds_tipc_fctrl_portid.cc
+++ b/src/mds/mds_tipc_fctrl_portid.cc
@@ -112,6 +112,7 @@ TipcPortId::TipcPortId(struct tipc_portid id, int
sock, uint16_t chksize,
uint64_t sock_buf_size):
id_(id), bsrsock_(sock), chunk_size_(chksize),
rcv_buf_size_(sock_buf_size) {
state_ = State::kStartup;
+ SendReset();
}
TipcPortId::~TipcPortId() {
@@ -189,7 +190,7 @@ uint32_t TipcPortId::Queue(const uint8_t* data,
uint16_t length,
sndwnd_.acked_.v(), sndwnd_.send_.v(), sndwnd_.nacked_space_);
} else {
++sndwnd_.send_;
- m_MDS_LOG_DBG("FCTRL: [me] --> [node:%x, ref:%u], "
+ m_MDS_LOG_NOTIFY("FCTRL: [me] --> [node:%x, ref:%u], "
"QueData[mseq:%u, mfrag:%u, fseq:%u, len:%u], "
"sndwnd[acked:%u, send:%u, nacked:%" PRIu64 "]",
id_.node, id_.ref,
@@ -248,10 +249,22 @@ void TipcPortId::SendNack(uint16_t fseq,
uint16_t
svc_id) {
Nack nack(svc_id, fseq);
nack.Encode(data);
Send(data, Nack::kNackMsgLength);
- m_MDS_LOG_DBG("FCTRL: [me] --> [node:%x, ref:%u], "
+ m_MDS_LOG_NOTIFY("FCTRL: [me] --> [node:%x, ref:%u], "
"SndNack[fseq:%u]", id_.node, id_.ref, fseq); }
+void TipcPortId::SendReset() {
+ uint8_t data[Reset::kResetMsgLength] = {0};
+
+ HeaderMessage header(Reset::kResetMsgLength, 0, 0, 0);
+ header.Encode(data);
+
+ Reset reset;
+ reset.Encode(data);
+ Send(data, Reset::kResetMsgLength);
+ m_MDS_LOG_NOTIFY("FCTRL: [me] --> [node:%x, ref:%u], "
+ "SndReset", id_.node, id_.ref); }
uint32_t TipcPortId::ReceiveData(uint32_t mseq, uint16_t mfrag,
uint16_t fseq, uint16_t svc_id) { @@ -330,8 +343,7 @@ uint32_t
TipcPortId::ReceiveData(uint32_t mseq, uint16_t mfrag,
// send nack
SendNack((rcvwnd_.rcv_ + Seq16(1)).v(), svc_id);
}
- }
- if (Seq16(fseq) <= rcvwnd_.rcv_) {
+ } else if (Seq16(fseq) <= rcvwnd_.rcv_) {
rc = NCSCC_RC_FAILURE;
// unexpected retransmission
m_MDS_LOG_ERR("FCTRL: [me] <-- [node:%x, ref:%u], "
@@ -399,7 +411,7 @@ void TipcPortId::ReceiveChunkAck(uint16_t fseq,
uint16_t
chksize) {
sndwnd_.nacked_space_ += msg->header_.msg_len_;
msg->is_sent_ = true;
resend_bytes += msg->header_.msg_len_;
- m_MDS_LOG_DBG("FCTRL: [me] --> [node:%x, ref:%u], "
+ m_MDS_LOG_NOTIFY("FCTRL: [me] --> [node:%x, ref:%u], "
"SndQData[fseq:%u, len:%u], "
"sndwnd[acked:%u, send:%u, nacked:%" PRIu64 "]",
id_.node, id_.ref,
@@ -443,7 +455,7 @@ void TipcPortId::ReceiveNack(uint32_t mseq,
uint16_t mfrag,
}
if (state_ == State::kRcvBuffOverflow) {
m_MDS_LOG_ERR("FCTRL: [me] <-- [node:%x, ref:%u], "
- "RcvNack[fseq:%u, state:%u]"
+ "RcvNack[fseq:%u, state:%u], "
"Warning[Ignore Nack]",
id_.node, id_.ref,
fseq, (uint8_t)state_);
@@ -462,7 +474,7 @@ void TipcPortId::ReceiveNack(uint32_t mseq,
uint16_t mfrag,
if (Send(msg->msg_data_, msg->header_.msg_len_) == NCSCC_RC_SUCCESS) {
msg->is_sent_ = true;
}
- m_MDS_LOG_DBG("FCTRL: [me] --> [node:%x, ref:%u], "
+ m_MDS_LOG_NOTIFY("FCTRL: [me] --> [node:%x, ref:%u], "
"RsndData[mseq:%u, mfrag:%u, fseq:%u], "
"sndwnd[acked:%u, send:%u, nacked:%" PRIu64 "]",
id_.node, id_.ref,
@@ -495,12 +507,11 @@ bool TipcPortId::ReceiveTmrTxProb(uint8_t
max_txprob) {
// receiver is at old mds version
if (state_ == State::kDisabled) {
FlushData();
+ m_MDS_LOG_NOTIFY("FCTRL: [node:%x, ref:%u], "
+ "TxProbExp, TxProb[retries:%u, state:%u]",
+ id_.node, id_.ref,
+ txprob_cnt_, (uint8_t)state_);
}
-
- m_MDS_LOG_NOTIFY("FCTRL: [node:%x, ref:%u], "
- "TxProbExp, TxProb[retries:%u, state:%u]",
- id_.node, id_.ref,
- txprob_cnt_, (uint8_t)state_);
}
return restart_txprob;
}
@@ -518,4 +529,18 @@ void TipcPortId::ReceiveTmrChunkAck() {
}
}
+void TipcPortId::ReceiveReset() {
+ m_MDS_LOG_NOTIFY("FCTRL: [me] <-- [node:%x, ref:%u], "
+ "RcvReset, "
+ "TxProb[retries:%u, state:%u]",
+ id_.node, id_.ref,
+ txprob_cnt_, (uint8_t)state_);
+ if (state_ == State::kStartup || state_ == State::kTxProb) {
+ state_ = State::kEnabled;
+ }
+ rcvwnd_.acked_ = 0;
+ rcvwnd_.rcv_ = 0;
+ rcvwnd_.acked_ = 0;
+}
+
} // end namespace mds
diff --git a/src/mds/mds_tipc_fctrl_portid.h
b/src/mds/mds_tipc_fctrl_portid.h index d238ac6..ffed2ad 100644
--- a/src/mds/mds_tipc_fctrl_portid.h
+++ b/src/mds/mds_tipc_fctrl_portid.h
@@ -134,11 +134,13 @@ class TipcPortId {
void ReceiveChunkAck(uint16_t fseq, uint16_t chunk_size);
void SendChunkAck(uint16_t fseq, uint16_t svc_id, uint16_t chunk_size);
void SendNack(uint16_t fseq, uint16_t svc_id);
+ void SendReset();
uint32_t ReceiveData(uint32_t mseq, uint16_t mfrag,
uint16_t fseq, uint16_t svc_id);
void ReceiveNack(uint32_t mseq, uint16_t mfrag, uint16_t fseq);
bool ReceiveTmrTxProb(uint8_t max_txprob);
void ReceiveTmrChunkAck();
+ void ReceiveReset();
void FlushData();
uint32_t Send(uint8_t* data, uint16_t length);
uint32_t Queue(const uint8_t* data, uint16_t length, bool
is_sent);
--
2.7.4
