[devel] [PATCH 0/1] Review Request for amfd: correct handling complete/apply callback on standby sc [#3082]
Summary: amfd: correct handling complete/apply callback on standby sc [#3082] Review request for Ticket(s): 3082 Peer Reviewer(s): Gary, Minh, Thuan Pull request to: Gary Affected branch(es): develop Development branch: ticket-3082 Base revision: eac09bf540f33cf25ee2a8902aeeef40ee6bbf3b Personal repository: git://git.code.sf.net/u/thangng/review Impacted area Impact y/n Docsn Build systemn RPM/packaging n Configuration files n Startup scripts n SAF servicesy OpenSAF servicesn Core libraries n Samples n Tests n Other n Comments (indicate scope for each "y" above): - revision a4256464994a8a610dff25ac3ad97d4598b638ee Author: thang.d.nguyen Date: Mon, 16 Sep 2019 13:18:23 +0700 amfd: correct handling complete/apply callback on standby sc [#3082] During stanby SC comes up, AMF config objects are deleted on active SC. It causes NOT_EXIST error on standby node. AMFD on standby should ignore this error in this case. Complete diffstat: -- src/amf/amfd/app.cc| 29 - src/amf/amfd/comp.cc | 18 +++--- src/amf/amfd/compcstype.cc | 14 ++ src/amf/amfd/csi.cc| 24 ++-- src/amf/amfd/nodegroup.cc | 7 --- src/amf/amfd/sg.cc | 32 ++-- src/amf/amfd/sgtype.cc | 11 +++ src/amf/amfd/si.cc | 29 ++--- src/amf/amfd/su.cc | 35 --- src/amf/amfd/sutype.cc | 12 10 files changed, 162 insertions(+), 49 deletions(-) Testing Commands: - N/A Testing, Expected Results: -- N/A Conditions of Submission: - Acked from reviwer Arch Built StartedLinux distro --- mipsn n mips64 n n x86 n n x86_64 y y powerpc n n powerpc64 n n Reviewer Checklist: --- [Submitters: make sure that your review doesn't trigger any checkmarks!] Your checkin has not passed review because (see checked entries): ___ Your RR template is generally incomplete; it has too many blank entries that need proper data filled in. ___ You have failed to nominate the proper persons for review and push. ___ Your patches do not have proper short+long header ___ You have grammar/spelling in your header that is unacceptable. ___ You have exceeded a sensible line length in your headers/comments/text. ___ You have failed to put in a proper Trac Ticket # into your commits. ___ You have incorrectly put/left internal data in your comments/files (i.e. internal bug tracking tool IDs, product names etc) ___ You have not given any evidence of testing beyond basic build tests. Demonstrate some level of runtime or other sanity testing. ___ You have ^M present in some of your files. These have to be removed. ___ You have needlessly changed whitespace or added whitespace crimes like trailing spaces, or spaces before tabs. ___ You have mixed real technical changes with whitespace and other cosmetic code cleanup changes. These have to be separate commits. ___ You need to refactor your submission into logical chunks; there is too much content into a single commit. ___ You have extraneous garbage in your review (merge commits etc) ___ You have giant attachments which should never have been sent; Instead you should place your content in a public tree to be pulled. ___ You have too many commits attached to an e-mail; resend as threaded commits, or place in a public tree for a pull. ___ You have resent this content multiple times without a clear indication of what has changed between each re-send. ___ You have failed to adequately and individually address all of the comments and change requests that were proposed in the initial review. ___ You have a misconfigured ~/.gitconfig file (i.e. user.name, user.email etc) ___ Your computer have a badly configured date and time; confusing the the threaded patch review. ___ Your changes affect IPC mechanism, and you don't present any results for in-service upgradability test. ___ Your changes affect user manual and documentation, your patch series do not contain the patch that updates the Doxygen manual. ___ Opensaf-devel mailing list Opensaf-devel@lists.sourceforge.net https://lists.sourceforge.net/lists/listinfo/opensaf-devel
[devel] [PATCH 1/1] amfd: correct handling complete/apply callback on standby sc [#3082]
During stanby SC comes up, AMF config objects are deleted on
active SC. It causes NOT_EXIST error on standby node.
AMFD on standby should ignore this error in this case.
---
src/amf/amfd/app.cc| 29 -
src/amf/amfd/comp.cc | 18 +++---
src/amf/amfd/compcstype.cc | 14 ++
src/amf/amfd/csi.cc| 24 ++--
src/amf/amfd/nodegroup.cc | 7 ---
src/amf/amfd/sg.cc | 32 ++--
src/amf/amfd/sgtype.cc | 11 +++
src/amf/amfd/si.cc | 29 ++---
src/amf/amfd/su.cc | 35 ---
src/amf/amfd/sutype.cc | 12
10 files changed, 162 insertions(+), 49 deletions(-)
diff --git a/src/amf/amfd/app.cc b/src/amf/amfd/app.cc
index 67e5e3e9d..17a259199 100644
--- a/src/amf/amfd/app.cc
+++ b/src/amf/amfd/app.cc
@@ -296,6 +296,11 @@ static void app_ccb_apply_cb(CcbUtilOperationData_t
*opdata) {
case CCBUTIL_MODIFY: {
const SaImmAttrModificationT_2 *attr_mod;
app = app_db->find(Amf::to_string(&opdata->objectName));
+ if (app == nullptr && avd_cb->is_active() == false) {
+LOG_WA("App modify apply (STDBY): app does not exist");
+break;
+ }
+ assert(app != nullptr);
int i = 0;
while ((attr_mod = opdata->param.modify.attrMods[i++]) != nullptr) {
const SaImmAttrValuesT_2 *attribute = &attr_mod->modAttr;
@@ -448,11 +453,12 @@ SaAisErrorT avd_app_config_get(void) {
searchParam.searchOneAttr.attrValueType = SA_IMM_ATTR_SASTRINGT;
searchParam.searchOneAttr.attrValue = &className;
- if (immutil_saImmOmSearchInitialize_2(
+ if ((rc = immutil_saImmOmSearchInitialize_2(
avd_cb->immOmHandle, nullptr, SA_IMM_SUBTREE,
SA_IMM_SEARCH_ONE_ATTR | SA_IMM_SEARCH_GET_SOME_ATTR, &searchParam,
- configAttributes, &searchHandle) != SA_AIS_OK) {
-LOG_ER("%s: saImmOmSearchInitialize_2 failed: %u", __FUNCTION__, error);
+ configAttributes, &searchHandle)) != SA_AIS_OK) {
+LOG_ER("%s: saImmOmSearchInitialize_2 failed: %u", __FUNCTION__, rc);
+error = rc;
goto done1;
}
@@ -468,9 +474,22 @@ SaAisErrorT avd_app_config_get(void) {
app_add_to_model(app);
-if (avd_sg_config_get(Amf::to_string(&dn), app) != SA_AIS_OK) goto done2;
+if ((rc = avd_sg_config_get(Amf::to_string(&dn), app)) != SA_AIS_OK) {
+ if ((rc == SA_AIS_ERR_NOT_EXIST) && (avd_cb->is_active() == false)) {
+avd_app_delete(app);
+continue;
+ } else {
+goto done2;
+ }
+}
-if (avd_si_config_get(app) != SA_AIS_OK) goto done2;
+if ((rc = avd_si_config_get(app)) != SA_AIS_OK) {
+ if ((rc == SA_AIS_ERR_NOT_EXIST) && (avd_cb->is_active() == false)) {
+avd_app_delete(app);
+ } else {
+goto done2;
+ }
+}
}
if (rc == SA_AIS_ERR_NOT_EXIST) {
diff --git a/src/amf/amfd/comp.cc b/src/amf/amfd/comp.cc
index 0ff365e55..7e46584db 100644
--- a/src/amf/amfd/comp.cc
+++ b/src/amf/amfd/comp.cc
@@ -1509,6 +1509,7 @@ SaAisErrorT avd_comp_config_get(const std::string
&su_name, AVD_SU *su) {
SA_IMM_SEARCH_ONE_ATTR | SA_IMM_SEARCH_GET_SOME_ATTR, &searchParam,
configAttributes, &searchHandle)) != SA_AIS_OK) {
LOG_ER("%s: saImmOmSearchInitialize_2 failed: %u", __FUNCTION__, rc);
+error = rc;
goto done1;
}
@@ -1524,9 +1525,15 @@ SaAisErrorT avd_comp_config_get(const std::string
&su_name, AVD_SU *su) {
num_of_comp_in_su++;
comp_add_to_model(comp);
-if (avd_compcstype_config_get(Amf::to_string(&comp_name), comp) !=
-SA_AIS_OK)
- goto done2;
+if ((rc = avd_compcstype_config_get(Amf::to_string(&comp_name), comp)) !=
+SA_AIS_OK) {
+ if ((rc == SA_AIS_ERR_NOT_EXIST) && (avd_cb->is_active() == false)) {
+avd_comp_delete(comp);
+num_of_comp_in_su--;
+ } else {
+goto done2;
+ }
+}
}
/* If there are no component in the SU, we treat it as invalid configuration.
@@ -1695,6 +1702,10 @@ static SaAisErrorT
ccb_completed_modify_hdlr(CcbUtilOperationData_t *opdata) {
TRACE_ENTER();
comp = comp_db->find(Amf::to_string(&opdata->objectName));
+ if (comp == nullptr && avd_cb->is_active() == false) {
+LOG_WA("Comp modify completed (STDBY): comp does not exist");
+return SA_AIS_OK;
+ }
while ((attr_mod = opdata->param.modify.attrMods[i++]) != nullptr) {
const SaImmAttrValuesT_2 *attribute = &attr_mod->modAttr;
@@ -2479,6 +2490,7 @@ void comp_ccb_apply_delete_hdlr(struct
CcbUtilOperationData *opdata) {
AVD_COMP *comp = comp_db->find(Amf::to_string(&opdata->objectName));
if (comp == nullptr && avd_cb->is_active() == false) {
+LOG_WA("Comp modify apply (STDBY): comp does not exist");
return;
}
/* comp should be found in the database even if it was
diff --git a/src/amf/amfd/compcstype.cc b/src/amf/a
Re: [devel] [PATCH 3/9] mds: Add implementation for TIPC buffer overflow solution [#1960]
Hi Vu,
I see it, will add.
Thanks
Minh
On 16/9/19 4:21 pm, Nguyen Minh Vu wrote:
Hi Minh,
See my responses to your comments below, started with [Vu2].
Regards, Vu
On 9/16/19 1:06 PM, Minh Hon Chau wrote:
Hi Vu,
Several comments with [M] too :).
Thanks
Minh
On 16/9/19 2:24 pm, Nguyen Minh Vu wrote:
Hi Minh,
I have several comments below, started with [Vu].
Regards, Vu
On 8/14/19 1:01 PM, Minh Chau wrote:
This is a collaborative patch of two participants:
- Tran Thuan
- Minh Chau
Main changes:
- Add mds_tipc_fctrl_intf.h, mds_tipc_fctrl_intf.cc: These two files
introduce new functions which are called in mds_dt_tipc.c if the flow
control is enabled
- Add mds_tipc_fctrl_portid.h, mds_tipc_fctrl_portid.cc: These files
implements the tipc portid instance, which supports the sliding
window,
mds msg queue
- Add mds_tipc_fctrl_msg.h, mds_tipc_fctrl_msg.cc: These files define
the event and messages which are used for this solution.
---
src/mds/Makefile.am | 10 +-
src/mds/mds_dt.h | 8 +-
src/mds/mds_dt_tipc.c | 188 +---
src/mds/mds_tipc_fctrl_intf.cc | 376
+++
src/mds/mds_tipc_fctrl_intf.h | 47 +
src/mds/mds_tipc_fctrl_msg.cc | 142 +++
src/mds/mds_tipc_fctrl_msg.h | 129 ++
src/mds/mds_tipc_fctrl_portid.cc | 261 +++
src/mds/mds_tipc_fctrl_portid.h | 87 +
9 files changed, 1184 insertions(+), 64 deletions(-)
create mode 100644 src/mds/mds_tipc_fctrl_intf.cc
create mode 100644 src/mds/mds_tipc_fctrl_intf.h
create mode 100644 src/mds/mds_tipc_fctrl_msg.cc
create mode 100644 src/mds/mds_tipc_fctrl_msg.h
create mode 100644 src/mds/mds_tipc_fctrl_portid.cc
create mode 100644 src/mds/mds_tipc_fctrl_portid.h
diff --git a/src/mds/Makefile.am b/src/mds/Makefile.am
index 2d7b652..d849e8f 100644
--- a/src/mds/Makefile.am
+++ b/src/mds/Makefile.am
@@ -48,10 +48,16 @@ lib_libopensaf_core_la_SOURCES += \
if ENABLE_TIPC_TRANSPORT
noinst_HEADERS += src/mds/mds_dt_tipc.h \
src/mds/mds_tipc_recvq_stats.h \
- src/mds/mds_tipc_recvq_stats_impl.h
+ src/mds/mds_tipc_recvq_stats_impl.h \
+ src/mds/mds_tipc_fctrl_intf.h \
+ src/mds/mds_tipc_fctrl_portid.h \
+ src/mds/mds_tipc_fctrl_msg.h
lib_libopensaf_core_la_SOURCES += src/mds/mds_dt_tipc.c \
src/mds/mds_tipc_recvq_stats.cc \
- src/mds/mds_tipc_recvq_stats_impl.cc
+ src/mds/mds_tipc_recvq_stats_impl.cc \
+ src/mds/mds_tipc_fctrl_intf.cc \
+ src/mds/mds_tipc_fctrl_portid.cc \
+ src/mds/mds_tipc_fctrl_msg.cc
endif
if ENABLE_TESTS
diff --git a/src/mds/mds_dt.h b/src/mds/mds_dt.h
index b645bb4..d9e8633 100644
--- a/src/mds/mds_dt.h
+++ b/src/mds/mds_dt.h
@@ -162,7 +162,7 @@ uint32_t
mdtm_del_from_ref_tbl(MDS_SUBTN_REF_VAL ref);
uint32_t mds_tmr_mailbox_processing(void);
uint32_t mdtm_get_from_ref_tbl(MDS_SUBTN_REF_VAL ref, MDS_SVC_HDL
*svc_hdl);
uint32_t mdtm_add_frag_hdr(uint8_t *buf_ptr, uint16_t len,
uint32_t seq_num,
- uint16_t frag_byte);
+ uint16_t frag_byte, uint16_t
fctrl_seq_num);
uint32_t mdtm_free_reassem_msg_mem(MDS_ENCODED_MSG *msg);
uint32_t mdtm_process_recv_data(uint8_t *buf, uint16_t len,
uint64_t tipc_id,
uint32_t *buff_dump);
@@ -240,9 +240,13 @@ bool mdtm_mailbox_mbx_cleanup(NCSCONTEXT arg,
NCSCONTEXT msg);
#define MDS_PROT 0xA0
#define MDS_VERSION 0x08
-#define MDS_PROT_VER_MASK (MDS_PROT | MDS_VERSION)
+#define MDS_PROT_VER_MASK 0xFC
#define MDTM_PRI_MASK 0x3
+/* MDS protocol/version for flow control */
+#define MDS_PROT_FCTRL (0xB0 | MDS_VERSION)
+#define MDS_PROT_FCTRL_ID 0x00AC13F5
+
/* Added for the subscription changes */
#define MDS_NCS_CHASSIS_ID (m_NCS_GET_NODE_ID & 0x00ff)
#define MDS_TIPC_COMMON_ID 0x01001000
diff --git a/src/mds/mds_dt_tipc.c b/src/mds/mds_dt_tipc.c
index 86b52bb..fef1c50 100644
--- a/src/mds/mds_dt_tipc.c
+++ b/src/mds/mds_dt_tipc.c
@@ -47,6 +47,7 @@
#include "mds_dt_tipc.h"
#include "mds_dt_tcp_disc.h"
#include "mds_core.h"
+#include "mds_tipc_fctrl_intf.h"
#include "mds_tipc_recvq_stats.h"
#include "base/osaf_utility.h"
#include "base/osaf_poll.h"
@@ -165,20 +166,22 @@ NCS_PATRICIA_TREE mdtm_reassembly_list;
uint32_t mdtm_global_frag_num;
const unsigned int MAX_RECV_THRESHOLD = 30;
+uint8_t gl_mds_pro_ver = MDS_PROT | MDS_VERSION;
-static bool get_tipc_port_id(int sock, uint32_t* port_id) {
+static bool get_tipc_port_id(int sock, struct tipc_portid* port_id) {
struct sockaddr_tipc addr;
socklen_t sz = sizeof(addr);
memset(&addr, 0, sizeof(addr));
- *port_id = 0;
+ port_id->node = 0;
+ port_id->ref = 0;
if (0 > getsockname(sock, (struct sockaddr *)&addr, &sz)) {
syslog(LOG_ERR, "MDTM:TIPC Failed to get socket name,
err: %s",
strerror(errno));
return
Re: [devel] [PATCH 5/9] mds: Add state machine for tipc portid instance [#1960]
Hi Vu,
Some comments with [M]
Thanks
Minh
On 16/9/19 2:56 pm, Nguyen Minh Vu wrote:
Hi Minh,
I has few comments below.
Regards, Vu
On 8/14/19 1:38 PM, Minh Chau wrote:
This patch adds state machine to support tx probation timer.
---
src/mds/mds_tipc_fctrl_intf.cc | 47 +++--
src/mds/mds_tipc_fctrl_msg.h | 1 +
src/mds/mds_tipc_fctrl_portid.cc | 109
+++
src/mds/mds_tipc_fctrl_portid.h | 22
4 files changed, 176 insertions(+), 3 deletions(-)
diff --git a/src/mds/mds_tipc_fctrl_intf.cc
b/src/mds/mds_tipc_fctrl_intf.cc
index bd0a8f6..c2d0922 100644
--- a/src/mds/mds_tipc_fctrl_intf.cc
+++ b/src/mds/mds_tipc_fctrl_intf.cc
@@ -34,6 +34,7 @@
using mds::Event;
using mds::TipcPortId;
+using mds::Timer;
using mds::DataMessage;
using mds::ChunkAck;
using mds::HeaderMessage;
@@ -65,6 +66,11 @@ uint64_t sock_buf_size = 0;
std::map portid_map;
std::mutex portid_map_mutex;
+// probation timer event to enable flow control at receivers
+const int64_t kBaseTimerInt = 200; // in centisecond
+const uint8_t kTxProbMaxRetries = 10;
+Timer txprob_timer(Event::Type::kEvtTmrTxProb);
+
// chunk ack parameters
// todo: The chunk ack timeout and chunk ack size should be
configurable
int kChunkAckTimeout = 1000; // in miliseconds
@@ -76,13 +82,37 @@ TipcPortId* portid_lookup(struct tipc_portid id) {
return portid_map[uid];
}
+void tmr_exp_cbk(void* uarg) {
+ Timer* timer = reinterpret_cast(uarg);
+ if (timer != nullptr) {
+ timer->is_active_ = false;
+ // send to fctrl thread
+ if (m_NCS_IPC_SEND(&mbx_events, new Event(timer->type_),
+ NCS_IPC_PRIORITY_HIGH) != NCSCC_RC_SUCCESS) {
+ m_MDS_LOG_ERR("FCTRL: Failed to send msg to mbx_events\n");
+ }
+ }
+}
+
void process_timer_event(const Event evt) {
+ bool txprob_restart = false;
for (auto i : portid_map) {
TipcPortId* portid = i.second;
+
+ if (evt.type_ == Event::Type::kEvtTmrTxProb) {
+ if (portid->ReceiveTmrTxProb(kTxProbMaxRetries) == true) {
+ txprob_restart = true;
+ }
+ }
+
if (evt.type_ == Event::Type::kEvtTmrChunkAck) {
portid->ReceiveTmrChunkAck();
}
}
+ if (txprob_restart) {
+ txprob_timer.Start(kBaseTimerInt, tmr_exp_cbk);
+ m_MDS_LOG_DBG("FCTRL: Restart txprob");
+ }
}
uint32_t process_flow_event(const Event evt) {
@@ -231,8 +261,10 @@ uint32_t mds_tipc_fctrl_sndqueue_capable(struct
tipc_portid id, uint16_t len,
id.node, id.ref, __LINE__);
rc = NCSCC_RC_FAILURE;
} else {
- // assign the sequence number of the outgoing message
- *next_seq = portid->GetCurrentSeq();
+ if (portid->state_ != TipcPortId::State::kDisabled) {
+ // assign the sequence number of the outgoing message
+ *next_seq = portid->GetCurrentSeq();
+ }
}
portid_map_mutex.unlock();
@@ -252,7 +284,16 @@ uint32_t mds_tipc_fctrl_trysend(const uint8_t
*buffer, uint16_t len,
id.node, id.ref, __LINE__);
rc = NCSCC_RC_FAILURE;
} else {
- portid->Queue(buffer, len);
+ if (portid->state_ != TipcPortId::State::kDisabled) {
+ portid->Queue(buffer, len);
+ // start txprob timer for the first msg sent out
+ // do not start for other states
+ if (portid->state_ == TipcPortId::State::kStartup) {
+ txprob_timer.Start(kBaseTimerInt, tmr_exp_cbk);
+ m_MDS_LOG_DBG("FCTRL: Start txprob");
+ portid->state_ = TipcPortId::State::kTxProb;
+ }
+ }
}
portid_map_mutex.unlock();
diff --git a/src/mds/mds_tipc_fctrl_msg.h b/src/mds/mds_tipc_fctrl_msg.h
index 8e6a874..69f8048 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
kEvtTmrAll,
+ kEvtTmrTxProb, // event that tx probation timer expired for once
kEvtTmrChunkAck, // event to send the chunk ack
};
NCS_IPC_MSG next_{0};
diff --git a/src/mds/mds_tipc_fctrl_portid.cc
b/src/mds/mds_tipc_fctrl_portid.cc
index 64115d5..84ecee9 100644
--- a/src/mds/mds_tipc_fctrl_portid.cc
+++ b/src/mds/mds_tipc_fctrl_portid.cc
@@ -23,6 +23,35 @@
namespace mds {
+Timer::Timer(Event::Type type) {
+ tmr_id_ = nullptr;
+ type_ = type;
+ is_active_ = false;
+}
+
+Timer::~Timer() {
[Vu] Is it required to stop the timer here if it still in active?
[M]: Yes, will add the Stop() here
+}
+
+void Timer::Start(int64_t period, void (*tmr_exp_func)(void*)) {
+ // timer will not start if it's already started
+ // period is in centiseconds
+ if (is_active_ == false) {
+ if (tmr_id_ == nullptr) {
+ tmr_id_ = ncs_tmr_alloc(nullptr, 0);
+ }
+ tmr_id_ = ncs_tmr_start(tmr_id_, period, tmr_exp_func, this,
+ nullptr, 0);
+ is_active_ = true;
+ }
+}
+
+void Timer::Stop() {
[Vu] This method is not called from anywhere. Is the
Re: [devel] [PATCH 3/9] mds: Add implementation for TIPC buffer overflow solution [#1960]
Hi Minh,
See my responses to your comments below, started with [Vu2].
Regards, Vu
On 9/16/19 1:06 PM, Minh Hon Chau wrote:
Hi Vu,
Several comments with [M] too :).
Thanks
Minh
On 16/9/19 2:24 pm, Nguyen Minh Vu wrote:
Hi Minh,
I have several comments below, started with [Vu].
Regards, Vu
On 8/14/19 1:01 PM, Minh Chau wrote:
This is a collaborative patch of two participants:
- Tran Thuan
- Minh Chau
Main changes:
- Add mds_tipc_fctrl_intf.h, mds_tipc_fctrl_intf.cc: These two files
introduce new functions which are called in mds_dt_tipc.c if the flow
control is enabled
- Add mds_tipc_fctrl_portid.h, mds_tipc_fctrl_portid.cc: These files
implements the tipc portid instance, which supports the sliding window,
mds msg queue
- Add mds_tipc_fctrl_msg.h, mds_tipc_fctrl_msg.cc: These files define
the event and messages which are used for this solution.
---
src/mds/Makefile.am | 10 +-
src/mds/mds_dt.h | 8 +-
src/mds/mds_dt_tipc.c | 188 +---
src/mds/mds_tipc_fctrl_intf.cc | 376
+++
src/mds/mds_tipc_fctrl_intf.h | 47 +
src/mds/mds_tipc_fctrl_msg.cc | 142 +++
src/mds/mds_tipc_fctrl_msg.h | 129 ++
src/mds/mds_tipc_fctrl_portid.cc | 261 +++
src/mds/mds_tipc_fctrl_portid.h | 87 +
9 files changed, 1184 insertions(+), 64 deletions(-)
create mode 100644 src/mds/mds_tipc_fctrl_intf.cc
create mode 100644 src/mds/mds_tipc_fctrl_intf.h
create mode 100644 src/mds/mds_tipc_fctrl_msg.cc
create mode 100644 src/mds/mds_tipc_fctrl_msg.h
create mode 100644 src/mds/mds_tipc_fctrl_portid.cc
create mode 100644 src/mds/mds_tipc_fctrl_portid.h
diff --git a/src/mds/Makefile.am b/src/mds/Makefile.am
index 2d7b652..d849e8f 100644
--- a/src/mds/Makefile.am
+++ b/src/mds/Makefile.am
@@ -48,10 +48,16 @@ lib_libopensaf_core_la_SOURCES += \
if ENABLE_TIPC_TRANSPORT
noinst_HEADERS += src/mds/mds_dt_tipc.h \
src/mds/mds_tipc_recvq_stats.h \
- src/mds/mds_tipc_recvq_stats_impl.h
+ src/mds/mds_tipc_recvq_stats_impl.h \
+ src/mds/mds_tipc_fctrl_intf.h \
+ src/mds/mds_tipc_fctrl_portid.h \
+ src/mds/mds_tipc_fctrl_msg.h
lib_libopensaf_core_la_SOURCES += src/mds/mds_dt_tipc.c \
src/mds/mds_tipc_recvq_stats.cc \
- src/mds/mds_tipc_recvq_stats_impl.cc
+ src/mds/mds_tipc_recvq_stats_impl.cc \
+ src/mds/mds_tipc_fctrl_intf.cc \
+ src/mds/mds_tipc_fctrl_portid.cc \
+ src/mds/mds_tipc_fctrl_msg.cc
endif
if ENABLE_TESTS
diff --git a/src/mds/mds_dt.h b/src/mds/mds_dt.h
index b645bb4..d9e8633 100644
--- a/src/mds/mds_dt.h
+++ b/src/mds/mds_dt.h
@@ -162,7 +162,7 @@ uint32_t mdtm_del_from_ref_tbl(MDS_SUBTN_REF_VAL
ref);
uint32_t mds_tmr_mailbox_processing(void);
uint32_t mdtm_get_from_ref_tbl(MDS_SUBTN_REF_VAL ref, MDS_SVC_HDL
*svc_hdl);
uint32_t mdtm_add_frag_hdr(uint8_t *buf_ptr, uint16_t len,
uint32_t seq_num,
- uint16_t frag_byte);
+ uint16_t frag_byte, uint16_t
fctrl_seq_num);
uint32_t mdtm_free_reassem_msg_mem(MDS_ENCODED_MSG *msg);
uint32_t mdtm_process_recv_data(uint8_t *buf, uint16_t len,
uint64_t tipc_id,
uint32_t *buff_dump);
@@ -240,9 +240,13 @@ bool mdtm_mailbox_mbx_cleanup(NCSCONTEXT arg,
NCSCONTEXT msg);
#define MDS_PROT 0xA0
#define MDS_VERSION 0x08
-#define MDS_PROT_VER_MASK (MDS_PROT | MDS_VERSION)
+#define MDS_PROT_VER_MASK 0xFC
#define MDTM_PRI_MASK 0x3
+/* MDS protocol/version for flow control */
+#define MDS_PROT_FCTRL (0xB0 | MDS_VERSION)
+#define MDS_PROT_FCTRL_ID 0x00AC13F5
+
/* Added for the subscription changes */
#define MDS_NCS_CHASSIS_ID (m_NCS_GET_NODE_ID & 0x00ff)
#define MDS_TIPC_COMMON_ID 0x01001000
diff --git a/src/mds/mds_dt_tipc.c b/src/mds/mds_dt_tipc.c
index 86b52bb..fef1c50 100644
--- a/src/mds/mds_dt_tipc.c
+++ b/src/mds/mds_dt_tipc.c
@@ -47,6 +47,7 @@
#include "mds_dt_tipc.h"
#include "mds_dt_tcp_disc.h"
#include "mds_core.h"
+#include "mds_tipc_fctrl_intf.h"
#include "mds_tipc_recvq_stats.h"
#include "base/osaf_utility.h"
#include "base/osaf_poll.h"
@@ -165,20 +166,22 @@ NCS_PATRICIA_TREE mdtm_reassembly_list;
uint32_t mdtm_global_frag_num;
const unsigned int MAX_RECV_THRESHOLD = 30;
+uint8_t gl_mds_pro_ver = MDS_PROT | MDS_VERSION;
-static bool get_tipc_port_id(int sock, uint32_t* port_id) {
+static bool get_tipc_port_id(int sock, struct tipc_portid* port_id) {
struct sockaddr_tipc addr;
socklen_t sz = sizeof(addr);
memset(&addr, 0, sizeof(addr));
- *port_id = 0;
+ port_id->node = 0;
+ port_id->ref = 0;
if (0 > getsockname(sock, (struct sockaddr *)&addr, &sz)) {
syslog(LOG_ERR, "MDTM:TIPC Failed to get socket name, err:
%s",
strerror(errno));
return false;
}
- *port_id = addr.addr.id.ref;
+ *port_id = addr.addr.id;
Re: [devel] [PATCH 4/9] mds: Add timeout for ack message [#1960]
Hi Vu,
Some comments with [M]
Thanks
Minh
On 16/9/19 2:37 pm, Nguyen Minh Vu wrote:
Hi Minh,
I have minor comments below.
Regards, Vu
On 8/14/19 1:38 PM, Minh Chau wrote:
If the ack size is configured greater than 1, there should be a timeout
at receiver ends to send the ack message back to senders.
The ack message timeout utilizes the poll timeout in flow control thread
to make mds lightweight (in contrast to additional timer threads).
---
src/mds/mds_tipc_fctrl_intf.cc | 33
++---
src/mds/mds_tipc_fctrl_msg.h | 6 ++
src/mds/mds_tipc_fctrl_portid.cc | 15 +++
src/mds/mds_tipc_fctrl_portid.h | 1 +
4 files changed, 52 insertions(+), 3 deletions(-)
diff --git a/src/mds/mds_tipc_fctrl_intf.cc
b/src/mds/mds_tipc_fctrl_intf.cc
index 91b9107..bd0a8f6 100644
--- a/src/mds/mds_tipc_fctrl_intf.cc
+++ b/src/mds/mds_tipc_fctrl_intf.cc
@@ -66,7 +66,8 @@ std::map portid_map;
std::mutex portid_map_mutex;
// chunk ack parameters
-// todo: The chunk ack size should be configurable
+// todo: The chunk ack timeout and chunk ack size should be
configurable
+int kChunkAckTimeout = 1000; // in miliseconds
uint16_t kChunkAckSize = 3;
TipcPortId* portid_lookup(struct tipc_portid id) {
@@ -75,6 +76,15 @@ TipcPortId* portid_lookup(struct tipc_portid id) {
return portid_map[uid];
}
+void process_timer_event(const Event evt) {
+ for (auto i : portid_map) {
+ TipcPortId* portid = i.second;
+ if (evt.type_ == Event::Type::kEvtTmrChunkAck) {
+ portid->ReceiveTmrChunkAck();
+ }
+ }
+}
+
uint32_t process_flow_event(const Event evt) {
uint32_t rc = NCSCC_RC_SUCCESS;
TipcPortId *portid = portid_lookup(evt.id_);
@@ -110,7 +120,7 @@ uint32_t process_flow_event(const Event evt) {
uint32_t process_all_events(void) {
enum { FD_FCTRL = 0, NUM_FDS };
- int poll_tmo = MDTM_TIPC_POLL_TIMEOUT;
+ int poll_tmo = kChunkAckTimeout;
while (true) {
int pollres;
struct pollfd pfd[NUM_FDS] = {{0}};
@@ -135,11 +145,24 @@ uint32_t process_all_events(void) {
if (evt == nullptr) continue;
portid_map_mutex.lock();
- process_flow_event(*evt);
+
+ if (evt->IsTimerEvent()) {
+ process_timer_event(*evt);
+ }
+ if (evt->IsFlowEvent()) {
+ process_flow_event(*evt);
+ }
+
[Vu] Should log something here if the event is none of above?
[M] Probably not, the event is created internally so we know there won't
be any rather than the above
delete evt;
portid_map_mutex.unlock();
}
}
+ // timeout, scan all portid and send ack msgs
+ if (pollres == 0) {
+ portid_map_mutex.lock();
+ process_timer_event(Event(Event::Type::kEvtTmrChunkAck));
+ portid_map_mutex.unlock();
+ }
} /* while */
return NCSCC_RC_SUCCESS;
}
@@ -368,6 +391,10 @@ uint32_t mds_tipc_fctrl_rcv_data(uint8_t
*buffer, uint16_t len,
portid_map_mutex.lock();
uint32_t rc = process_flow_event(Event(Event::Type::kEvtRcvData,
id, data.svc_id_, header.mseq_, header.mfrag_,
header.fseq_));
+ if (rc == NCSCC_RC_CONTINUE) {
+ process_timer_event(Event(Event::Type::kEvtTmrChunkAck));
[Vu] Missed to unlock the mutex here
[M] It's not missed, it's called before return
+ rc = NCSCC_RC_SUCCESS;
+ }
portid_map_mutex.unlock();
return rc;
}
diff --git a/src/mds/mds_tipc_fctrl_msg.h b/src/mds/mds_tipc_fctrl_msg.h
index 677f256..8e6a874 100644
--- a/src/mds/mds_tipc_fctrl_msg.h
+++ b/src/mds/mds_tipc_fctrl_msg.h
@@ -44,6 +44,8 @@ class Event {
// selective data msgs (not supported)
kEvtDropData, // event reported from tipc that a
message is not
// delivered
+ kEvtTmrAll,
+ kEvtTmrChunkAck, // event to send the chunk ack
};
NCS_IPC_MSG next_{0};
Type type_;
@@ -68,6 +70,10 @@ class Event {
fseq_(f_seg_num), chunk_size_(chunk_size) {
type_ = type;
}
+ bool IsTimerEvent() { return (type_ > Type::kEvtTmrAll); }
+ bool IsFlowEvent() {
+ return (Type::kEvtDataFlowAll < type_ && type_ < Type::kEvtTmrAll);
+ }
[Vu] Consider making these ones to be constant methods if they do not
change any of their attribute values.
[M] Yes, will add const
};
class BaseMessage {
diff --git a/src/mds/mds_tipc_fctrl_portid.cc
b/src/mds/mds_tipc_fctrl_portid.cc
index 24d13ee..64115d5 100644
--- a/src/mds/mds_tipc_fctrl_portid.cc
+++ b/src/mds/mds_tipc_fctrl_portid.cc
@@ -67,6 +67,8 @@ TipcPortId::TipcPortId(struct tipc_portid id, int
sock, uint16_t chksize,
}
TipcPortId::~TipcPortId() {
+ // Fake a TmrChunkAck event to ack all received messages
+ ReceiveTmrChunkAck();
// clear all msg in sndqueue_
sndqueue_.Clear();
}
@@ -156,6 +158,7 @@ uint32_t TipcPortId::ReceiveData(uint32_t mseq,
uint16_t mfrag,
// send ack for @chunk_size_ msgs starting
Re: [devel] [PATCH 3/9] mds: Add implementation for TIPC buffer overflow solution [#1960]
Hi Vu,
Several comments with [M] too :).
Thanks
Minh
On 16/9/19 2:24 pm, Nguyen Minh Vu wrote:
Hi Minh,
I have several comments below, started with [Vu].
Regards, Vu
On 8/14/19 1:01 PM, Minh Chau wrote:
This is a collaborative patch of two participants:
- Tran Thuan
- Minh Chau
Main changes:
- Add mds_tipc_fctrl_intf.h, mds_tipc_fctrl_intf.cc: These two files
introduce new functions which are called in mds_dt_tipc.c if the flow
control is enabled
- Add mds_tipc_fctrl_portid.h, mds_tipc_fctrl_portid.cc: These files
implements the tipc portid instance, which supports the sliding window,
mds msg queue
- Add mds_tipc_fctrl_msg.h, mds_tipc_fctrl_msg.cc: These files define
the event and messages which are used for this solution.
---
src/mds/Makefile.am | 10 +-
src/mds/mds_dt.h | 8 +-
src/mds/mds_dt_tipc.c | 188 +---
src/mds/mds_tipc_fctrl_intf.cc | 376
+++
src/mds/mds_tipc_fctrl_intf.h | 47 +
src/mds/mds_tipc_fctrl_msg.cc | 142 +++
src/mds/mds_tipc_fctrl_msg.h | 129 ++
src/mds/mds_tipc_fctrl_portid.cc | 261 +++
src/mds/mds_tipc_fctrl_portid.h | 87 +
9 files changed, 1184 insertions(+), 64 deletions(-)
create mode 100644 src/mds/mds_tipc_fctrl_intf.cc
create mode 100644 src/mds/mds_tipc_fctrl_intf.h
create mode 100644 src/mds/mds_tipc_fctrl_msg.cc
create mode 100644 src/mds/mds_tipc_fctrl_msg.h
create mode 100644 src/mds/mds_tipc_fctrl_portid.cc
create mode 100644 src/mds/mds_tipc_fctrl_portid.h
diff --git a/src/mds/Makefile.am b/src/mds/Makefile.am
index 2d7b652..d849e8f 100644
--- a/src/mds/Makefile.am
+++ b/src/mds/Makefile.am
@@ -48,10 +48,16 @@ lib_libopensaf_core_la_SOURCES += \
if ENABLE_TIPC_TRANSPORT
noinst_HEADERS += src/mds/mds_dt_tipc.h \
src/mds/mds_tipc_recvq_stats.h \
- src/mds/mds_tipc_recvq_stats_impl.h
+ src/mds/mds_tipc_recvq_stats_impl.h \
+ src/mds/mds_tipc_fctrl_intf.h \
+ src/mds/mds_tipc_fctrl_portid.h \
+ src/mds/mds_tipc_fctrl_msg.h
lib_libopensaf_core_la_SOURCES += src/mds/mds_dt_tipc.c \
src/mds/mds_tipc_recvq_stats.cc \
- src/mds/mds_tipc_recvq_stats_impl.cc
+ src/mds/mds_tipc_recvq_stats_impl.cc \
+ src/mds/mds_tipc_fctrl_intf.cc \
+ src/mds/mds_tipc_fctrl_portid.cc \
+ src/mds/mds_tipc_fctrl_msg.cc
endif
if ENABLE_TESTS
diff --git a/src/mds/mds_dt.h b/src/mds/mds_dt.h
index b645bb4..d9e8633 100644
--- a/src/mds/mds_dt.h
+++ b/src/mds/mds_dt.h
@@ -162,7 +162,7 @@ uint32_t mdtm_del_from_ref_tbl(MDS_SUBTN_REF_VAL
ref);
uint32_t mds_tmr_mailbox_processing(void);
uint32_t mdtm_get_from_ref_tbl(MDS_SUBTN_REF_VAL ref, MDS_SVC_HDL
*svc_hdl);
uint32_t mdtm_add_frag_hdr(uint8_t *buf_ptr, uint16_t len, uint32_t
seq_num,
- uint16_t frag_byte);
+ uint16_t frag_byte, uint16_t fctrl_seq_num);
uint32_t mdtm_free_reassem_msg_mem(MDS_ENCODED_MSG *msg);
uint32_t mdtm_process_recv_data(uint8_t *buf, uint16_t len,
uint64_t tipc_id,
uint32_t *buff_dump);
@@ -240,9 +240,13 @@ bool mdtm_mailbox_mbx_cleanup(NCSCONTEXT arg,
NCSCONTEXT msg);
#define MDS_PROT 0xA0
#define MDS_VERSION 0x08
-#define MDS_PROT_VER_MASK (MDS_PROT | MDS_VERSION)
+#define MDS_PROT_VER_MASK 0xFC
#define MDTM_PRI_MASK 0x3
+/* MDS protocol/version for flow control */
+#define MDS_PROT_FCTRL (0xB0 | MDS_VERSION)
+#define MDS_PROT_FCTRL_ID 0x00AC13F5
+
/* Added for the subscription changes */
#define MDS_NCS_CHASSIS_ID (m_NCS_GET_NODE_ID & 0x00ff)
#define MDS_TIPC_COMMON_ID 0x01001000
diff --git a/src/mds/mds_dt_tipc.c b/src/mds/mds_dt_tipc.c
index 86b52bb..fef1c50 100644
--- a/src/mds/mds_dt_tipc.c
+++ b/src/mds/mds_dt_tipc.c
@@ -47,6 +47,7 @@
#include "mds_dt_tipc.h"
#include "mds_dt_tcp_disc.h"
#include "mds_core.h"
+#include "mds_tipc_fctrl_intf.h"
#include "mds_tipc_recvq_stats.h"
#include "base/osaf_utility.h"
#include "base/osaf_poll.h"
@@ -165,20 +166,22 @@ NCS_PATRICIA_TREE mdtm_reassembly_list;
uint32_t mdtm_global_frag_num;
const unsigned int MAX_RECV_THRESHOLD = 30;
+uint8_t gl_mds_pro_ver = MDS_PROT | MDS_VERSION;
-static bool get_tipc_port_id(int sock, uint32_t* port_id) {
+static bool get_tipc_port_id(int sock, struct tipc_portid* port_id) {
struct sockaddr_tipc addr;
socklen_t sz = sizeof(addr);
memset(&addr, 0, sizeof(addr));
- *port_id = 0;
+ port_id->node = 0;
+ port_id->ref = 0;
if (0 > getsockname(sock, (struct sockaddr *)&addr, &sz)) {
syslog(LOG_ERR, "MDTM:TIPC Failed to get socket name, err:
%s",
strerror(errno));
return false;
}
- *port_id = addr.addr.id.ref;
+ *port_id = addr.addr.id;
return true;
}
@@ -240,12 +243,13 @@ uint32_t mdtm_tipc_init(NODE_ID nodeid,
uint32_t *mds_tipc_ref)
}
/* Cod
Re: [devel] [PATCH 5/9] mds: Add state machine for tipc portid instance [#1960]
Hi Minh,
I has few comments below.
Regards, Vu
On 8/14/19 1:38 PM, Minh Chau wrote:
This patch adds state machine to support tx probation timer.
---
src/mds/mds_tipc_fctrl_intf.cc | 47 +++--
src/mds/mds_tipc_fctrl_msg.h | 1 +
src/mds/mds_tipc_fctrl_portid.cc | 109 +++
src/mds/mds_tipc_fctrl_portid.h | 22
4 files changed, 176 insertions(+), 3 deletions(-)
diff --git a/src/mds/mds_tipc_fctrl_intf.cc b/src/mds/mds_tipc_fctrl_intf.cc
index bd0a8f6..c2d0922 100644
--- a/src/mds/mds_tipc_fctrl_intf.cc
+++ b/src/mds/mds_tipc_fctrl_intf.cc
@@ -34,6 +34,7 @@
using mds::Event;
using mds::TipcPortId;
+using mds::Timer;
using mds::DataMessage;
using mds::ChunkAck;
using mds::HeaderMessage;
@@ -65,6 +66,11 @@ uint64_t sock_buf_size = 0;
std::map portid_map;
std::mutex portid_map_mutex;
+// probation timer event to enable flow control at receivers
+const int64_t kBaseTimerInt = 200; // in centisecond
+const uint8_t kTxProbMaxRetries = 10;
+Timer txprob_timer(Event::Type::kEvtTmrTxProb);
+
// chunk ack parameters
// todo: The chunk ack timeout and chunk ack size should be configurable
int kChunkAckTimeout = 1000; // in miliseconds
@@ -76,13 +82,37 @@ TipcPortId* portid_lookup(struct tipc_portid id) {
return portid_map[uid];
}
+void tmr_exp_cbk(void* uarg) {
+ Timer* timer = reinterpret_cast(uarg);
+ if (timer != nullptr) {
+timer->is_active_ = false;
+// send to fctrl thread
+if (m_NCS_IPC_SEND(&mbx_events, new Event(timer->type_),
+NCS_IPC_PRIORITY_HIGH) != NCSCC_RC_SUCCESS) {
+ m_MDS_LOG_ERR("FCTRL: Failed to send msg to mbx_events\n");
+}
+ }
+}
+
void process_timer_event(const Event evt) {
+ bool txprob_restart = false;
for (auto i : portid_map) {
TipcPortId* portid = i.second;
+
+if (evt.type_ == Event::Type::kEvtTmrTxProb) {
+ if (portid->ReceiveTmrTxProb(kTxProbMaxRetries) == true) {
+txprob_restart = true;
+ }
+}
+
if (evt.type_ == Event::Type::kEvtTmrChunkAck) {
portid->ReceiveTmrChunkAck();
}
}
+ if (txprob_restart) {
+txprob_timer.Start(kBaseTimerInt, tmr_exp_cbk);
+m_MDS_LOG_DBG("FCTRL: Restart txprob");
+ }
}
uint32_t process_flow_event(const Event evt) {
@@ -231,8 +261,10 @@ uint32_t mds_tipc_fctrl_sndqueue_capable(struct
tipc_portid id, uint16_t len,
id.node, id.ref, __LINE__);
rc = NCSCC_RC_FAILURE;
} else {
-// assign the sequence number of the outgoing message
-*next_seq = portid->GetCurrentSeq();
+if (portid->state_ != TipcPortId::State::kDisabled) {
+ // assign the sequence number of the outgoing message
+ *next_seq = portid->GetCurrentSeq();
+}
}
portid_map_mutex.unlock();
@@ -252,7 +284,16 @@ uint32_t mds_tipc_fctrl_trysend(const uint8_t *buffer,
uint16_t len,
id.node, id.ref, __LINE__);
rc = NCSCC_RC_FAILURE;
} else {
-portid->Queue(buffer, len);
+if (portid->state_ != TipcPortId::State::kDisabled) {
+ portid->Queue(buffer, len);
+ // start txprob timer for the first msg sent out
+ // do not start for other states
+ if (portid->state_ == TipcPortId::State::kStartup) {
+txprob_timer.Start(kBaseTimerInt, tmr_exp_cbk);
+m_MDS_LOG_DBG("FCTRL: Start txprob");
+portid->state_ = TipcPortId::State::kTxProb;
+ }
+}
}
portid_map_mutex.unlock();
diff --git a/src/mds/mds_tipc_fctrl_msg.h b/src/mds/mds_tipc_fctrl_msg.h
index 8e6a874..69f8048 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
kEvtTmrAll,
+kEvtTmrTxProb,// event that tx probation timer expired for once
kEvtTmrChunkAck, // event to send the chunk ack
};
NCS_IPC_MSG next_{0};
diff --git a/src/mds/mds_tipc_fctrl_portid.cc b/src/mds/mds_tipc_fctrl_portid.cc
index 64115d5..84ecee9 100644
--- a/src/mds/mds_tipc_fctrl_portid.cc
+++ b/src/mds/mds_tipc_fctrl_portid.cc
@@ -23,6 +23,35 @@
namespace mds {
+Timer::Timer(Event::Type type) {
+ tmr_id_ = nullptr;
+ type_ = type;
+ is_active_ = false;
+}
+
+Timer::~Timer() {
[Vu] Is it required to stop the timer here if it still in active?
+}
+
+void Timer::Start(int64_t period, void (*tmr_exp_func)(void*)) {
+ // timer will not start if it's already started
+ // period is in centiseconds
+ if (is_active_ == false) {
+if (tmr_id_ == nullptr) {
+ tmr_id_ = ncs_tmr_alloc(nullptr, 0);
+}
+tmr_id_ = ncs_tmr_start(tmr_id_, period, tmr_exp_func, this,
+nullptr, 0);
+is_active_ = true;
+ }
+}
+
+void Timer::Stop() {
[Vu] This method is not called from anywhere. Is there any case the
timer should be stopped before the timer gets expired?
+ if (is_active_ == true) {
+ncs_tmr_stop(tm
Re: [devel] [PATCH 4/9] mds: Add timeout for ack message [#1960]
Hi Minh,
I have minor comments below.
Regards, Vu
On 8/14/19 1:38 PM, Minh Chau wrote:
If the ack size is configured greater than 1, there should be a timeout
at receiver ends to send the ack message back to senders.
The ack message timeout utilizes the poll timeout in flow control thread
to make mds lightweight (in contrast to additional timer threads).
---
src/mds/mds_tipc_fctrl_intf.cc | 33 ++---
src/mds/mds_tipc_fctrl_msg.h | 6 ++
src/mds/mds_tipc_fctrl_portid.cc | 15 +++
src/mds/mds_tipc_fctrl_portid.h | 1 +
4 files changed, 52 insertions(+), 3 deletions(-)
diff --git a/src/mds/mds_tipc_fctrl_intf.cc b/src/mds/mds_tipc_fctrl_intf.cc
index 91b9107..bd0a8f6 100644
--- a/src/mds/mds_tipc_fctrl_intf.cc
+++ b/src/mds/mds_tipc_fctrl_intf.cc
@@ -66,7 +66,8 @@ std::map portid_map;
std::mutex portid_map_mutex;
// chunk ack parameters
-// todo: The chunk ack size should be configurable
+// todo: The chunk ack timeout and chunk ack size should be configurable
+int kChunkAckTimeout = 1000; // in miliseconds
uint16_t kChunkAckSize = 3;
TipcPortId* portid_lookup(struct tipc_portid id) {
@@ -75,6 +76,15 @@ TipcPortId* portid_lookup(struct tipc_portid id) {
return portid_map[uid];
}
+void process_timer_event(const Event evt) {
+ for (auto i : portid_map) {
+TipcPortId* portid = i.second;
+if (evt.type_ == Event::Type::kEvtTmrChunkAck) {
+ portid->ReceiveTmrChunkAck();
+}
+ }
+}
+
uint32_t process_flow_event(const Event evt) {
uint32_t rc = NCSCC_RC_SUCCESS;
TipcPortId *portid = portid_lookup(evt.id_);
@@ -110,7 +120,7 @@ uint32_t process_flow_event(const Event evt) {
uint32_t process_all_events(void) {
enum { FD_FCTRL = 0, NUM_FDS };
- int poll_tmo = MDTM_TIPC_POLL_TIMEOUT;
+ int poll_tmo = kChunkAckTimeout;
while (true) {
int pollres;
struct pollfd pfd[NUM_FDS] = {{0}};
@@ -135,11 +145,24 @@ uint32_t process_all_events(void) {
if (evt == nullptr) continue;
portid_map_mutex.lock();
-process_flow_event(*evt);
+
+if (evt->IsTimerEvent()) {
+ process_timer_event(*evt);
+}
+if (evt->IsFlowEvent()) {
+ process_flow_event(*evt);
+}
+
[Vu] Should log something here if the event is none of above?
delete evt;
portid_map_mutex.unlock();
}
}
+// timeout, scan all portid and send ack msgs
+if (pollres == 0) {
+ portid_map_mutex.lock();
+ process_timer_event(Event(Event::Type::kEvtTmrChunkAck));
+ portid_map_mutex.unlock();
+}
} /* while */
return NCSCC_RC_SUCCESS;
}
@@ -368,6 +391,10 @@ uint32_t mds_tipc_fctrl_rcv_data(uint8_t *buffer, uint16_t
len,
portid_map_mutex.lock();
uint32_t rc = process_flow_event(Event(Event::Type::kEvtRcvData,
id, data.svc_id_, header.mseq_, header.mfrag_, header.fseq_));
+ if (rc == NCSCC_RC_CONTINUE) {
+process_timer_event(Event(Event::Type::kEvtTmrChunkAck));
[Vu] Missed to unlock the mutex here
+rc = NCSCC_RC_SUCCESS;
+ }
portid_map_mutex.unlock();
return rc;
}
diff --git a/src/mds/mds_tipc_fctrl_msg.h b/src/mds/mds_tipc_fctrl_msg.h
index 677f256..8e6a874 100644
--- a/src/mds/mds_tipc_fctrl_msg.h
+++ b/src/mds/mds_tipc_fctrl_msg.h
@@ -44,6 +44,8 @@ class Event {
// selective data msgs (not supported)
kEvtDropData, // event reported from tipc that a message is not
// delivered
+kEvtTmrAll,
+kEvtTmrChunkAck, // event to send the chunk ack
};
NCS_IPC_MSG next_{0};
Type type_;
@@ -68,6 +70,10 @@ class Event {
fseq_(f_seg_num), chunk_size_(chunk_size) {
type_ = type;
}
+ bool IsTimerEvent() { return (type_ > Type::kEvtTmrAll); }
+ bool IsFlowEvent() {
+return (Type::kEvtDataFlowAll < type_ && type_ < Type::kEvtTmrAll);
+ }
[Vu] Consider making these ones to be constant methods if they do not
change any of their attribute values.
};
class BaseMessage {
diff --git a/src/mds/mds_tipc_fctrl_portid.cc b/src/mds/mds_tipc_fctrl_portid.cc
index 24d13ee..64115d5 100644
--- a/src/mds/mds_tipc_fctrl_portid.cc
+++ b/src/mds/mds_tipc_fctrl_portid.cc
@@ -67,6 +67,8 @@ TipcPortId::TipcPortId(struct tipc_portid id, int sock,
uint16_t chksize,
}
TipcPortId::~TipcPortId() {
+ // Fake a TmrChunkAck event to ack all received messages
+ ReceiveTmrChunkAck();
// clear all msg in sndqueue_
sndqueue_.Clear();
}
@@ -156,6 +158,7 @@ uint32_t TipcPortId::ReceiveData(uint32_t mseq, uint16_t
mfrag,
// send ack for @chunk_size_ msgs starting from fseq
SendChunkAck(fseq, svc_id, chunk_size_);
rcvwnd_.acked_ = rcvwnd_.rcv_;
+ rc = NCSCC_RC_CONTINUE;
}
} else {
// todo: update rcvwnd_.nacked_space_.
@@ -258,4 +261,16 @@ void TipcPortId::ReceiveNack(uint32_t mseq,
Re: [devel] [PATCH 3/9] mds: Add implementation for TIPC buffer overflow solution [#1960]
Hi Minh,
I have several comments below, started with [Vu].
Regards, Vu
On 8/14/19 1:01 PM, Minh Chau wrote:
This is a collaborative patch of two participants:
- Tran Thuan
- Minh Chau
Main changes:
- Add mds_tipc_fctrl_intf.h, mds_tipc_fctrl_intf.cc: These two files
introduce new functions which are called in mds_dt_tipc.c if the flow
control is enabled
- Add mds_tipc_fctrl_portid.h, mds_tipc_fctrl_portid.cc: These files
implements the tipc portid instance, which supports the sliding window,
mds msg queue
- Add mds_tipc_fctrl_msg.h, mds_tipc_fctrl_msg.cc: These files define
the event and messages which are used for this solution.
---
src/mds/Makefile.am | 10 +-
src/mds/mds_dt.h | 8 +-
src/mds/mds_dt_tipc.c| 188 +---
src/mds/mds_tipc_fctrl_intf.cc | 376 +++
src/mds/mds_tipc_fctrl_intf.h| 47 +
src/mds/mds_tipc_fctrl_msg.cc| 142 +++
src/mds/mds_tipc_fctrl_msg.h | 129 ++
src/mds/mds_tipc_fctrl_portid.cc | 261 +++
src/mds/mds_tipc_fctrl_portid.h | 87 +
9 files changed, 1184 insertions(+), 64 deletions(-)
create mode 100644 src/mds/mds_tipc_fctrl_intf.cc
create mode 100644 src/mds/mds_tipc_fctrl_intf.h
create mode 100644 src/mds/mds_tipc_fctrl_msg.cc
create mode 100644 src/mds/mds_tipc_fctrl_msg.h
create mode 100644 src/mds/mds_tipc_fctrl_portid.cc
create mode 100644 src/mds/mds_tipc_fctrl_portid.h
diff --git a/src/mds/Makefile.am b/src/mds/Makefile.am
index 2d7b652..d849e8f 100644
--- a/src/mds/Makefile.am
+++ b/src/mds/Makefile.am
@@ -48,10 +48,16 @@ lib_libopensaf_core_la_SOURCES += \
if ENABLE_TIPC_TRANSPORT
noinst_HEADERS += src/mds/mds_dt_tipc.h \
src/mds/mds_tipc_recvq_stats.h \
- src/mds/mds_tipc_recvq_stats_impl.h
+ src/mds/mds_tipc_recvq_stats_impl.h \
+ src/mds/mds_tipc_fctrl_intf.h \
+ src/mds/mds_tipc_fctrl_portid.h \
+ src/mds/mds_tipc_fctrl_msg.h
lib_libopensaf_core_la_SOURCES += src/mds/mds_dt_tipc.c \
src/mds/mds_tipc_recvq_stats.cc \
- src/mds/mds_tipc_recvq_stats_impl.cc
+ src/mds/mds_tipc_recvq_stats_impl.cc \
+ src/mds/mds_tipc_fctrl_intf.cc \
+ src/mds/mds_tipc_fctrl_portid.cc \
+ src/mds/mds_tipc_fctrl_msg.cc
endif
if ENABLE_TESTS
diff --git a/src/mds/mds_dt.h b/src/mds/mds_dt.h
index b645bb4..d9e8633 100644
--- a/src/mds/mds_dt.h
+++ b/src/mds/mds_dt.h
@@ -162,7 +162,7 @@ uint32_t mdtm_del_from_ref_tbl(MDS_SUBTN_REF_VAL ref);
uint32_t mds_tmr_mailbox_processing(void);
uint32_t mdtm_get_from_ref_tbl(MDS_SUBTN_REF_VAL ref, MDS_SVC_HDL *svc_hdl);
uint32_t mdtm_add_frag_hdr(uint8_t *buf_ptr, uint16_t len, uint32_t seq_num,
- uint16_t frag_byte);
+ uint16_t frag_byte, uint16_t fctrl_seq_num);
uint32_t mdtm_free_reassem_msg_mem(MDS_ENCODED_MSG *msg);
uint32_t mdtm_process_recv_data(uint8_t *buf, uint16_t len, uint64_t tipc_id,
uint32_t *buff_dump);
@@ -240,9 +240,13 @@ bool mdtm_mailbox_mbx_cleanup(NCSCONTEXT arg, NCSCONTEXT
msg);
#define MDS_PROT 0xA0
#define MDS_VERSION 0x08
-#define MDS_PROT_VER_MASK (MDS_PROT | MDS_VERSION)
+#define MDS_PROT_VER_MASK 0xFC
#define MDTM_PRI_MASK 0x3
+/* MDS protocol/version for flow control */
+#define MDS_PROT_FCTRL (0xB0 | MDS_VERSION)
+#define MDS_PROT_FCTRL_ID 0x00AC13F5
+
/* Added for the subscription changes */
#define MDS_NCS_CHASSIS_ID (m_NCS_GET_NODE_ID & 0x00ff)
#define MDS_TIPC_COMMON_ID 0x01001000
diff --git a/src/mds/mds_dt_tipc.c b/src/mds/mds_dt_tipc.c
index 86b52bb..fef1c50 100644
--- a/src/mds/mds_dt_tipc.c
+++ b/src/mds/mds_dt_tipc.c
@@ -47,6 +47,7 @@
#include "mds_dt_tipc.h"
#include "mds_dt_tcp_disc.h"
#include "mds_core.h"
+#include "mds_tipc_fctrl_intf.h"
#include "mds_tipc_recvq_stats.h"
#include "base/osaf_utility.h"
#include "base/osaf_poll.h"
@@ -165,20 +166,22 @@ NCS_PATRICIA_TREE mdtm_reassembly_list;
uint32_t mdtm_global_frag_num;
const unsigned int MAX_RECV_THRESHOLD = 30;
+uint8_t gl_mds_pro_ver = MDS_PROT | MDS_VERSION;
-static bool get_tipc_port_id(int sock, uint32_t* port_id) {
+static bool get_tipc_port_id(int sock, struct tipc_portid* port_id) {
struct sockaddr_tipc addr;
socklen_t sz = sizeof(addr);
memset(&addr, 0, sizeof(addr));
- *port_id = 0;
+ port_id->node = 0;
+ port_id->ref = 0;
if (0 > getsockname(sock, (struct sockaddr *)&addr, &sz)) {
syslog(LOG_ERR, "MDTM:TIPC Failed to get socket name, err: %s",
strerror(errno));
return false;
}
- *port_id = addr.addr.id.ref;
+ *port_id = addr.addr.id;
return true;
}
@@ -240,12 +243,13 @@ uint32_t mdtm_tipc_init(NODE_ID nodeid, uint32_t *mds_tipc_ref)
}
/* Code for getting the self tipc ra
Re: [devel] [PATCH 3/9] mds: Add implementation for TIPC buffer overflow solution [#1960]
Hi Hans, Gary, Vu
Do you have any comments on remaining patches?
Thanks
Minh
On 11/9/19 11:01 am, Minh Hon Chau wrote:
Hi Gary,
Thanks for the review, please find comments with [M].
/Minh
On 10/9/19 6:02 pm, Gary Lee wrote:
Hi Minh & Thuan
Some minor comments marked with [GL].
On 14/8/19 4:38 pm, Minh Chau wrote:
This is a collaborative patch of two participants:Thuan, Minh.
Main changes:
- Add mds_tipc_fctrl_intf.h, mds_tipc_fctrl_intf.cc: These two files
introduce new functions which are called in mds_dt_tipc.c if the flow
control is enabled
- Add mds_tipc_fctrl_portid.h, mds_tipc_fctrl_portid.cc: These files
implements the tipc portid instance, which supports the sliding window,
mds msg queue
- Add mds_tipc_fctrl_msg.h, mds_tipc_fctrl_msg.cc: These files define
the event and messages which are used for this solution.
---
src/mds/Makefile.am | 10 +-
src/mds/mds_dt.h | 8 +-
src/mds/mds_dt_tipc.c | 188 +---
src/mds/mds_tipc_fctrl_intf.cc | 376
+++
src/mds/mds_tipc_fctrl_intf.h | 47 +
src/mds/mds_tipc_fctrl_msg.cc | 142 +++
src/mds/mds_tipc_fctrl_msg.h | 129 ++
src/mds/mds_tipc_fctrl_portid.cc | 261 +++
src/mds/mds_tipc_fctrl_portid.h | 87 +
9 files changed, 1184 insertions(+), 64 deletions(-)
create mode 100644 src/mds/mds_tipc_fctrl_intf.cc
create mode 100644 src/mds/mds_tipc_fctrl_intf.h
create mode 100644 src/mds/mds_tipc_fctrl_msg.cc
create mode 100644 src/mds/mds_tipc_fctrl_msg.h
create mode 100644 src/mds/mds_tipc_fctrl_portid.cc
create mode 100644 src/mds/mds_tipc_fctrl_portid.h
diff --git a/src/mds/Makefile.am b/src/mds/Makefile.am
index 2d7b652..d849e8f 100644
--- a/src/mds/Makefile.am
+++ b/src/mds/Makefile.am
@@ -48,10 +48,16 @@ lib_libopensaf_core_la_SOURCES += \
if ENABLE_TIPC_TRANSPORT
noinst_HEADERS += src/mds/mds_dt_tipc.h \
src/mds/mds_tipc_recvq_stats.h \
- src/mds/mds_tipc_recvq_stats_impl.h
+ src/mds/mds_tipc_recvq_stats_impl.h \
+ src/mds/mds_tipc_fctrl_intf.h \
+ src/mds/mds_tipc_fctrl_portid.h \
+ src/mds/mds_tipc_fctrl_msg.h
lib_libopensaf_core_la_SOURCES += src/mds/mds_dt_tipc.c \
src/mds/mds_tipc_recvq_stats.cc \
- src/mds/mds_tipc_recvq_stats_impl.cc
+ src/mds/mds_tipc_recvq_stats_impl.cc \
+ src/mds/mds_tipc_fctrl_intf.cc \
+ src/mds/mds_tipc_fctrl_portid.cc \
+ src/mds/mds_tipc_fctrl_msg.cc
endif
if ENABLE_TESTS
diff --git a/src/mds/mds_dt.h b/src/mds/mds_dt.h
index b645bb4..d9e8633 100644
--- a/src/mds/mds_dt.h
+++ b/src/mds/mds_dt.h
@@ -162,7 +162,7 @@ uint32_t mdtm_del_from_ref_tbl(MDS_SUBTN_REF_VAL
ref);
uint32_t mds_tmr_mailbox_processing(void);
uint32_t mdtm_get_from_ref_tbl(MDS_SUBTN_REF_VAL ref, MDS_SVC_HDL
*svc_hdl);
uint32_t mdtm_add_frag_hdr(uint8_t *buf_ptr, uint16_t len,
uint32_t seq_num,
- uint16_t frag_byte);
+ uint16_t frag_byte, uint16_t
fctrl_seq_num);
uint32_t mdtm_free_reassem_msg_mem(MDS_ENCODED_MSG *msg);
uint32_t mdtm_process_recv_data(uint8_t *buf, uint16_t len,
uint64_t tipc_id,
uint32_t *buff_dump);
@@ -240,9 +240,13 @@ bool mdtm_mailbox_mbx_cleanup(NCSCONTEXT arg,
NCSCONTEXT msg);
#define MDS_PROT 0xA0
#define MDS_VERSION 0x08
-#define MDS_PROT_VER_MASK (MDS_PROT | MDS_VERSION)
+#define MDS_PROT_VER_MASK 0xFC
#define MDTM_PRI_MASK 0x3
+/* MDS protocol/version for flow control */
+#define MDS_PROT_FCTRL (0xB0 | MDS_VERSION)
+#define MDS_PROT_FCTRL_ID 0x00AC13F5
+
/* Added for the subscription changes */
#define MDS_NCS_CHASSIS_ID (m_NCS_GET_NODE_ID & 0x00ff)
#define MDS_TIPC_COMMON_ID 0x01001000
diff --git a/src/mds/mds_dt_tipc.c b/src/mds/mds_dt_tipc.c
index 86b52bb..fef1c50 100644
--- a/src/mds/mds_dt_tipc.c
+++ b/src/mds/mds_dt_tipc.c
@@ -47,6 +47,7 @@
#include "mds_dt_tipc.h"
#include "mds_dt_tcp_disc.h"
#include "mds_core.h"
+#include "mds_tipc_fctrl_intf.h"
#include "mds_tipc_recvq_stats.h"
#include "base/osaf_utility.h"
#include "base/osaf_poll.h"
@@ -165,20 +166,22 @@ NCS_PATRICIA_TREE mdtm_reassembly_list;
uint32_t mdtm_global_frag_num;
const unsigned int MAX_RECV_THRESHOLD = 30;
+uint8_t gl_mds_pro_ver = MDS_PROT | MDS_VERSION;
-static bool get_tipc_port_id(int sock, uint32_t* port_id) {
+static bool get_tipc_port_id(int sock, struct tipc_portid* port_id) {
struct sockaddr_tipc addr;
socklen_t sz = sizeof(addr);
memset(&addr, 0, sizeof(addr));
- *port_id = 0;
+ port_id->node = 0;
+ port_id->ref = 0;
if (0 > getsockname(sock, (struct sockaddr *)&addr, &sz)) {
syslog(LOG_ERR, "MDTM:TIPC Failed to get socket name, err:
%s",
strerror(errno));
return false;
}
- *port_id = addr.addr.id.ref;
+ *port_id = addr.addr.id;
return true;
}
@@ -240,
