Re: [devel] [PATCH 1/1] mds: Fix mds flow control keep all messages in queue [#3123]
Hi Thuan,
ack with comments.
Thanks
Minh
On 28/11/19 6:55 pm, thuan.tran wrote:
When overflow happens, mds with flow control enabled may keep
all messages in queue if it fails to send a message when receiving
Nack or ChunkAck since no more trigger come after that.
MDS flow control should retry to send message in this scenario.
---
src/mds/mds_tipc_fctrl_portid.cc | 47 ++--
1 file changed, 32 insertions(+), 15 deletions(-)
diff --git a/src/mds/mds_tipc_fctrl_portid.cc b/src/mds/mds_tipc_fctrl_portid.cc
index 316e1ba75..d5314d5bc 100644
--- a/src/mds/mds_tipc_fctrl_portid.cc
+++ b/src/mds/mds_tipc_fctrl_portid.cc
@@ -17,6 +17,7 @@
#include "mds/mds_tipc_fctrl_portid.h"
#include "base/ncssysf_def.h"
+#include "base/osaf_time.h"
#include "mds/mds_dt.h"
#include "mds/mds_log.h"
@@ -149,23 +150,24 @@ void TipcPortId::FlushData() {
uint32_t TipcPortId::Send(uint8_t* data, uint16_t length) {
struct sockaddr_tipc server_addr;
- ssize_t send_len = 0;
- uint32_t rc = NCSCC_RC_SUCCESS;
-
memset(_addr, 0, sizeof(server_addr));
server_addr.family = AF_TIPC;
server_addr.addrtype = TIPC_ADDR_ID;
server_addr.addr.id = id_;
- send_len = sendto(bsrsock_, data, length, 0,
-(struct sockaddr *)_addr, sizeof(server_addr));
-
- if (send_len == length) {
-rc = NCSCC_RC_SUCCESS;
- } else {
-m_MDS_LOG_ERR("FCTRL: sendto() failed, Error[%s]", strerror(errno));
-rc = NCSCC_RC_FAILURE;
+ int retry = 5;
+ while (retry >= 0) {
+ssize_t send_len = sendto(bsrsock_, data, length, 0,
+ (struct sockaddr *)_addr, sizeof(server_addr));
+
+if (send_len == length) {
+ return NCSCC_RC_SUCCESS;
+} else if (retry-- > 0) {
+ assert(errno == ENOMEM || errno == ENOBUFS);
+ osaf_nanosleep();
+}
}
[Minh] It might be a good thing to make a wrapper of sendto(), since the
sendto() is currently called in fctrl_portid.cc and mds_dt_tipc.c. So we
only call the wrapper of sendto(), which handles the error code of
sendto(). I think the only EINTR code to be checked, there are a few
places in opensaf that is handling error code of sendto() which we can
take as reference.
- return rc;
+ m_MDS_LOG_ERR("FCTRL: sendto() failed, Error[%s]", strerror(errno));
+ return NCSCC_RC_FAILURE;
}
uint32_t TipcPortId::Queue(const uint8_t* data, uint16_t length,
@@ -440,13 +442,16 @@ void TipcPortId::ReceiveChunkAck(uint16_t fseq, uint16_t
chksize) {
// try to send a few pending msg
DataMessage* msg = nullptr;
uint16_t send_msg_cnt = 0;
-while (send_msg_cnt++ < chunk_size_) {
+int retry = 0;
+while (send_msg_cnt < chunk_size_) {
// find the lowest sequence unsent yet
msg = sndqueue_.FirstUnsent();
if (msg == nullptr) {
break;
} else {
if (Send(msg->msg_data_, msg->header_.msg_len_) ==
NCSCC_RC_SUCCESS) {
+retry = 0;
+send_msg_cnt++;
msg->is_sent_ = true;
m_MDS_LOG_NOTIFY("FCTRL: [me] --> [node:%x, ref:%u], "
"SndQData[fseq:%u, len:%u], "
@@ -454,6 +459,12 @@ void TipcPortId::ReceiveChunkAck(uint16_t fseq, uint16_t
chksize) {
id_.node, id_.ref,
msg->header_.fseq_, msg->header_.msg_len_,
sndwnd_.acked_.v(), sndwnd_.send_.v(), sndwnd_.nacked_space_);
+ } else if (send_msg_cnt == 0) {
+// If not retry, all messages are kept in queue
+// and no more trigger to send messages
+retry++;
+assert(retry < 100);
+continue;
[Minh] We can accept to use the assert for now, and 100 should be
defined as constant. But I do think we need a fallback mechanism, if the
socket fd is not able to send data, we can terminate the portid, and
trigger a MDS_DOWN event, ... and this could be looked in another ticket.
Also, the patch title does not seem to be right in the context of this
ticket, where we have problem of "Cannot allocate memeory", we might not
be able to send any more message (not that for all) and hit the assert.
We can say "Add retry for tipc sendto()" or you have a better
description for it.
} else {
break;
}
@@ -508,9 +519,15 @@ void TipcPortId::ReceiveNack(uint32_t mseq, uint16_t mfrag,
DataMessage* msg = sndqueue_.Find(Seq16(fseq));
if (msg != nullptr) {
// Resend the msg found
-if (Send(msg->msg_data_, msg->header_.msg_len_) == NCSCC_RC_SUCCESS) {
- msg->is_sent_ = true;
+int retry = 0;
+while (Send(msg->msg_data_, msg->header_.msg_len_) != NCSCC_RC_SUCCESS) {
+ // If not retry, all messages are kept in queue
+ // and no more trigger to send messages
+ retry++;
+ assert(retry < 100);
+ continue;
}
+msg->is_sent_ = true;
m_MDS_LOG_NOTIFY("FCTRL: [me] --> [node:%x, ref:%u], "
"RsndData[mseq:%u, mfrag:%u, fseq:%u], "
[devel] [PATCH 1/2] mds: Improve readibility [#3089]
Correct indent and reduce code lines (<80 chars) for
mds_mdtm_send_tipc() and mdtm_frag_and_send()
---
src/mds/mds_dt_tipc.c | 490 ++
1 file changed, 256 insertions(+), 234 deletions(-)
diff --git a/src/mds/mds_dt_tipc.c b/src/mds/mds_dt_tipc.c
index fdf0da7..722076f 100644
--- a/src/mds/mds_dt_tipc.c
+++ b/src/mds/mds_dt_tipc.c
@@ -2561,16 +2561,16 @@ uint32_t mds_mdtm_send_tipc(MDTM_SEND_REQ *req)
send message
*/
uint32_t status = 0;
- uint32_t sum_mds_hdr_plus_mdtm_hdr_plus_len;
+ uint32_t mds_and_mdtm_hdr_len;
uint16_t fctrl_seq_num = 0;
int version = req->msg_arch_word & 0x7;
if (version > 1) {
- sum_mds_hdr_plus_mdtm_hdr_plus_len =
+ mds_and_mdtm_hdr_len =
(SUM_MDS_HDR_PLUS_MDTM_HDR_PLUS_LEN +
gl_mds_mcm_cb->node_name_len);
} else {
/* sending message to Old version Node */
- sum_mds_hdr_plus_mdtm_hdr_plus_len =
+ mds_and_mdtm_hdr_len =
(SUM_MDS_HDR_PLUS_MDTM_HDR_PLUS_LEN - 1);
}
@@ -2598,13 +2598,13 @@ uint32_t mds_mdtm_send_tipc(MDTM_SEND_REQ *req)
/* This is exclusively for the Bcast ENC and ENC_FLAT case */
if (recv.msg.encoding == MDS_ENC_TYPE_FULL) {
ncs_dec_init_space(_uba,
- recv.msg.data.fullenc_uba.start);
+ recv.msg.data.fullenc_uba.start);
recv.msg_arch_word = req->msg_arch_word;
} else if (recv.msg.encoding == MDS_ENC_TYPE_FLAT) {
/* This case will not arise, but just to be on safe side
*/
ncs_dec_init_space(_uba,
- recv.msg.data.flat_uba.start);
+ recv.msg.data.flat_uba.start);
} else {
/* Do nothing for the DIrect buff and Copy case */
}
@@ -2620,19 +2620,18 @@ uint32_t mds_mdtm_send_tipc(MDTM_SEND_REQ *req)
uint32_t frag_seq_num = 0, node_status = 0;
node_status = m_MDS_CHECK_NCS_NODE_ID_RANGE(
- m_MDS_GET_NODE_ID_FROM_ADEST(req->adest));
+ m_MDS_GET_NODE_ID_FROM_ADEST(req->adest));
if (NCSCC_RC_SUCCESS == node_status) {
tipc_id.node = m_MDS_GET_TIPC_NODE_ID_FROM_NCS_NODE_ID(
- m_MDS_GET_NODE_ID_FROM_ADEST(req->adest));
+ m_MDS_GET_NODE_ID_FROM_ADEST(req->adest));
tipc_id.ref = (uint32_t)(req->adest);
} else {
- if (req->snd_type !=
- MDS_SENDTYPE_ACK) { /* This check is becoz in ack
- cases we are only sending the
- hdr and no data part is being
- send, so no message free ,
- fix me */
+ if (req->snd_type != MDS_SENDTYPE_ACK) {
+ /* This check is becoz in ack cases we are only
+* sending the hdr and no data part is being
+* send, so no message free. fix me
+*/
mdtm_free_reassem_msg_mem(>msg);
}
return NCSCC_RC_FAILURE;
@@ -2643,43 +2642,45 @@ uint32_t mds_mdtm_send_tipc(MDTM_SEND_REQ *req)
/* Only for the ack and not for any other message */
if (req->snd_type == MDS_SENDTYPE_ACK ||
req->snd_type == MDS_SENDTYPE_RACK) {
- uint8_t len = sum_mds_hdr_plus_mdtm_hdr_plus_len;
+ uint8_t len = mds_and_mdtm_hdr_len;
uint8_t buffer_ack[len];
/* Add mds_hdr */
- if (NCSCC_RC_SUCCESS !=
- mdtm_add_mds_hdr(buffer_ack, req)) {
+ if (mdtm_add_mds_hdr(buffer_ack, req)
+ != NCSCC_RC_SUCCESS) {
+ return NCSCC_RC_FAILURE;
+ }
+ /* if sndqueue is capable, then obtain the current
+* sending seq
+*/
+ if (mds_tipc_fctrl_sndqueue_capable(tipc_id,
+ _seq_num) == NCSCC_RC_FAILURE){
+ m_MDS_LOG_ERR("FCTRL: Failed to send message"
+ " len :%d", len);
return NCSCC_RC_FAILURE;
}
-
[devel] [PATCH 0/2] Review Request for mds: Avoid message reallocation [#3089] V3
Summary: mds: Avoid message reallocation [#3089] Review request for Ticket(s): 3089 Peer Reviewer(s): Thuan, Vu, Gary Pull request to: *** LIST THE PERSON WITH PUSH ACCESS HERE *** Affected branch(es): develop Development branch: ticket-3089 Base revision: 8e07c19aed63c249f4e7fa8470270d2de1a56046 Personal repository: git://git.code.sf.net/u/minh-chau/review Impacted area Impact y/n Docsn Build systemn RPM/packaging n Configuration files n Startup scripts n SAF servicesn OpenSAF servicesn Core libraries y Samples n Tests n Other n NOTE: Patch(es) contain lines longer than 80 characers Comments (indicate scope for each "y" above): - *** EXPLAIN/COMMENT THE PATCH SERIES HERE *** revision d3bdf53e99523785cdc932d62b25267ea900c643 Author: Minh Chau Date: Thu, 28 Nov 2019 21:08:50 +1100 mds: Avoid message reallocation [#3089] The patch avoids message reallocation if the message is in retransmission queue revision 7be0f5404ebb8ec5b8752813899d6aefd1ef6c33 Author: Minh Chau Date: Thu, 28 Nov 2019 21:08:38 +1100 mds: Improve readibility [#3089] Correct indent and reduce code lines (<80 chars) for mds_mdtm_send_tipc() and mdtm_frag_and_send() Complete diffstat: -- src/mds/mds_dt_tipc.c| 534 +-- src/mds/mds_tipc_fctrl_intf.cc | 6 +- src/mds/mds_tipc_fctrl_intf.h| 4 +- src/mds/mds_tipc_fctrl_msg.cc| 2 +- src/mds/mds_tipc_fctrl_portid.cc | 9 +- 5 files changed, 294 insertions(+), 261 deletions(-) Testing Commands: - *** LIST THE COMMAND LINE TOOLS/STEPS TO TEST YOUR CHANGES *** Testing, Expected Results: -- *** PASTE COMMAND OUTPUTS / TEST RESULTS *** Conditions of Submission: - *** HOW MANY DAYS BEFORE PUSHING, CONSENSUS ETC *** Arch Built StartedLinux distro --- mipsn n mips64 n n x86 n n x86_64 n n 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 2/2] mds: Avoid message reallocation [#3089]
The patch avoids message reallocation if the message is in
retransmission queue
---
src/mds/mds_dt_tipc.c| 68 +++-
src/mds/mds_tipc_fctrl_intf.cc | 6 ++--
src/mds/mds_tipc_fctrl_intf.h| 4 +--
src/mds/mds_tipc_fctrl_msg.cc| 2 +-
src/mds/mds_tipc_fctrl_portid.cc | 9 ++
5 files changed, 50 insertions(+), 39 deletions(-)
diff --git a/src/mds/mds_dt_tipc.c b/src/mds/mds_dt_tipc.c
index 722076f..3d4f468 100644
--- a/src/mds/mds_dt_tipc.c
+++ b/src/mds/mds_dt_tipc.c
@@ -120,7 +120,7 @@ uint32_t mds_mdtm_send_tipc(MDTM_SEND_REQ *req);
/* Tipc actual send, can be made as Macro even*/
static uint32_t mdtm_sendto(uint8_t *buffer, uint16_t buff_len,
- struct tipc_portid tipc_id);
+ struct tipc_portid tipc_id, uint8_t *is_queued);
static uint32_t mdtm_mcast_sendto(void *buffer, size_t size,
const MDTM_SEND_REQ *req);
@@ -2643,7 +2643,8 @@ uint32_t mds_mdtm_send_tipc(MDTM_SEND_REQ *req)
if (req->snd_type == MDS_SENDTYPE_ACK ||
req->snd_type == MDS_SENDTYPE_RACK) {
uint8_t len = mds_and_mdtm_hdr_len;
- uint8_t buffer_ack[len];
+ uint8_t *buffer_ack = calloc(1, len);
+ uint8_t is_queued = 0;
/* Add mds_hdr */
if (mdtm_add_mds_hdr(buffer_ack, req)
@@ -2657,18 +2658,24 @@ uint32_t mds_mdtm_send_tipc(MDTM_SEND_REQ *req)
_seq_num) == NCSCC_RC_FAILURE){
m_MDS_LOG_ERR("FCTRL: Failed to send message"
" len :%d", len);
+ free(buffer_ack);
return NCSCC_RC_FAILURE;
}
/* Add frag_hdr */
if (mdtm_add_frag_hdr(buffer_ack, len, frag_seq_num,
0, fctrl_seq_num) != NCSCC_RC_SUCCESS) {
+ free(buffer_ack);
return NCSCC_RC_FAILURE;
}
m_MDS_LOG_DBG("MDTM:Sending message with Service"
" Seqno=%d, TO Dest_Tipc_id=<0x%08x:%u> ",
req->svc_seq_num, tipc_id.node, tipc_id.ref);
- return mdtm_sendto(buffer_ack, len, tipc_id);
+ status = mdtm_sendto(buffer_ack, len, tipc_id,
+ _queued);
+ if (is_queued == 0)
+ free(buffer_ack);
+ return status;
}
if (req->msg.encoding == MDS_ENC_TYPE_FLAT) {
@@ -2730,6 +2737,7 @@ uint32_t mds_mdtm_send_tipc(MDTM_SEND_REQ *req)
} else {
uint8_t *p8;
uint8_t *body = NULL;
+ uint8_t is_queued = 0;
body = calloc(1, len +
mds_and_mdtm_hdr_len);
@@ -2806,8 +2814,8 @@ uint32_t mds_mdtm_send_tipc(MDTM_SEND_REQ *req)
req->dest_svc_id);
return NCSCC_RC_FAILURE;
}
- if (mdtm_mcast_sendto(body, len, req)
- != NCSCC_RC_SUCCESS) {
+ status = mdtm_mcast_sendto(body, len,
req);
+ if (status != NCSCC_RC_SUCCESS) {
m_MDS_LOG_ERR("MDTM: Failed to"
" send Multicast"
" message Data
lenght=%d"
@@ -2819,24 +2827,20 @@ uint32_t mds_mdtm_send_tipc(MDTM_SEND_REQ *req)
get_svc_names(req->dest_svc_id),
req->dest_svc_id,
strerror(errno));
- m_MMGR_FREE_BUFR_LIST(usrbuf);
- free(body);
- return NCSCC_RC_FAILURE;
}
} else {
- if (mdtm_sendto(body, len, tipc_id)
- != NCSCC_RC_SUCCESS) {
+ status = mdtm_sendto(body, len,
+ tipc_id, _queued);
+ if (status !=
[devel] [PATCH 1/5] log: improve the resilience of log service [#3116]
In order to improve resilience of OpenSAF LOG service when underlying file system is unresponsive, a queue is introduced to hold async write request up to an configurable time that is around 15 - 30 seconds. The readiness of the I/O thread will periodically check, and if it turns to ready state, the front element will go first. Returns SA_AIS_ERR_TRY_AGAIN to client if the element stays in the queue longer than the setting time. The queue capacity and the resilient time are configurable via the attributes: `logMaxPendingWriteRequests` and `logResilienceTimeout`. In default, this feature is disabled to keep log server backward compatible. --- src/log/Makefile.am | 21 +- src/log/config/logsv_classes.xml | 43 ++- src/log/logd/lgs_cache.cc| 469 +++ src/log/logd/lgs_cache.h | 287 +++ src/log/logd/lgs_config.cc | 78 - src/log/logd/lgs_config.h| 10 +- src/log/logd/lgs_evt.cc | 161 +++ src/log/logd/lgs_evt.h | 10 + src/log/logd/lgs_file.cc | 8 +- src/log/logd/lgs_filehdl.cc | 58 ++-- src/log/logd/lgs_imm.cc | 40 ++- src/log/logd/lgs_main.cc | 24 +- src/log/logd/lgs_mbcsv.cc| 447 +++-- src/log/logd/lgs_mbcsv.h | 19 +- src/log/logd/lgs_mbcsv_cache.cc | 372 src/log/logd/lgs_mbcsv_cache.h | 110 src/log/logd/lgs_mbcsv_v1.cc | 1 + src/log/logd/lgs_mbcsv_v2.cc | 2 + 18 files changed, 1889 insertions(+), 271 deletions(-) create mode 100644 src/log/logd/lgs_cache.cc create mode 100644 src/log/logd/lgs_cache.h create mode 100644 src/log/logd/lgs_mbcsv_cache.cc create mode 100644 src/log/logd/lgs_mbcsv_cache.h diff --git a/src/log/Makefile.am b/src/log/Makefile.am index f63a4a053..3367ef4f6 100644 --- a/src/log/Makefile.am +++ b/src/log/Makefile.am @@ -95,7 +95,9 @@ noinst_HEADERS += \ src/log/logd/lgs_nildest.h \ src/log/logd/lgs_unixsock_dest.h \ src/log/logd/lgs_common.h \ - src/log/logd/lgs_amf.h + src/log/logd/lgs_amf.h \ + src/log/logd/lgs_cache.h \ + src/log/logd/lgs_mbcsv_cache.h bin_PROGRAMS += bin/saflogger @@ -123,6 +125,15 @@ bin_osaflogd_CPPFLAGS = \ -DSA_EXTENDED_NAME_SOURCE \ $(AM_CPPFLAGS) +# Enable this flag to simulate the case that file system is unresponsive +# during write log record. Mainly for testing the following enhancement: +# log: improve the resilience of log service [#3116]. +# When enabled, log handle thread will be suspended 17 seconds every 02 write +# requests and only take affect if the `logMaxPendingWriteRequests` is set +# to an non-zero value. +bin_osaflogd_CPPFLAGS += -DSIMULATE_NFS_UNRESPONSE + + bin_osaflogd_SOURCES = \ src/log/logd/lgs_amf.cc \ src/log/logd/lgs_clm.cc \ @@ -147,7 +158,9 @@ bin_osaflogd_SOURCES = \ src/log/logd/lgs_util.cc \ src/log/logd/lgs_dest.cc \ src/log/logd/lgs_nildest.cc \ - src/log/logd/lgs_unixsock_dest.cc + src/log/logd/lgs_unixsock_dest.cc \ + src/log/logd/lgs_cache.cc \ + src/log/logd/lgs_mbcsv_cache.cc bin_osaflogd_LDADD = \ lib/libosaf_common.la \ @@ -183,6 +196,10 @@ bin_logtest_CPPFLAGS = \ -DSA_EXTENDED_NAME_SOURCE \ $(AM_CPPFLAGS) +# Enable this flag to add test cases for following enhancement: +# log: improve the resilience of log service [#3116]. +bin_logtest_CPPFLAGS += -DSIMULATE_NFS_UNRESPONSE + bin_logtest_SOURCES = \ src/log/apitest/logtest.c \ src/log/apitest/logutil.c \ diff --git a/src/log/config/logsv_classes.xml b/src/log/config/logsv_classes.xml index 9359823ff..084e8915d 100644 --- a/src/log/config/logsv_classes.xml +++ b/src/log/config/logsv_classes.xml @@ -195,7 +195,7 @@ to ensure that default global values in the implementation are also changed acco SA_UINT32_T SA_CONFIG SA_WRITABLE -1024 + 1024 logStreamFileFormat @@ -208,42 +208,42 @@ to ensure that default global values in the implementation are also changed acco SA_UINT32_T SA_CONFIG SA_WRITABLE -0 + 0 logStreamSystemLowLimit SA_UINT32_T SA_CONFIG SA_WRITABLE -0 + 0 logStreamAppHighLimit SA_UINT32_T SA_CONFIG SA_WRITABLE -0 + 0 logStreamAppLowLimit SA_UINT32_T
[devel] [PATCH 5/5] log: add test cases of improving the log resilience [#3116]
Adding 08 new test cases into 02 suites:
1) Suite 20 with 07 test cases, including:
- Test changing queue size & resilient timeout;
- Test if a write async is dropped if its timeout setting is overdue,
also verify if log server has kept the request in proper time.
- Test if getting write callback right away if the cache is full.
- Test if the cache is fully and correctly synced with standby.
2) Suite 21 with one test case:
Test if LOG agent notifies all lost invocation to log client.
As the suite 21 requires manual interaction, it is put into
'extended' tests. Only run with option '-e'.
---
src/log/Makefile.am | 3 +-
src/log/apitest/logtest.c | 7 +
src/log/apitest/logtest.h | 7 +-
src/log/apitest/logutil.c | 14 +-
src/log/apitest/tet_log_runtime_cfgobj.c | 2 +-
.../apitest/tet_saLogWriteLogAsync_cache.c| 648 ++
6 files changed, 667 insertions(+), 14 deletions(-)
create mode 100644 src/log/apitest/tet_saLogWriteLogAsync_cache.c
diff --git a/src/log/Makefile.am b/src/log/Makefile.am
index 3367ef4f6..3ec03c097 100644
--- a/src/log/Makefile.am
+++ b/src/log/Makefile.am
@@ -224,7 +224,8 @@ bin_logtest_SOURCES = \
src/log/apitest/tet_log_longDN.c \
src/log/apitest/tet_Log_clm.c \
src/log/apitest/tet_cfg_destination.c \
- src/log/apitest/tet_multiple_thread.c
+ src/log/apitest/tet_multiple_thread.c \
+ src/log/apitest/tet_saLogWriteLogAsync_cache.c
bin_logtest_LDADD = \
lib/libapitest.la \
diff --git a/src/log/apitest/logtest.c b/src/log/apitest/logtest.c
index aabd1e578..149d27d93 100644
--- a/src/log/apitest/logtest.c
+++ b/src/log/apitest/logtest.c
@@ -96,6 +96,7 @@ SaLogCallbacksT logCallbacks = {NULL, NULL, NULL};
SaInvocationT invocation = 0;
SaSelectionObjectT selectionObject;
char log_root_path[PATH_MAX];
+SaLogAckFlagsT ack_flags = 0;
void init_logrootpath(void)
{
@@ -465,6 +466,9 @@ int main(int argc, char **argv)
add_suite_14();
add_suite_15();
add_suite_16();
+#ifdef SIMULATE_NFS_UNRESPONSE
+ add_suite_21();
+#endif
test_list();
exit(0);
case 'e':
@@ -493,6 +497,9 @@ int main(int argc, char **argv)
add_suite_14();
add_suite_15();
add_suite_16();
+#ifdef SIMULATE_NFS_UNRESPONSE
+ add_suite_21();
+#endif
break;
case 'v':
if (silent_flg == true) {
diff --git a/src/log/apitest/logtest.h b/src/log/apitest/logtest.h
index 68f9df608..e04492086 100644
--- a/src/log/apitest/logtest.h
+++ b/src/log/apitest/logtest.h
@@ -76,7 +76,7 @@ extern SaSelectionObjectT selectionObject;
extern SaNameT logSvcUsrName;
extern SaLogRecordT genLogRecord;
extern char log_root_path[];
-
+extern SaLogAckFlagsT ack_flags;
const static SaVersionT kLogVersion = {'A', 0x02, 0x03};
const static SaVersionT kImmVersion = {'A', 02, 11};
@@ -105,6 +105,11 @@ void add_suite_12(void);
void add_suite_14();
void add_suite_15();
void add_suite_16();
+
+#ifdef SIMULATE_NFS_UNRESPONSE
+void add_suite_21();
+#endif
+
int get_active_sc(void);
int get_attr_value(SaNameT *inObjName, char *inAttr, void *outValue);
diff --git a/src/log/apitest/logutil.c b/src/log/apitest/logutil.c
index 59d255515..d3e0c6297 100644
--- a/src/log/apitest/logutil.c
+++ b/src/log/apitest/logutil.c
@@ -52,15 +52,7 @@ void cond_check(void)
int systemCall(const char *command)
{
int rc = system(command);
- if (rc == -1) {
- fprintf(stderr, "system() retuned -1 Failed \n");
- } else {
- rc = WEXITSTATUS(rc);
- if (rc != 0)
- fprintf(stderr, " Failed in command: %s \n", command);
- }
-
- return rc;
+ return WEXITSTATUS(rc);
}
/*
@@ -144,8 +136,8 @@ logAppStreamOpen(const SaNameT *logStreamName,
*/
SaAisErrorT logWriteAsync(const SaLogRecordT *logRecord)
{
- SaAisErrorT rc =
- saLogWriteLogAsync(logStreamHandle, invocation, 0, logRecord);
+ SaAisErrorT rc = saLogWriteLogAsync(logStreamHandle, invocation,
+ ack_flags, logRecord);
unsigned int nTries = 1;
while (rc == SA_AIS_ERR_TRY_AGAIN && nTries < logProfile.nTries) {
usleep(logProfile.retryInterval * 1000);
diff --git a/src/log/apitest/tet_log_runtime_cfgobj.c
b/src/log/apitest/tet_log_runtime_cfgobj.c
index ae83e655e..98aab7f5f 100644
--- a/src/log/apitest/tet_log_runtime_cfgobj.c
+++ b/src/log/apitest/tet_log_runtime_cfgobj.c
@@ -84,7 +84,7 @@ void log_rt_cf_obj_compare(void)
* 1 more attribute than the configuration object
*/
r_cnt--;
- if (c_cnt != r_cnt) {
+ if (c_cnt
[devel] [PATCH 2/5] log: notify all lost log records when cluster goes to headless [#3116]
This change introduces a light list keeping all invocations that not yet
get the acknowledgement from log server. If the server is disappeared
in case of headless, log agent will notify all lost invocations to log client
with error code SA_AIS_ERR_TRY_AGAIN.
---
src/log/agent/lga_agent.cc | 2 ++
src/log/agent/lga_client.cc | 8 +++-
src/log/agent/lga_client.h | 34 ++
3 files changed, 43 insertions(+), 1 deletion(-)
diff --git a/src/log/agent/lga_agent.cc b/src/log/agent/lga_agent.cc
index 1000bb3fd..0049323c8 100644
--- a/src/log/agent/lga_agent.cc
+++ b/src/log/agent/lga_agent.cc
@@ -1296,6 +1296,8 @@ SaAisErrorT
LogAgent::saLogWriteLogAsync(SaLogStreamHandleT logStreamHandle,
if (NCSCC_RC_SUCCESS !=
lga_mds_msg_async_send(, MDS_SEND_PRIORITY_MEDIUM)) {
ais_rc = SA_AIS_ERR_TRY_AGAIN;
+ } else {
+client->KeepTrack(invocation, ackFlags);
}
return ais_rc;
diff --git a/src/log/agent/lga_client.cc b/src/log/agent/lga_client.cc
index 386c84929..cdc54904a 100644
--- a/src/log/agent/lga_client.cc
+++ b/src/log/agent/lga_client.cc
@@ -84,7 +84,9 @@ LogClient::~LogClient() {
for (auto& stream : stream_list_) {
if (stream != nullptr) delete stream;
}
+
stream_list_.clear();
+ unacked_invocations_.clear();
// Free the client handle allocated to this log client
if (handle_ != 0) {
@@ -129,9 +131,11 @@ void LogClient::InvokeCallback(const lgsv_msg_t* msg) {
// Invoke the corresponding callback
switch (cbk_info->type) {
case LGSV_WRITE_LOG_CALLBACK_IND: {
- if (callbacks_.saLogWriteLogCallback)
+ if (callbacks_.saLogWriteLogCallback) {
+RemoveTrack(cbk_info->inv);
callbacks_.saLogWriteLogCallback(cbk_info->inv,
cbk_info->write_cbk.error);
+ }
} break;
case LGSV_SEVERITY_FILTER_CALLBACK: {
@@ -395,6 +399,8 @@ void LogClient::NoLogServer() {
// When LOG server restart from headless, Log agent will do recover them.
stream->SetRecoveryFlag(false);
}
+
+ NotifyClientAboutLostInvocations();
}
uint32_t LogClient::SendMsgToMbx(lgsv_msg_t* msg, MDS_SEND_PRIORITY_TYPE prio)
{
diff --git a/src/log/agent/lga_client.h b/src/log/agent/lga_client.h
index d7060cc13..f5fa6faa4 100644
--- a/src/log/agent/lga_client.h
+++ b/src/log/agent/lga_client.h
@@ -20,6 +20,7 @@
#include
#include
+#include
#include
#include
#include "base/mutex.h"
@@ -169,6 +170,35 @@ class LogClient {
return ref_counter_object_.RestoreRefCounter(caller, value, updated);
}
+ // Track the number of write requests sent to log server but not yet
+ // get acknowledgement from it.
+ void KeepTrack(SaInvocationT inv, uint32_t ack_flags) {
+if (ack_flags != SA_LOG_RECORD_WRITE_ACK) return;
+unacked_invocations_.push_back(inv);
+ }
+
+ // Got an acknowledgment, so remove from the track list.
+ void RemoveTrack(SaInvocationT inv) { unacked_invocations_.remove(inv); }
+
+ void NotifyClientAboutLostInvocations() {
+for (const auto& i : unacked_invocations_) {
+ TRACE("The write async with this invocation %lld has been lost", i);
+ // the below memory will be freed by lga_msg_destroy(cbk_msg)
+ // after done processing with this msg from the mailbox.
+ lgsv_msg_t* msg = static_cast(malloc(sizeof(lgsv_msg_t)));
+ assert(msg && "Failed to allocate memory for lgsv_msg_t");
+ memset(msg, 0, sizeof(lgsv_msg_t));
+ msg->type = LGSV_LGS_CBK_MSG;
+ msg->info.cbk_info.type = LGSV_WRITE_LOG_CALLBACK_IND;
+ msg->info.cbk_info.lgs_client_id = client_id_;
+ msg->info.cbk_info.write_cbk.error = SA_AIS_ERR_TRY_AGAIN;
+ msg->info.cbk_info.inv = i;
+
+ SendMsgToMbx(msg, MDS_SEND_PRIORITY_HIGH);
+}
+unacked_invocations_.clear();
+ }
+
// true if the client is successfully done recovery.
// or the client has just borned.
// Introduce this method to avoid locking the successful recovered client
@@ -256,6 +286,10 @@ class LogClient {
// Hold all log streams belong to @this client
std::vector stream_list_;
+ // Hold all invocations that not yet get acknowledgement from log server.
+ // If cluster goes to headless, log agent will inform to log client with
+ // SA_AIS_ERR_TRY_AGAIN code for these invocations.
+ std::list unacked_invocations_{};
// LOG handle (derived from hdl-mngr)
SaLogHandleT handle_;
--
2.17.1
___
Opensaf-devel mailing list
Opensaf-devel@lists.sourceforge.net
https://lists.sourceforge.net/lists/listinfo/opensaf-devel
[devel] [PATCH 4/5] log: update README file for improvement of log resilience [#3116]
--- src/log/README | 38 ++ 1 file changed, 38 insertions(+) diff --git a/src/log/README b/src/log/README index b83d472e4..ab96a8157 100644 --- a/src/log/README +++ b/src/log/README @@ -764,3 +764,41 @@ on AMF role is unnecessary delay the CLM state of a Node (CLM state will available as soon as CLM started), so LGS is a taking AVD Up event as trigger to do CLM initialize. + +4. Improve the resilience of OpenSAF LOG service (#3116) +- +When the file system is unresponsive, log client gets try-again from write +callback very shortly after I/O timeout reaches the setting; the value of I/O +timeout is configurable via the attribute logFileIoTimeout within this valid +range [500ms – 5000ms]. This is legacy behavior. + +This ticket improves the resilience of LOG service, so that log service can +cache async write requests up to an configurable time that is around 15-30 +seconds before returning status to log client via write async callback. + +The cache size is configurable via a new attribute `logMaxPendingWriteRequests`. +Default value is zero (0) - means this feature is disabled. The valid range is +[current queue size - 1000]. To know what is the current size of the queue, +fetching the value of pure runtime attribute `logCurrentPendingWriteRequests` +of `OpenSafLogCurrentConfig` class. When the cache size reaches the limit, +all coming requests will get acknowledgement right away with +SA_AIS_ERR_TRY_AGAIN. + +The resilient timeout can also be configurable via a new attribute +`logResilienceTimeout`. The valid range is [15-30] seconds. When a pending write +async can be dropped and removed from the queue in cases: +a) Stays in the queue longer than the given resilient timeout. +b) The targeting stream has been closed. + +The queue is always kept in sync with standby. + +Besides, log agent has a light list keeping track all invocations which not yet +get acknowledgements from log server. If cluster goes to headless; in other +words, log server is disappeared and all cached data has been lost, +log agent (library) will notify all lost invocations to log client via write +async callback with SA_AIS_ERR_TRY_AGAIN error code. + +To test this feature, a gcc flag is added during compile time to simulate +the case the underlying file system is unresponsive, and it only takes +affect when the cache size is given to an non-zero value. With that, +the I/O thread will sleep *16 seconds* every 02 write requests. \ No newline at end of file -- 2.17.1 ___ Opensaf-devel mailing list Opensaf-devel@lists.sourceforge.net https://lists.sourceforge.net/lists/listinfo/opensaf-devel
[devel] [PATCH 0/5] Review Request for log: improve the resilience of log service [#3116]
Summary: log: improve the resilience of log service [#3116] Review request for Ticket(s): 3116 Peer Reviewer(s): Lennart, Gary, Minh Pull request to: *** LIST THE PERSON WITH PUSH ACCESS HERE *** Affected branch(es): develop Development branch: ticket-3116 Base revision: 8e07c19aed63c249f4e7fa8470270d2de1a56046 Personal repository: git://git.code.sf.net/u/winhvu/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 NOTE: Patch(es) contain lines longer than 80 characers Comments (indicate scope for each "y" above): - *** EXPLAIN/COMMENT THE PATCH SERIES HERE *** revision 2643f8f829bb7ba638193df19cc6f20f86acb497 Author: Vu Minh Nguyen Date: Thu, 28 Nov 2019 15:19:58 +0700 log: add test cases of improving the log resilience [#3116] Adding 08 new test cases into 02 suites: 1) Suite 20 with 07 test cases, including: - Test changing queue size & resilient timeout; - Test if a write async is dropped if its timeout setting is overdue, also verify if log server has kept the request in proper time. - Test if getting write callback right away if the cache is full. - Test if the cache is fully and correctly synced with standby. 2) Suite 21 with one test case: Test if LOG agent notifies all lost invocation to log client. As the suite 21 requires manual interaction, it is put into 'extended' tests. Only run with option '-e'. revision 2a714a95f9f714684840047ae1e02d7c11bca32f Author: Vu Minh Nguyen Date: Thu, 28 Nov 2019 15:19:58 +0700 log: update README file for improvement of log resilience [#3116] revision a103db0f56db32a1abe6f43eda7f06d48b54d469 Author: Vu Minh Nguyen Date: Thu, 28 Nov 2019 15:19:58 +0700 saflogger: make timeout waiting for getting acknowledgment configurable [#3116] Introducing a new option `-t second` or `--timeout=second` to let user input his desired timeout of waiting for write async acknowledgment. Default timeout is 20 seconds to keep saflogger backward compatible. revision 54d46abd5c433ee884294ada6b277b04042f444f Author: Vu Minh Nguyen Date: Thu, 28 Nov 2019 15:19:58 +0700 log: notify all lost log records when cluster goes to headless [#3116] This change introduces a light list keeping all invocations that not yet get the acknowledgement from log server. If the server is disappeared in case of headless, log agent will notify all lost invocations to log client with error code SA_AIS_ERR_TRY_AGAIN. revision 6260044621a2ecfad70bd5eeece3e811dc032188 Author: Vu Minh Nguyen Date: Thu, 28 Nov 2019 15:19:58 +0700 log: improve the resilience of log service [#3116] In order to improve resilience of OpenSAF LOG service when underlying file system is unresponsive, a queue is introduced to hold async write request up to an configurable time that is around 15 - 30 seconds. The readiness of the I/O thread will periodically check, and if it turns to ready state, the front element will go first. Returns SA_AIS_ERR_TRY_AGAIN to client if the element stays in the queue longer than the setting time. The queue capacity and the resilient time are configurable via the attributes: `logMaxPendingWriteRequests` and `logResilienceTimeout`. In default, this feature is disabled to keep log server backward compatible. Added Files: src/log/apitest/tet_saLogWriteLogAsync_cache.c src/log/logd/lgs_cache.cc src/log/logd/lgs_cache.h src/log/logd/lgs_mbcsv_cache.cc src/log/logd/lgs_mbcsv_cache.h Complete diffstat: -- src/log/Makefile.am| 24 +- src/log/README | 38 ++ src/log/agent/lga_agent.cc | 2 + src/log/agent/lga_client.cc| 8 +- src/log/agent/lga_client.h | 34 ++ src/log/apitest/logtest.c | 7 + src/log/apitest/logtest.h | 7 +- src/log/apitest/logutil.c | 14 +- src/log/apitest/tet_log_runtime_cfgobj.c | 2 +- src/log/apitest/tet_saLogWriteLogAsync_cache.c | 648 + src/log/config/logsv_classes.xml | 43 +- src/log/logd/lgs_cache.cc | 469 ++ src/log/logd/lgs_cache.h | 287 +++ src/log/logd/lgs_config.cc | 78 ++- src/log/logd/lgs_config.h | 10 +- src/log/logd/lgs_evt.cc| 161 ++ src/log/logd/lgs_evt.h | 10 + src/log/logd/lgs_file.cc | 8 +- src/log/logd/lgs_filehdl.cc| 58 +-- src/log/logd/lgs_imm.cc
