Hi Minh,
I thought you will update check state of port id to know FCTRL or LEGACY. Since if (msg_len_ - fseq_ - 2 == MDTM_FRAG_HDR_LEN) may be not LEGACY protocol. Agree if (msg_len_ - fseq_ - 2 != MDTM_FRAG_HDR_LEN) 100% is FCTRL protocol. Best Regards, ThuanTr From: Minh Hon Chau <minh.c...@dektech.com.au> Sent: Thursday, November 14, 2019 4:28 PM To: Tran Thuan <thuan.t...@dektech.com.au>; hans.nordeb...@ericsson.com; gary....@dektech.com.au; vu.m.ngu...@dektech.com.au Cc: opensaf-devel@lists.sourceforge.net Subject: Re: [PATCH 1/3] mds: Distinguish protocol version of fragment [#3111] Hi Thuan, Are you happy with my reply? Thanks Minh On 14/11/19 9:35 am, Minh Hon Chau wrote: Hi Thuan, Please see my reply inline. Thanks Minh On 13/11/19 9:54 pm, Tran Thuan wrote: Hi Minh, See my comment inline. Best Regards, ThuanTr -----Original Message----- From: Minh Chau <mailto:minh.c...@dektech.com.au> <minh.c...@dektech.com.au> Sent: Friday, November 8, 2019 5:33 PM To: hans.nordeb...@ericsson.com <mailto:hans.nordeb...@ericsson.com> ; gary....@dektech.com.au <mailto:gary....@dektech.com.au> ; vu.m.ngu...@dektech.com.au <mailto:vu.m.ngu...@dektech.com.au> ; thuan.t...@dektech.com.au <mailto:thuan.t...@dektech.com.au> Cc: opensaf-devel@lists.sourceforge.net <mailto:opensaf-devel@lists.sourceforge.net> ; Minh Chau <mailto:minh.c...@dektech.com.au> <minh.c...@dektech.com.au> Subject: [PATCH 1/3] mds: Distinguish protocol version of fragment [#3111] The legacy mds encodes the protocol version in either non fragment message or the first fragment only. Hence, the subsequent fragment after the first one is not able for mds to determine the protocol version. The patch maintains the encoding of lengthcheck as same as the legacy mds version. Also, the subsequent fragments needs to consult the stateful portid to determine the protocol version, so that the fragment will be skipped if it is sent from legacy mds, or inspected the sequence if it is sent from new mds. --- src/mds/mds_dt.h | 6 ++ src/mds/mds_dt_tipc.c | 11 ++- src/mds/mds_tipc_fctrl_intf.cc | 154 ++++++++++++++++++++++----------------- src/mds/mds_tipc_fctrl_msg.cc | 86 +++++++++++++++------- src/mds/mds_tipc_fctrl_msg.h | 5 ++ src/mds/mds_tipc_fctrl_portid.cc | 23 ++++++ src/mds/mds_tipc_fctrl_portid.h | 1 + 7 files changed, 193 insertions(+), 93 deletions(-) diff --git a/src/mds/mds_dt.h b/src/mds/mds_dt.h index 64da600..007ff98 100644 --- a/src/mds/mds_dt.h +++ b/src/mds/mds_dt.h @@ -243,6 +243,12 @@ bool mdtm_mailbox_mbx_cleanup(NCSCONTEXT arg, NCSCONTEXT msg); #define MDS_PROT_VER_MASK 0xFC #define MDTM_PRI_MASK 0x3 +/* Unknown or undefined MDS protocol/version */ +#define MDS_PROT_UNDEFINED 0x00 + +/* MDS protocol/version for non flow control (legacy) */ +#define MDS_PROT_LEGACY (MDS_PROT | MDS_VERSION) + /* MDS protocol/version for flow control */ #define MDS_PROT_FCTRL (0xB0 | MDS_VERSION) #define MDS_PROT_FCTRL_ID 0xFDAC13F5 diff --git a/src/mds/mds_dt_tipc.c b/src/mds/mds_dt_tipc.c index e085de7..fdf0da7 100644 --- a/src/mds/mds_dt_tipc.c +++ b/src/mds/mds_dt_tipc.c @@ -166,7 +166,7 @@ NCS_PATRICIA_TREE mdtm_reassembly_list; uint32_t mdtm_global_frag_num; const unsigned int MAX_RECV_THRESHOLD = 30; -static uint8_t gl_mds_pro_ver = MDS_PROT | MDS_VERSION; +static uint8_t gl_mds_pro_ver = MDS_PROT_LEGACY; static int gl_mds_fctrl_acksize = -1; static int gl_mds_fctrl_ackto = -1; @@ -381,7 +381,7 @@ uint32_t mdtm_tipc_init(NODE_ID nodeid, uint32_t *mds_tipc_ref) "MDTM:TIPC Failed to unset MDS_TIPC_FCTRL_ACKSIZE"); } } else { - gl_mds_pro_ver = MDS_PROT | MDS_VERSION; + gl_mds_pro_ver = MDS_PROT_LEGACY; syslog(LOG_ERR, "MDTM:TIPC Invalid value of" "MDS_TIPC_FCTRL_ENABLED"); } @@ -3125,7 +3125,12 @@ uint32_t mdtm_add_frag_hdr(uint8_t *buf_ptr, uint16_t len, uint32_t seq_num, * hereafter, these 2 bytes will be used as sequence number in flow control * (per tipc portid) * */ - ncs_encode_16bit(&data, fctrl_seq_num); + if (gl_mds_pro_ver == MDS_PROT_FCTRL) { + ncs_encode_16bit(&data, fctrl_seq_num); + } else { + ncs_encode_16bit(&data, len - MDTM_FRAG_HDR_LEN - 2); + } + #endif return NCSCC_RC_SUCCESS; } diff --git a/src/mds/mds_tipc_fctrl_intf.cc b/src/mds/mds_tipc_fctrl_intf.cc index c9073b2..3d92290 100644 --- a/src/mds/mds_tipc_fctrl_intf.cc +++ b/src/mds/mds_tipc_fctrl_intf.cc @@ -132,8 +132,16 @@ uint32_t process_flow_event(const Event& evt) { portid = new TipcPortId(evt.id_, data_sock_fd, chunk_ack_size, sock_buf_size); portid_map[TipcPortId::GetUniqueId(evt.id_)] = portid; - rc = portid->ReceiveData(evt.mseq_, evt.mfrag_, - evt.fseq_, evt.svc_id_, evt.snd_type_, is_mcast_enabled); + if (evt.legacy_data_ == true) { + // we create portid and set state kDisabled even though we know + // this portid has no flow control. It is because the 2nd, 3rd fragment + // could not reflect the protocol version, so need to keep this portid + // remained stateful + portid->ChangeState(TipcPortId::State::kDisabled); + } else { + rc = portid->ReceiveData(evt.mseq_, evt.mfrag_, + evt.fseq_, evt.svc_id_, evt.snd_type_, is_mcast_enabled); + } } else if (evt.type_ == Event::Type::kEvtRcvIntro) { portid = new TipcPortId(evt.id_, data_sock_fd, chunk_ack_size, sock_buf_size); @@ -146,8 +154,12 @@ uint32_t process_flow_event(const Event& evt) { } } else { if (evt.type_ == Event::Type::kEvtRcvData) { - rc = portid->ReceiveData(evt.mseq_, evt.mfrag_, - evt.fseq_, evt.svc_id_, evt.snd_type_, is_mcast_enabled); + if (evt.legacy_data_ == true) { + portid->ChangeState(TipcPortId::State::kDisabled); + } else { + rc = portid->ReceiveData(evt.mseq_, evt.mfrag_, + evt.fseq_, evt.svc_id_, evt.snd_type_, is_mcast_enabled); + } } if (evt.type_ == Event::Type::kEvtRcvChunkAck) { portid->ReceiveChunkAck(evt.fseq_, evt.chunk_size_); @@ -474,76 +486,88 @@ uint32_t mds_tipc_fctrl_rcv_data(uint8_t *buffer, uint16_t len, struct tipc_portid id) { if (is_fctrl_enabled == false) return NCSCC_RC_SUCCESS; + uint32_t rc = NCSCC_RC_SUCCESS; HeaderMessage header; header.Decode(buffer); Event* pevt = nullptr; // if mds support flow control - if ((header.pro_ver_ & MDS_PROT_VER_MASK) == MDS_PROT_FCTRL) { - if (header.pro_id_ == MDS_PROT_FCTRL_ID) { - if (header.msg_type_ == ChunkAck::kChunkAckMsgType) { - // receive single ack message - ChunkAck ack; - ack.Decode(buffer); - // send to the event thread - pevt = new Event(Event::Type::kEvtRcvChunkAck, id, ack.svc_id_, - header.mseq_, header.mfrag_, ack.acked_fseq_); - pevt->chunk_size_ = ack.chunk_size_; - if (m_NCS_IPC_SEND(&mbx_events, pevt, - NCS_IPC_PRIORITY_HIGH) != NCSCC_RC_SUCCESS) { - m_MDS_LOG_ERR("FCTRL: Failed to send msg to mbx_events, Error[%s]", - strerror(errno)); - } - } else if (header.msg_type_ == Nack::kNackMsgType) { - // receive nack message - Nack nack; - nack.Decode(buffer); - // send to the event thread - pevt = new Event(Event::Type::kEvtRcvNack, id, nack.svc_id_, - header.mseq_, header.mfrag_, nack.nacked_fseq_); - if (m_NCS_IPC_SEND(&mbx_events, pevt, - NCS_IPC_PRIORITY_HIGH) != NCSCC_RC_SUCCESS) { - m_MDS_LOG_ERR("FCTRL: Failed to send msg to mbx_events, Error[%s]", - strerror(errno)); - } - } else if (header.msg_type_ == Intro::kIntroMsgType) { - // no need to decode intro message - // the decoding intro message type is done in header decoding - // send to the event thread - pevt = new Event(Event::Type::kEvtRcvIntro, id, 0, 0, 0, 0); - if (m_NCS_IPC_SEND(&mbx_events, pevt, - 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]", - id.node, id.ref, header.msg_type_); + if (header.IsControlMessage()) { + if (header.msg_type_ == ChunkAck::kChunkAckMsgType) { + // receive single ack message + ChunkAck ack; + ack.Decode(buffer); + // send to the event thread + pevt = new Event(Event::Type::kEvtRcvChunkAck, id, ack.svc_id_, + header.mseq_, header.mfrag_, ack.acked_fseq_); + pevt->chunk_size_ = ack.chunk_size_; + if (m_NCS_IPC_SEND(&mbx_events, pevt, + NCS_IPC_PRIORITY_HIGH) != NCSCC_RC_SUCCESS) { + m_MDS_LOG_ERR("FCTRL: Failed to send msg to mbx_events, Error[%s]", + strerror(errno)); } - // return NCSCC_RC_FAILURE, so the tipc receiving thread (legacy) will - // skip this data msg - return NCSCC_RC_FAILURE; - } else { - // receive data message - DataMessage data; - data.Decode(buffer); - portid_map_mutex.lock(); - Event evt(Event::Type::kEvtRcvData, id, data.svc_id_, header.mseq_, - header.mfrag_, header.fseq_); - evt.snd_type_ = data.snd_type_; - uint32_t rc = process_flow_event(evt); - if (rc == NCSCC_RC_CONTINUE) { - process_timer_event(Event(Event::Type::kEvtTmrChunkAck)); - rc = NCSCC_RC_SUCCESS; + } else if (header.msg_type_ == Nack::kNackMsgType) { + // receive nack message + Nack nack; + nack.Decode(buffer); + // send to the event thread + pevt = new Event(Event::Type::kEvtRcvNack, id, nack.svc_id_, + header.mseq_, header.mfrag_, nack.nacked_fseq_); + if (m_NCS_IPC_SEND(&mbx_events, pevt, + NCS_IPC_PRIORITY_HIGH) != NCSCC_RC_SUCCESS) { + m_MDS_LOG_ERR("FCTRL: Failed to send msg to mbx_events, Error[%s]", + strerror(errno)); } - portid_map_mutex.unlock(); - return rc; + } else if (header.msg_type_ == Intro::kIntroMsgType) { + // no need to decode intro message + // the decoding intro message type is done in header decoding + // send to the event thread + pevt = new Event(Event::Type::kEvtRcvIntro, id, 0, 0, 0, 0); + if (m_NCS_IPC_SEND(&mbx_events, pevt, + 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]", + id.node, id.ref, header.msg_type_); } - } else { + // return NCSCC_RC_FAILURE, so the tipc receiving thread (legacy) will + // skip this data msg + rc = NCSCC_RC_FAILURE; + } + if (header.IsLegacyMessage()) { m_MDS_LOG_DBG("FCTRL: [me] <-- [node:%x, ref:%u], " - "Receive non-flow-control data message, " + "Receive legacy data message, " "header.pro_ver:%u", id.node, id.ref, header.pro_ver_); + // Receiving the first fragment legacy message, change the state to + // kDisabled so the subsequent fragment will skip the flow control + if ((header.mfrag_ & 0x7fff) == 1) { + portid_map_mutex.lock(); + Event evt(Event::Type::kEvtRcvData, id, 0, 0, 0, 0); + evt.legacy_data_ = true; + process_flow_event(evt); + portid_map_mutex.unlock(); + } } - return NCSCC_RC_SUCCESS; + if (header.IsFlowMessage() || header.IsUndefinedMessage()) { + // receive flow message explicitly identified by protocol version + // or if it is still undefined, then consult the stateful portid + // to proceed. + DataMessage data; + data.Decode(buffer); + portid_map_mutex.lock(); + Event evt(Event::Type::kEvtRcvData, id, data.svc_id_, header.mseq_, + header.mfrag_, header.fseq_); + evt.snd_type_ = data.snd_type_; + rc = process_flow_event(evt); + if (rc == NCSCC_RC_CONTINUE) { + process_timer_event(Event(Event::Type::kEvtTmrChunkAck)); + rc = NCSCC_RC_SUCCESS; + } + portid_map_mutex.unlock(); + } + + return rc; } diff --git a/src/mds/mds_tipc_fctrl_msg.cc b/src/mds/mds_tipc_fctrl_msg.cc index 180dcb6..454c02c 100644 --- a/src/mds/mds_tipc_fctrl_msg.cc +++ b/src/mds/mds_tipc_fctrl_msg.cc @@ -22,7 +22,7 @@ namespace mds { HeaderMessage::HeaderMessage(uint16_t msg_len, uint32_t mseq, uint16_t mfrag, uint16_t fseq): msg_len_(msg_len), mseq_(mseq), mfrag_(mfrag), fseq_(fseq) { - pro_ver_ = MDS_PROT_FCTRL; + pro_ver_ = MDS_PROT_UNDEFINED; pro_id_ = 0; msg_type_ = 0; } @@ -47,7 +47,9 @@ void HeaderMessage::Encode(uint8_t *msg) { void HeaderMessage::Decode(uint8_t *msg) { uint8_t *ptr; - + pro_id_ = 0; + msg_type_ = 0; + pro_ver_ = MDS_PROT_UNDEFINED; // decode message length ptr = &msg[HeaderMessage::FieldIndex::kMessageLength]; msg_len_ = ncs_decode_16bit(&ptr); @@ -57,36 +59,70 @@ void HeaderMessage::Decode(uint8_t *msg) { // decode fragment number ptr = &msg[HeaderMessage::FieldIndex::kFragmentNumber]; mfrag_ = ncs_decode_16bit(&ptr); - // decode protocol version - ptr = &msg[HeaderMessage::FieldIndex::kProtocolVersion]; - pro_ver_ = ncs_decode_8bit(&ptr); - if ((pro_ver_ & MDS_PROT_VER_MASK) == MDS_PROT_FCTRL) { - // decode flow control sequence number - ptr = &msg[HeaderMessage::FieldIndex::kFlowControlSequenceNumber]; - fseq_ = ncs_decode_16bit(&ptr); - // decode protocol identifier if the mfrag_ and mseq_ are 0 - // otherwise it is always DataMessage within non-zero mseq_ and mfrag_ - if (mfrag_ == 0 && mseq_ == 0) { - ptr = &msg[ChunkAck::FieldIndex::kProtocolIdentifier]; - pro_id_ = ncs_decode_32bit(&ptr); - if (pro_id_ == MDS_PROT_FCTRL_ID) { - // decode message type - ptr = &msg[ChunkAck::FieldIndex::kFlowControlMessageType]; - msg_type_ = ncs_decode_8bit(&ptr); + // The unfragment or 1st fragment always has encoded the protocol version + // at oct10, refer to mdtm_add_mds_hdr(). Therefore we can rely on oct10 + // to identify the flow control message + if (mfrag_ == 0 || ((mfrag_ & 0x7fff) == 1)) { + // decode protocol version + ptr = &msg[HeaderMessage::FieldIndex::kProtocolVersion]; + pro_ver_ = ncs_decode_8bit(&ptr); + if ((pro_ver_ & MDS_PROT_VER_MASK) == MDS_PROT_FCTRL) { + // decode flow control sequence number + ptr = &msg[HeaderMessage::FieldIndex::kFlowControlSequenceNumber]; + fseq_ = ncs_decode_16bit(&ptr); + // decode protocol identifier if the mfrag_ and mseq_ are 0 + // otherwise it is always DataMessage within non-zero mseq_ and mfrag_ + if (mfrag_ == 0 && mseq_ == 0) { + ptr = &msg[ChunkAck::FieldIndex::kProtocolIdentifier]; + pro_id_ = ncs_decode_32bit(&ptr); + if (pro_id_ == MDS_PROT_FCTRL_ID) { + // decode message type + ptr = &msg[ChunkAck::FieldIndex::kFlowControlMessageType]; + msg_type_ = ncs_decode_8bit(&ptr); + } } - } else { - pro_id_ = 0; - msg_type_ = 0; } } else { - if (mfrag_ != 0) { - ptr = &msg[HeaderMessage::FieldIndex::kFlowControlSequenceNumber]; - fseq_ = ncs_decode_16bit(&ptr); - if (fseq_ != 0) pro_ver_ = MDS_PROT_FCTRL; + // this is the other fragment after 1st fragment (i.e. 2nd, 3rd, etc...) + // the oct10 is written by fragment data, it is not encoded for + // protocol version anymore. + // In legacy protocol version, lengthcheck(oct8) = messagelength(oct0) - 10 + // Thus, if fseq(oct8) != messagelength(oct0) - 10, this is MDS_PROT_FCTRL + // Otherwise, it could be either MDS_PROT_FCTRL or MDS_PROT_LEGACY + // so set the pro_ver_ as MDS_PROT_UNDEFINED and refer to the portidmap + // to decide the protocol version + ptr = &msg[HeaderMessage::FieldIndex::kFlowControlSequenceNumber]; + fseq_ = ncs_decode_16bit(&ptr); + if (msg_len_ - fseq_ - 2 != MDTM_FRAG_HDR_LEN) { + pro_ver_ = MDS_PROT_FCTRL; [Thuan] I think this check is uncertain, can be coincident true/false depend on msg_len and fseq. How about use state of portid to decide LEGACY or FCTRL? Since this is 2nd fragmented msg (state of portid is ready to use) [Minh] The check looks cryptic so I added the code comment above the lines. This is where the legacy code encodes the oct8 and oct9 ncs_encode_16bit(&data, len - MDTM_FRAG_HDR_LEN - 2); In legacy code, the lengthcheck at oct8&oct9 is always equal to the (length - 10). So if you see the lengthcheck is not equal to the (length - 10), it is 100% used for flow sequence number, case [1]. However, if oct8&oct9 is equal to (length - 10), we can not conclude it is the legacy lengthcheck. Because the sequence number coincidentally (as you said) can increase to a value being equal to (length - 10), and this is where we need to consult the port id state. The other case [1], we don't need to use the portid state, the values can tell. } } } +bool HeaderMessage::IsControlMessage() { + if (((pro_ver_ & MDS_PROT_VER_MASK) == MDS_PROT_FCTRL) && + pro_id_ == MDS_PROT_FCTRL_ID) { + return true; + } + return false; +} + +bool HeaderMessage::IsLegacyMessage() { + return (pro_ver_ & MDS_PROT_VER_MASK) == MDS_PROT_LEGACY; +} + +bool HeaderMessage::IsFlowMessage() { + if (((pro_ver_ & MDS_PROT_VER_MASK) == MDS_PROT_FCTRL) && + pro_id_ != MDS_PROT_FCTRL_ID) { + return true; + } + return false; +} + +bool HeaderMessage::IsUndefinedMessage() { + return (pro_ver_ & MDS_PROT_VER_MASK) == MDS_PROT_UNDEFINED; +} + void DataMessage::Decode(uint8_t *msg) { uint8_t *ptr; diff --git a/src/mds/mds_tipc_fctrl_msg.h b/src/mds/mds_tipc_fctrl_msg.h index 1964c0f..c4641ed 100644 --- a/src/mds/mds_tipc_fctrl_msg.h +++ b/src/mds/mds_tipc_fctrl_msg.h @@ -61,6 +61,7 @@ class Event { uint16_t fseq_{0}; uint16_t chunk_size_{1}; uint8_t snd_type_{0}; + bool legacy_data_{false}; explicit Event(Type type):type_(type) {} Event(Type type, struct tipc_portid id, uint16_t svc_id, uint32_t mseq, uint16_t mfrag, uint16_t f_seq_num): @@ -105,6 +106,10 @@ class HeaderMessage: public BaseMessage { virtual ~HeaderMessage() {} void Decode(uint8_t *msg) override; void Encode(uint8_t *msg) override; + bool IsControlMessage(); + bool IsLegacyMessage(); + bool IsFlowMessage(); + bool IsUndefinedMessage(); }; class DataMessage: public BaseMessage { diff --git a/src/mds/mds_tipc_fctrl_portid.cc b/src/mds/mds_tipc_fctrl_portid.cc index 3704bad..9b87c74 100644 --- a/src/mds/mds_tipc_fctrl_portid.cc +++ b/src/mds/mds_tipc_fctrl_portid.cc @@ -583,6 +583,29 @@ void TipcPortId::ReceiveTmrChunkAck() { } } +void TipcPortId::ChangeState(State newState) { + if (state_ == newState) return; + + if (newState == State::kDisabled) { + // at kDisabled state, clear all message in sndqueue_, + // receiver is at old mds version + FlushData(); + } + m_MDS_LOG_NOTIFY("FCTRL: [node:%x, ref:%u], " + "ChangeState[%u -> %u], " + "TxProb[retries:%u], " + "sndwnd[acked:%u, send:%u, nacked:%" PRIu64 "], " + "rcvwnd[acked:%u, rcv:%u, nacked:%" PRIu64 "], " + "queue[size:%" PRIu64 "]", + id_.node, id_.ref, + (uint8_t)state_, (uint8_t)newState, + txprob_cnt_, + sndwnd_.acked_.v(), sndwnd_.send_.v(), sndwnd_.nacked_space_, + rcvwnd_.acked_.v(), rcvwnd_.rcv_.v(), rcvwnd_.nacked_space_, + sndqueue_.Size()); + state_ = newState; +} + void TipcPortId::ReceiveIntro() { m_MDS_LOG_NOTIFY("FCTRL: [me] <-- [node:%x, ref:%u], " "RcvIntro, " diff --git a/src/mds/mds_tipc_fctrl_portid.h b/src/mds/mds_tipc_fctrl_portid.h index 24fb195..52e0780 100644 --- a/src/mds/mds_tipc_fctrl_portid.h +++ b/src/mds/mds_tipc_fctrl_portid.h @@ -141,6 +141,7 @@ class TipcPortId { bool ReceiveTmrTxProb(uint8_t max_txprob); void ReceiveTmrChunkAck(); void ReceiveIntro(); + void ChangeState(State newState); void FlushData(); uint32_t Send(uint8_t* data, uint16_t length); uint32_t Queue(const uint8_t* data, uint16_t length, bool is_sent); _______________________________________________ Opensaf-devel mailing list Opensaf-devel@lists.sourceforge.net https://lists.sourceforge.net/lists/listinfo/opensaf-devel
