Hi Canh,
ack with comments, please see with [M]
Thanks
Minh
On 9/1/19 8:22 pm, Canh Van Truong wrote:
When writing the notificaion fail with TRY_AGAIN in callback, the notificaion
is pushed
again to the list. If this happens for long time, the list is going to be very
big.
This cause NTFD take time to process writing all the notification in the list
and
the request from NTFA come this time may be timeout.
The patch does:
- Limit the logger buffer
- Provide the env variable that user can set the value of the limit
- Return TRY_AGAIN error in case the limit of buffer is reached and write
all the
notifications in the buffer to the log file. The current of notification
isn't
written to log file.
---
src/ntf/README | 10 ++++
src/ntf/ntfd/NtfAdmin.cc | 44 +++++++++-----
src/ntf/ntfd/NtfLogger.cc | 149 +++++++++++++++++++++++++++-------------------
src/ntf/ntfd/NtfLogger.h | 11 +++-
src/ntf/ntfd/ntfd.conf | 10 ++++
src/ntf/ntfd/ntfs.h | 2 +
6 files changed, 147 insertions(+), 79 deletions(-)
diff --git a/src/ntf/README b/src/ntf/README
index 6dd5173e1..5bf670647 100644
--- a/src/ntf/README
+++ b/src/ntf/README
@@ -233,6 +233,16 @@ NTFSV_ENV_CACHE_SIZE
The size of the notification cache in the NTF server processes running on the
Controller nodes.
The default value is 10000 notification.
+NTFSV_LOGGER_BUFFER_CAPACITY
+
+The logger buffer is used to store the notification if writing notification
+to log file fail. This variable is set for limit of logger buffer size in
+NTFD. If the logger buffer is full and NTFD receives new notification,
+the TRY_AGAIN error is returned to user. The limit should be set with relevant
+value to avoid congestion in NTFD. Because if this value is set too big and
+writing notification is fail for long time, NTF has to write a big number of
+notifications whenever handling sending notification request and that will
delay
+to handle other requests come to NTFD. The value of variable is from 10 to
5000.
for debug see DEBUG.
diff --git a/src/ntf/ntfd/NtfAdmin.cc b/src/ntf/ntfd/NtfAdmin.cc
index 2cb99457c..6c2d69b43 100644
--- a/src/ntf/ntfd/NtfAdmin.cc
+++ b/src/ntf/ntfd/NtfAdmin.cc
@@ -193,19 +193,32 @@ void NtfAdmin::processNotification(unsigned int clientId,
notificationId, notificationType,
(unsigned int)notificationMap.size());
- // log the notification. Callback from SAF log will confirm later.
- logger.log(notification, activeController());
-
- /* send notification to standby */
- sendNotificationUpdate(clientId, notification->getNotInfo());
+ if ((logger.isLoggerBufferFull() == true) &&
+ (logger.isAlarmNotification(notification) == true)) {
+ NtfClient *client = getClient(clientId);
+ MDS_DEST dest = client->getMdsDest();
+ LOG_WA("The logger buffer is full. Check if there is issue in writing");
+ if (activeController())
+ notfication_result_lib(SA_AIS_ERR_TRY_AGAIN, notificationId,
+ mdsCtxt, dest);
+ } else {
+ /* send notification to standby */
+ sendNotificationUpdate(clientId, notification->getNotInfo());
- ClientMap::iterator pos;
- for (pos = clientMap.begin(); pos != clientMap.end(); pos++) {
- NtfClient *client = pos->second;
- client->notificationReceived(clientId, notification, mdsCtxt);
+ ClientMap::iterator pos;
+ for (pos = clientMap.begin(); pos != clientMap.end(); pos++) {
+ NtfClient *client = pos->second;
+ client->notificationReceived(clientId, notification, mdsCtxt);
+ }
}
- /* remove notification if sent to all subscribers and logged */
+ // Log the notification. Callback from SAF log will confirm later.
+ if (activeController())
+ logger.log(notification);
+ // Add the notification to Reader list
+ logger.addNotificationToReaderList(notification);
+
+ // Remove the notification if it is sent to all subscribers and logged
if (notification->isSubscriptionListEmpty() && notification->loggedOk()) {
NotificationMap::iterator posNot;
posNot = notificationMap.find(notificationId);
[M]: If ntfd decides to return TRY_AGAIN, then the notification should
not be added for readers to read, and for subscription checking, etc....
I think it looks like this
if (the buffer is not empty) {
// try to flush all pending log
}
if (the buffer is still full) {
// return try again
} else {
// add this to buffer, checkpoint, add to reader lists, .... as normal
}
@@ -341,9 +354,9 @@ void NtfAdmin::notificationReceivedColdSync(
TRACE_LEAVE();
}
/**
- * A cached notification is received in Cold Sync.
- * This cached notification will be marked as logged, and stored
- * only in NtfLogger class to serve the reader.
+ * A cached notifications are received in Cold Sync.
+ * This cached notifications are stored in NtfLogger
+ * class to serve the reader.
*
* @param clientId Node-wide unique id for the client who sent the
* notification.
@@ -363,8 +376,7 @@ void NtfAdmin::cachedNotificationReceivedColdSync(
TRACE_2("cached notification %u received", (unsigned int)notificationId);
NtfSmartPtr notification(new NtfNotification(notificationId,
notificationType, sendNotInfo));
- notification->notificationLoggedConfirmed();
- logger.log(notification, false);
+ logger.addNotificationToReaderList(notification);
TRACE_LEAVE();
}
@@ -706,7 +718,7 @@ void NtfAdmin::checkNotificationList() {
if (notification->loggedOk() == false) {
/* When reader API works check if already logged */
- logger.log(notification, true);
+ logger.log(notification);
}
if (!notification->isSubscriptionListEmpty()) {
TRACE_2("list not empty check subscriptions for %llu",
diff --git a/src/ntf/ntfd/NtfLogger.cc b/src/ntf/ntfd/NtfLogger.cc
index fd17c58a2..a7fa90c2a 100644
--- a/src/ntf/ntfd/NtfLogger.cc
+++ b/src/ntf/ntfd/NtfLogger.cc
@@ -75,6 +75,22 @@ NtfLogger::NtfLogger() : readCounter(0) {
LOG_ER("initialize saflog failed exiting...");
exit(EXIT_FAILURE);
}
+
+ // Get the capacity of logger buffer from the environment variable.
+ // The value should be from 10 to 5000. If this value is too big,
+ // NTFD will take long time to process "saNtfNotificationSend request"
+ char *tmp_buffer_capacity = getenv("NTFSV_LOGGER_BUFFER_CAPACITY");
+ if (tmp_buffer_capacity != nullptr) {
+ logger_buffer_capacity = atoi(tmp_buffer_capacity);
+ if (logger_buffer_capacity < NTFSV_LOGGER_BUFFER_CAPACITY_DEFAULT)
+ logger_buffer_capacity = NTFSV_LOGGER_BUFFER_CAPACITY_DEFAULT;
+ else if (logger_buffer_capacity > NTFSV_LOGGER_BUFFER_CAPACITY_MAX)
+ logger_buffer_capacity = NTFSV_LOGGER_BUFFER_CAPACITY_MAX;
+ TRACE("The capacity of logger buffer is configured: %d",
+ logger_buffer_capacity);
+ } else {
+ logger_buffer_capacity = NTFSV_LOGGER_BUFFER_CAPACITY_DEFAULT;
+ }
}
[M]: I think you can use base::GetEnv, it would look a bit simpler and
not using atoi
/* Callbacks */
@@ -116,62 +132,43 @@ void saLogWriteLogCallback(SaInvocationT invocation,
SaAisErrorT error) {
TRACE_LEAVE();
}
-void NtfLogger::log(NtfSmartPtr& notif, bool isLocal) {
- unsigned int collSize = (unsigned int)coll_.size();
- TRACE_ENTER();
- TRACE_2("notification Id=%llu received in logger with size %d",
- notif->getNotificationId(), collSize);
-
- if (isLocal) {
- TRACE_2("IS LOCAL, logging");
- /* Currently only alarm notifations are logged */
- this->checkQueueAndLog(notif);
- }
+void NtfLogger::log(NtfSmartPtr& newNotification) {
+ TRACE_ENTER2("Notification Id=%llu received in logger. Logger buffer size
%d",
+ newNotification->getNotificationId(),
+ static_cast<int>(queuedNotificationList.size()));
+ uint32_t is_buffer_full = isLoggerBufferFull();
- if ((notif->sendNotInfo_->notificationType == SA_NTF_TYPE_ALARM) ||
- (notif->sendNotInfo_->notificationType == SA_NTF_TYPE_SECURITY_ALARM)) {
- TRACE_2("template queue handling...");
- if (coll_.size() < ntfs_cb->cache_size) {
- TRACE_2("push_back");
- coll_.push_back(notif);
- } else {
- TRACE_2("pop_front");
- coll_.pop_front();
- TRACE_2("push_back");
- coll_.push_back(notif);
- }
- }
- TRACE_LEAVE();
-}
-
-void NtfLogger::queueNotifcation(NtfSmartPtr& notif) {
- TRACE_2("Queue notification: %llu", notif->getNotificationId());
- queuedNotificationList.push_back(notif);
-}
-
-void NtfLogger::checkQueueAndLog(NtfSmartPtr& newNotif) {
- TRACE_ENTER();
- /* Check if there are not logged notifications in queue */
- while (!queuedNotificationList.empty()) {
+ // Check if there are not logged notifications in logger buffer
+ while (queuedNotificationList.empty() == false) {
NtfSmartPtr notification = queuedNotificationList.front();
queuedNotificationList.pop_front();
TRACE_2("Log queued notification: %llu",
notification->getNotificationId());
if (SA_AIS_OK != this->logNotification(notification)) {
TRACE_2("Push back queued notification: %llu",
notification->getNotificationId());
- queuedNotificationList.push_front(notification); /* keep order */
- queueNotifcation(newNotif);
+ queuedNotificationList.push_front(notification); // Keep order
+ queueNotifcation(newNotification);
TRACE_LEAVE();
return;
}
}
- if (SA_AIS_OK != this->logNotification(newNotif)) {
- queueNotifcation(newNotif);
+ // The new notification should be only logged if the buffer is not full
+ // and the type of notification is alarm.
+ if ((is_buffer_full == false) &&
+ (isAlarmNotification(newNotification) == true)) {
+ if (logNotification(newNotification) != SA_AIS_OK)
+ queueNotifcation(newNotification);
}
+
TRACE_LEAVE();
}
+void NtfLogger::queueNotifcation(NtfSmartPtr& notif) {
+ TRACE_2("Queue notification: %llu", notif->getNotificationId());
+ queuedNotificationList.push_back(notif);
+}
+
SaAisErrorT NtfLogger::logNotification(NtfSmartPtr& notif) {
/* Write to the log if we're the local node */
SaAisErrorT errorCode = SA_AIS_OK;
@@ -215,27 +212,23 @@ SaAisErrorT NtfLogger::logNotification(NtfSmartPtr&
notif) {
SaLogRecordT logRecord = {*ntfHeader->eventTime, SA_LOG_NTF_HEADER,
logHeader,
&logBuffer};
- /* Also write alarms and security alarms to the alarm log */
- if ((notif->sendNotInfo_->notificationType == SA_NTF_TYPE_ALARM) ||
- (notif->sendNotInfo_->notificationType == SA_NTF_TYPE_SECURITY_ALARM)) {
- TRACE_2("Logging notification to alarm stream");
- errorCode =
- saLogWriteLogAsync(alarmStreamHandle, notif->getNotificationId(),
- SA_LOG_RECORD_WRITE_ACK, &logRecord);
- switch (errorCode) {
- case SA_AIS_OK:
- break;
-
- /* LOGsv is busy. Put the notification to queue and re-send next time */
- case SA_AIS_ERR_TRY_AGAIN:
- case SA_AIS_ERR_TIMEOUT:
- TRACE("Failed to log notification (ret: %d). Try next time.",
- errorCode);
- break;
-
- default:
- osaf_abort(errorCode);
- }
+ TRACE_2("Logging notification to alarm stream");
+ errorCode =
+ saLogWriteLogAsync(alarmStreamHandle, notif->getNotificationId(),
+ SA_LOG_RECORD_WRITE_ACK, &logRecord);
+ switch (errorCode) {
+ case SA_AIS_OK:
+ break;
+
+ /* LOGsv is busy. Put the notification to queue and re-send next time */
+ case SA_AIS_ERR_TRY_AGAIN:
+ case SA_AIS_ERR_TIMEOUT:
+ TRACE("Failed to log notification (ret: %d). Try next time.",
+ errorCode);
+ break;
+
+ default:
+ osaf_abort(errorCode);
}
TRACE_LEAVE();
@@ -361,3 +354,37 @@ void NtfLogger::printInfo() {
TRACE(" logQueueList size: %u", (unsigned
int)queuedNotificationList.size());
TRACE(" reader cache size: %u", (unsigned int)coll_.size());
}
+
+
+void NtfLogger::addNotificationToReaderList(NtfSmartPtr& notification) {
+ TRACE_ENTER();
+
+ // Just add the notification to Reader list if it is alarm notification
+ // and the logger buffer also isn't full
+ if ((isAlarmNotification(notification) == false) ||
+ (isLoggerBufferFull() == true)) return;
+
+ TRACE_2("Add notification to the Reader list: %d",
+ static_cast<int>(coll_.size()));
+ if (coll_.size() < ntfs_cb->cache_size) {
+ TRACE_2("push_back");
+ coll_.push_back(notification);
+ } else {
+ TRACE_2("pop_front");
+ coll_.pop_front();
+ TRACE_2("push_back");
+ coll_.push_back(notification);
+ }
+}
+
+bool NtfLogger::isLoggerBufferFull() {
+ return (queuedNotificationList.size() >= logger_buffer_capacity);
+}
+
+bool NtfLogger::isAlarmNotification(NtfSmartPtr& notif) {
+ if ((notif->sendNotInfo_->notificationType == SA_NTF_TYPE_ALARM) ||
+ (notif->sendNotInfo_->notificationType == SA_NTF_TYPE_SECURITY_ALARM))
+ return true;
+ else
+ return false;
+}
diff --git a/src/ntf/ntfd/NtfLogger.h b/src/ntf/ntfd/NtfLogger.h
index cc0ac7dba..4bf2a195e 100644
--- a/src/ntf/ntfd/NtfLogger.h
+++ b/src/ntf/ntfd/NtfLogger.h
@@ -55,13 +55,19 @@ class NtfLogger {
NtfLogger();
// virtual ~NtfLogger();
- void log(NtfSmartPtr& notif, bool isLocal);
- void checkQueueAndLog(NtfSmartPtr& notif);
+ void log(NtfSmartPtr& newNotification);
SaAisErrorT logNotification(NtfSmartPtr& notif);
void queueNotifcation(NtfSmartPtr& notif);
void printInfo();
void syncRequest(NCS_UBAID *uba);
+ // Add the notification to the reader list
+ void addNotificationToReaderList(NtfSmartPtr& notification);
+ // Check if the buffer is already full or not
+ bool isLoggerBufferFull();
+ // Check if the type of notification is alarm/alarm security or not
+ bool isAlarmNotification(NtfSmartPtr& notif);
+
private:
SaAisErrorT initLog();
@@ -69,6 +75,7 @@ class NtfLogger {
unsigned int readCounter;
typedef std::list<NtfSmartPtr> QueuedNotificationsList;
QueuedNotificationsList queuedNotificationList;
+ uint32_t logger_buffer_capacity;
};
#endif // NTF_NTFD_NTFLOGGER_H_
diff --git a/src/ntf/ntfd/ntfd.conf b/src/ntf/ntfd/ntfd.conf
index c480bace2..91bfcd2e2 100644
--- a/src/ntf/ntfd/ntfd.conf
+++ b/src/ntf/ntfd/ntfd.conf
@@ -37,3 +37,13 @@ export NTFSV_ENV_HEALTHCHECK_KEY="Default"
# It can be disabled if set THREAD_TRACE_BUFFER as 0, the maximum value
# can be set as 65535.
# export THREAD_TRACE_BUFFER=10240
+
+# The logger buffer is used to store the notification if writing notification
+# to log file fail. This variable is set for limit of logger buffer size in
+# NTFD. If the logger buffer is full and NTFD receives new notification,
+# the TRY_AGAIN error is returned to user. The limit should be set with
relevant
+# value to avoid congestion in NTFD. Because if this value is set too big and
+# writing notification is fail for long time, NTF has to write a big number of
+# notifications whenever handling sending notification request and that will
delay
+# to handle other requests come to NTFD. The value of variable is from 10 to
5000.
+#export NTFSV_LOGGER_BUFFER_CAPACITY=10
diff --git a/src/ntf/ntfd/ntfs.h b/src/ntf/ntfd/ntfs.h
index 24da616fb..b3e8b90dc 100644
--- a/src/ntf/ntfd/ntfs.h
+++ b/src/ntf/ntfd/ntfs.h
@@ -50,6 +50,8 @@
* ========================================================================
*/
#define NTFSV_READER_CACHE_DEFAULT 10000
+#define NTFSV_LOGGER_BUFFER_CAPACITY_DEFAULT 10
+#define NTFSV_LOGGER_BUFFER_CAPACITY_MAX 5000
/* ========================================================================
* TYPE DEFINITIONS
_______________________________________________
Opensaf-devel mailing list
Opensaf-devel@lists.sourceforge.net
https://lists.sourceforge.net/lists/listinfo/opensaf-devel