Assumpt that the reader information is updated to the standby ntfd via
checkpoint
upon reception of reader APIs requests. However, if the standby controller
reboots
and comes up, the standby ntfd still has none of readers information which is
available at the active ntfd. Now if a switchover happens, the new active will
not
be able to process the reader APIs requests with existing reader handles.
This patch adds reader information as part of cold sync
---
src/ntf/common/ntfsv_enc_dec.c | 1 -
src/ntf/ntfd/NtfAdmin.cc | 63 +++++++++++++++++-
src/ntf/ntfd/NtfAdmin.h | 5 ++
src/ntf/ntfd/NtfClient.cc | 7 ++
src/ntf/ntfd/NtfReader.cc | 18 ++++-
src/ntf/ntfd/NtfReader.h | 1 +
src/ntf/ntfd/ntfs_com.c | 42 +++++++++---
src/ntf/ntfd/ntfs_com.h | 8 +++
src/ntf/ntfd/ntfs_mbcsv.c | 147 ++++++++++++++++++++++++++++-------------
9 files changed, 235 insertions(+), 57 deletions(-)
diff --git a/src/ntf/common/ntfsv_enc_dec.c b/src/ntf/common/ntfsv_enc_dec.c
index d38c401..5cecabf 100644
--- a/src/ntf/common/ntfsv_enc_dec.c
+++ b/src/ntf/common/ntfsv_enc_dec.c
@@ -2275,7 +2275,6 @@ uint32_t ntfsv_enc_read_finalize_msg(NCS_UBAID *uba,
ntfsv_reader_finalize_req_t
ncs_encode_32bit(&p8, param->client_id);
ncs_encode_32bit(&p8, param->readerId);
ncs_enc_claim_space(uba, 8);
-
TRACE_LEAVE();
return NCSCC_RC_SUCCESS;
}
diff --git a/src/ntf/ntfd/NtfAdmin.cc b/src/ntf/ntfd/NtfAdmin.cc
index 1242129..2cb9945 100644
--- a/src/ntf/ntfd/NtfAdmin.cc
+++ b/src/ntf/ntfd/NtfAdmin.cc
@@ -627,6 +627,12 @@ void NtfAdmin::syncRequest(NCS_UBAID *uba) {
LOG_ER("sendNewClient: %u failed", client->getClientId());
}
}
+ // Need to sync cached notification before syncing readers.
+ // In decoding side, all notifications will be restored before
+ // decoding readers, so that the internal cache list of every
+ // readers will be up-to-dated
+ logger.syncRequest(uba);
+
for (pos = clientMap.begin(); pos != clientMap.end(); pos++) {
pos->second->syncRequest(uba); /* subscriptions are synched here */
}
@@ -659,7 +665,6 @@ void NtfAdmin::syncRequest(NCS_UBAID *uba) {
NtfSmartPtr notification = posNot->second;
notification->syncRequest(uba);
}
- logger.syncRequest(uba);
TRACE_LEAVE();
}
@@ -770,6 +775,50 @@ NtfReader*
NtfAdmin::createReaderWithoutFilter(ntfsv_reader_init_req_t rp,
return newReader;
}
/**
+ * The method is called in cold sync to restore reader instance
+ * with filter at standby NTFD
+ *
+ * @param rp: the original reader initialize request version 2
+ * @param readerId: the current reader Id of reader instance exists
+ * at active side
+ * @param fIter: current iteration to read the notification
+ * @param firstRead: flag is used in NtfReader::next
+ * @return none
+ */
+void NtfAdmin::restoreReaderWithFilter(ntfsv_reader_init_req_2_t rp,
+ uint32_t readerId, uint32_t fIter, bool firstRead) {
+ TRACE_ENTER();
+ NtfReader *reader = createReaderWithFilter(rp, NULL);
+ if (reader != nullptr) {
+ reader->setReaderId(readerId);
+ reader->setReaderIteration(fIter);
+ reader->setFirstRead(firstRead);
+ }
+ TRACE_LEAVE();
+}
+/**
+ * The method is called in cold sync to restore reader instance
+ * without at standby NTFD
+ *
+ * @param rp: the original reader initialize request version 2
+ * @param readerId: the current reader Id of reader instance exists
+ * at active side
+ * @param fIter: current iteration to read the notification
+ * @param firstRead: flag is used in NtfReader::next
+ * @return none
+ */
+void NtfAdmin::restoreReaderWithoutFilter(ntfsv_reader_init_req_t rp,
+ uint32_t readerId, uint32_t fIter, bool firstRead) {
+ TRACE_ENTER();
+ NtfReader *reader = createReaderWithoutFilter(rp, NULL);
+ if (reader != nullptr) {
+ reader->setReaderId(readerId);
+ reader->setReaderIteration(fIter);
+ reader->setFirstRead(firstRead);
+ }
+ TRACE_LEAVE();
+}
+/**
* The method create a new instance of NtfReader that
* has filter
*
@@ -1144,6 +1193,18 @@ void createReaderWithFilter(ntfsv_reader_init_req_2_t
rp, MDS_SYNC_SND_CTXT *mds
NtfAdmin::theNtfAdmin->createReaderWithFilter(rp, mdsCtxt);
}
+void restoreReaderWithFilter(ntfsv_reader_init_req_2_t rp, uint32_t readerId,
+ uint32_t fIter, bool firstRead) {
+ osafassert(NtfAdmin::theNtfAdmin != NULL);
+ NtfAdmin::theNtfAdmin->restoreReaderWithFilter(rp, readerId, fIter,
firstRead);
+}
+
+void restoreReaderWithoutFilter(ntfsv_reader_init_req_t rp, uint32_t readerId,
+ uint32_t fIter, bool firstRead) {
+ osafassert(NtfAdmin::theNtfAdmin != NULL);
+ NtfAdmin::theNtfAdmin->restoreReaderWithoutFilter(rp, readerId, fIter,
firstRead);
+}
+
void readNext(ntfsv_read_next_req_t reqNextReq,
MDS_SYNC_SND_CTXT *mdsCtxt) {
osafassert(NtfAdmin::theNtfAdmin != NULL);
diff --git a/src/ntf/ntfd/NtfAdmin.h b/src/ntf/ntfd/NtfAdmin.h
index e569ec0..6e7be87 100644
--- a/src/ntf/ntfd/NtfAdmin.h
+++ b/src/ntf/ntfd/NtfAdmin.h
@@ -79,6 +79,11 @@ class NtfAdmin {
void readNext(ntfsv_read_next_req_t readNextReq,
MDS_SYNC_SND_CTXT *mdsCtxt);
+ void restoreReaderWithFilter(ntfsv_reader_init_req_2_t rp, uint32_t readerId,
+ uint32_t fIter, bool firstRead);
+ void restoreReaderWithoutFilter(ntfsv_reader_init_req_t rp, uint32_t
readerId,
+ uint32_t fIter, bool firstRead);
+
void printInfo();
void storeMatchingSubscription(SaNtfIdentifierT notificationId,
unsigned int clientId,
diff --git a/src/ntf/ntfd/NtfClient.cc b/src/ntf/ntfd/NtfClient.cc
index 4cd511f..6393a36 100644
--- a/src/ntf/ntfd/NtfClient.cc
+++ b/src/ntf/ntfd/NtfClient.cc
@@ -266,6 +266,13 @@ void NtfClient::syncRequest(NCS_UBAID* uba) {
subscription->getSubscriptionId(), clientId_);
subscription->syncRequest(uba);
}
+ // scan through all readers
+ sendNoOfReaders(readerMap.size(), uba);
+ ReaderMapT::iterator rpos;
+ for (rpos = readerMap.begin(); rpos != readerMap.end(); rpos++) {
+ NtfReader* reader = rpos->second;
+ reader->syncRequest(uba);
+ }
}
void NtfClient::sendNotConfirmedNotification(
diff --git a/src/ntf/ntfd/NtfReader.cc b/src/ntf/ntfd/NtfReader.cc
index 5f2f2c7..c2c6b00 100644
--- a/src/ntf/ntfd/NtfReader.cc
+++ b/src/ntf/ntfd/NtfReader.cc
@@ -164,7 +164,6 @@ NtfSmartPtr NtfReader::next(SaNtfSearchDirectionT direction,
TRACE_ENTER();
*error = SA_AIS_ERR_NOT_EXIST;
-
if (direction == SA_NTF_SEARCH_YOUNGER ||
searchCriteria_.searchMode == SA_NTF_SEARCH_AT_TIME ||
searchCriteria_.searchMode == SA_NTF_SEARCH_ONLY_FILTER) {
@@ -205,6 +204,23 @@ NtfSmartPtr NtfReader::next(SaNtfSearchDirectionT
direction,
return notif;
}
}
+/*
+ * This method is to sync the reader information stored in this
+ * class to the standby NTFD
+ * param: uba, encoder pointer
+ */
+void NtfReader::syncRequest(NCS_UBAID* uba) {
+ TRACE_ENTER();
+ syncReaderInfo(readerId_, c_filter_ != nullptr ? 1 : 0,
+ (uint32_t)std::distance(coll_.begin(), ffIter),
+ uint8_t(firstRead_), uba);
+ if (c_filter_) {
+ syncReaderWithFilter(&read_init_2_req_, uba);
+ } else {
+ syncReaderWithoutFilter(&read_init_req_, uba);
+ }
+ TRACE_LEAVE();
+}
unsigned int NtfReader::getId() { return readerId_; }
diff --git a/src/ntf/ntfd/NtfReader.h b/src/ntf/ntfd/NtfReader.h
index 7ae983d..8be8985 100644
--- a/src/ntf/ntfd/NtfReader.h
+++ b/src/ntf/ntfd/NtfReader.h
@@ -56,6 +56,7 @@ class NtfReader {
void filterCacheList(NtfLogger& ntfLogger);
NtfSmartPtr next(SaNtfSearchDirectionT direction, SaAisErrorT* error);
unsigned int getId();
+ void syncRequest(NCS_UBAID* uba);
void setReaderId(unsigned int readerId) { readerId_ = readerId; }
void setReaderIteration(unsigned int iterPos) {ffIter = coll_.begin() +
iterPos;}
void setFirstRead(bool firstRead) {firstRead_ = firstRead; }
diff --git a/src/ntf/ntfd/ntfs_com.c b/src/ntf/ntfd/ntfs_com.c
index bfef875..35435bb 100644
--- a/src/ntf/ntfd/ntfs_com.c
+++ b/src/ntf/ntfd/ntfs_com.c
@@ -484,6 +484,40 @@ int sendNewSubscription(ntfsv_subscribe_req_t *s,
NCS_UBAID *uba)
return 1;
};
+int syncReaderWithoutFilter(ntfsv_reader_init_req_t *s, NCS_UBAID *uba)
+{
+ if (ntfs_cb->peer_mbcsv_version > NTFS_MBCSV_VERSION_2) {
+ return ntfsv_enc_reader_initialize_msg(uba, s);
+ }
+ return NCSCC_RC_SUCCESS;
+}
+
+int syncReaderWithFilter(ntfsv_reader_init_req_2_t *s, NCS_UBAID *uba)
+{
+ if (ntfs_cb->peer_mbcsv_version > NTFS_MBCSV_VERSION_2) {
+ return ntfsv_enc_reader_initialize_2_msg(uba, s);
+ }
+ return NCSCC_RC_SUCCESS;
+}
+
+int syncReaderInfo(uint32_t readerId, uint8_t hasFilter,
+ uint32_t fIter, uint8_t firstRead, NCS_UBAID *uba)
+{
+ if (ntfs_cb->peer_mbcsv_version > NTFS_MBCSV_VERSION_2) {
+ uint8_t *p8;
+ p8 = ncs_enc_reserve_space(uba, 10);
+ if (!p8) {
+ TRACE("NULL pointer");
+ return NCSCC_RC_OUT_OF_MEM;
+ }
+ ncs_encode_32bit(&p8, readerId);
+ ncs_encode_8bit(&p8, hasFilter);
+ ncs_encode_32bit(&p8, fIter);
+ ncs_encode_8bit(&p8, firstRead);
+ ncs_enc_claim_space(uba, 10);
+ }
+ return NCSCC_RC_SUCCESS;
+}
int syncLoggedConfirm(unsigned int logged, NCS_UBAID *uba)
{
TRACE_ENTER2("NOT IMPLEMENTED");
@@ -592,9 +626,7 @@ void sendNotConfirmUpdate(unsigned int clientId,
void sendReaderInitialize2Update(ntfsv_reader_init_req_2_t*
readerInitializeReq)
{
if (ntfs_cb->peer_mbcsv_version > NTFS_MBCSV_VERSION_2) {
-
ntfsv_ckpt_msg_t ckpt;
-
memset(&ckpt, 0, sizeof(ckpt));
ckpt.header.ckpt_rec_type = NTFS_CKPT_READER_INITIALIZE_2;
ckpt.header.num_ckpt_records = 1;
@@ -607,9 +639,7 @@ void sendReaderInitialize2Update(ntfsv_reader_init_req_2_t*
readerInitializeReq)
void sendReadNextUpdate(ntfsv_read_next_req_t* readNextReq)
{
if (ntfs_cb->peer_mbcsv_version > NTFS_MBCSV_VERSION_2) {
-
ntfsv_ckpt_msg_t ckpt;
-
memset(&ckpt, 0, sizeof(ckpt));
ckpt.header.ckpt_rec_type = NTFS_CKPT_READ_NEXT;
ckpt.header.num_ckpt_records = 1;
@@ -622,9 +652,7 @@ void sendReadNextUpdate(ntfsv_read_next_req_t* readNextReq)
void sendReadFinalizeUpdate(ntfsv_reader_finalize_req_t* readFinalizeReq)
{
if (ntfs_cb->peer_mbcsv_version > NTFS_MBCSV_VERSION_2) {
-
ntfsv_ckpt_msg_t ckpt;
-
memset(&ckpt, 0, sizeof(ckpt));
ckpt.header.ckpt_rec_type = NTFS_CKPT_READ_FINALIZE;
ckpt.header.num_ckpt_records = 1;
@@ -637,9 +665,7 @@ void sendReadFinalizeUpdate(ntfsv_reader_finalize_req_t*
readFinalizeReq)
void sendReaderInitializeUpdate(ntfsv_reader_init_req_t* readerInitializeReq)
{
if (ntfs_cb->peer_mbcsv_version > NTFS_MBCSV_VERSION_2) {
-
ntfsv_ckpt_msg_t ckpt;
-
memset(&ckpt, 0, sizeof(ckpt));
ckpt.header.ckpt_rec_type = NTFS_CKPT_READER_INITIALIZE;
ckpt.header.num_ckpt_records = 1;
diff --git a/src/ntf/ntfd/ntfs_com.h b/src/ntf/ntfd/ntfs_com.h
index 1c1b283..96f111a 100644
--- a/src/ntf/ntfd/ntfs_com.h
+++ b/src/ntf/ntfd/ntfs_com.h
@@ -90,6 +90,14 @@ void deleteReader(ntfsv_reader_finalize_req_t
readFinalizeReq,
MDS_SYNC_SND_CTXT *mdsCtxt);
void readNext(ntfsv_read_next_req_t readNextReq,
MDS_SYNC_SND_CTXT *mdsCtxt);
+void restoreReaderWithFilter(ntfsv_reader_init_req_2_t rp, uint32_t readerId,
+ uint32_t fIter, bool firstRead);
+void restoreReaderWithoutFilter(ntfsv_reader_init_req_t rp, uint32_t readerId,
+ uint32_t fIter, bool firstRead);
+int syncReaderWithoutFilter(ntfsv_reader_init_req_t *s, NCS_UBAID *uba);
+int syncReaderWithFilter(ntfsv_reader_init_req_2_t *s, NCS_UBAID *uba);
+int syncReaderInfo(uint32_t readerId, uint8_t hasFilter,
+ uint32_t fIter, uint8_t firstRead, NCS_UBAID *uba);
void sendReaderInitialize2Update(ntfsv_reader_init_req_2_t*
readerInitializeReq);
void sendReaderInitializeUpdate(ntfsv_reader_init_req_t* readerInitializeReq);
void sendReadNextUpdate(ntfsv_read_next_req_t* readNextReq);
diff --git a/src/ntf/ntfd/ntfs_mbcsv.c b/src/ntf/ntfd/ntfs_mbcsv.c
index cdf2aa5..dc890cc 100644
--- a/src/ntf/ntfd/ntfs_mbcsv.c
+++ b/src/ntf/ntfd/ntfs_mbcsv.c
@@ -991,6 +991,22 @@ static uint32_t decode_subscribe_msg(NCS_UBAID *uba,
return ntfsv_dec_subscribe_msg(uba, param);
}
+static uint32_t decode_reader_info(NCS_UBAID *uba, uint32_t* readerId,
+ uint8_t* hasFilter, uint32_t *fIter, bool *firstRead)
+{
+ uint8_t *p8;
+ uint8_t local_data[9];
+
+ /* releaseCode, majorVersion, minorVersion */
+ p8 = ncs_dec_flatten_space(uba, local_data, 10);
+ *readerId = ncs_decode_32bit(&p8);
+ *hasFilter = ncs_decode_8bit(&p8);
+ *fIter = ncs_decode_32bit(&p8);
+ *firstRead = (bool)ncs_decode_8bit(&p8);
+ ncs_dec_skip_space(uba, 10);
+ return NCSCC_RC_SUCCESS;
+}
+
static uint32_t decode_not_log_confirm_msg(NCS_UBAID *uba,
ntfs_ckpt_not_log_confirm_t *param)
{
@@ -1091,13 +1107,59 @@ static uint32_t ckpt_decode_cold_sync(ntfs_cb_t *cb,
NCS_MBCSV_CB_ARG *cbk_arg)
--num_rec;
} /*End while, reg records */
+ if (ntfs_cb->peer_mbcsv_version > NTFS_MBCSV_VERSION_2) {
+
+ /* Decode the current message header */
+ if ((rc = dec_ckpt_header(&cbk_arg->info.decode.i_uba,
+ &data->header)) !=
NCSCC_RC_SUCCESS) {
+ goto done;
+ }
+
+ TRACE_2("ckpt_rec_type: %u", data->header.ckpt_rec_type);
+
+ /* Check if the second in the order of records is notification
record */
+ if (data->header.ckpt_rec_type != NTFS_CKPT_NOTIFICATION) {
+ TRACE(
+ "FAILED data->header.ckpt_rec_type !=
NTFS_CKPT_NOTIFICATION");
+ rc = NCSCC_RC_FAILURE;
+ goto done;
+ }
+
+ num_rec = data->header.num_ckpt_records;
+ TRACE_2("Cached Notifications num_rec: %u", num_rec);
+ while (num_rec) {
+ /* freed in NtfNotification destructor */
+ ntfsv_send_not_req_t *notification_rec =
+ calloc(1, sizeof(ntfsv_send_not_req_t));
+
+ if (notification_rec == NULL) {
+ TRACE("calloc FAILED");
+ rc = NCSCC_RC_FAILURE;
+ goto done;
+ }
+
+ rc = ntfsv_dec_not_msg(&cbk_arg->info.decode.i_uba,
+
notification_rec);
+ if (rc != NCSCC_RC_SUCCESS) {
+ TRACE("decode_subscribe_msg FAILED");
+ goto done;
+ }
+
cachedNotificationReceivedColdSync(notification_rec->client_id,
+
notification_rec->notificationType,
+
notification_rec);
+
+ num_rec--;
+ }
+ }
+
while (num_clients) {
+ TRACE("num_clients: %u", num_clients);
/* Decode the current message header */
if ((rc = dec_ckpt_header(&cbk_arg->info.decode.i_uba,
&data->header)) != NCSCC_RC_SUCCESS) {
goto done;
}
-
+ TRACE("subcribers ckpt_rec_type:%u",
data->header.ckpt_rec_type);
/* Check if the second in the order of records is subscription
* record */
if (data->header.ckpt_rec_type != NTFS_CKPT_SUBSCRIBE) {
@@ -1107,7 +1169,7 @@ static uint32_t ckpt_decode_cold_sync(ntfs_cb_t *cb,
NCS_MBCSV_CB_ARG *cbk_arg)
goto done;
}
num_rec = data->header.num_ckpt_records;
- TRACE_2("subscribers num_rec: %u", num_rec);
+ TRACE("subscribers num_rec: %u", num_rec);
while (num_rec) {
ntfsv_subscribe_req_t subscribe_rec;
rc = decode_subscribe_msg(&cbk_arg->info.decode.i_uba,
@@ -1119,6 +1181,42 @@ static uint32_t ckpt_decode_cold_sync(ntfs_cb_t *cb,
NCS_MBCSV_CB_ARG *cbk_arg)
subscriptionAdded(subscribe_rec, NULL);
num_rec--;
}
+
+ // Decode reader
+ if (ntfs_cb->peer_mbcsv_version > NTFS_MBCSV_VERSION_2) {
+ if ((rc = dec_ckpt_header(&cbk_arg->info.decode.i_uba,
+ &data->header)) != NCSCC_RC_SUCCESS) {
+ goto done;
+ }
+ if (data->header.ckpt_rec_type !=
NTFS_CKPT_READER_INITIALIZE_2) {
+ TRACE(
+ "FAILED data->header.ckpt_rec_type !=
NTFS_CKPT_READER_INITIALIZE_2");
+ rc = NCSCC_RC_FAILURE;
+ goto done;
+ }
+ num_rec = data->header.num_ckpt_records;
+ TRACE("readers num_rec: %u", num_rec);
+ while (num_rec) {
+ uint32_t fIter, readerId;
+ bool firstRead;
+ uint8_t hasFilter;
+ decode_reader_info(&cbk_arg->info.decode.i_uba,
+ &readerId, &hasFilter, &fIter,
&firstRead);
+ if (hasFilter == 1) {
+ ntfsv_reader_init_req_2_t read_init_2;
+ rc =
ntfsv_dec_reader_initialize_2_msg(&cbk_arg->info.decode.i_uba,
+ &read_init_2);
+ restoreReaderWithFilter(read_init_2,
readerId, fIter, firstRead);
+ } else if (hasFilter == 0) {
+ ntfsv_reader_init_req_t read_init;
+ rc =
ntfsv_dec_reader_initialize_msg(&cbk_arg->info.decode.i_uba,
+ &read_init);
+ restoreReaderWithoutFilter(read_init,
readerId, fIter, firstRead);
+ }
+ num_rec--;
+ }
+ }
+
num_clients--;
}
@@ -1192,50 +1290,7 @@ static uint32_t ckpt_decode_cold_sync(ntfs_cb_t *cb,
NCS_MBCSV_CB_ARG *cbk_arg)
}
num_rec--;
}
- if (ntfs_cb->peer_mbcsv_version > NTFS_MBCSV_VERSION_2) {
-
- /* Decode the current message header */
- if ((rc = dec_ckpt_header(&cbk_arg->info.decode.i_uba,
- &data->header)) != NCSCC_RC_SUCCESS) {
- goto done;
- }
-
- TRACE_2("ckpt_rec_type: %u", data->header.ckpt_rec_type);
-
- /* Check if the second in the order of records is notification record */
- if (data->header.ckpt_rec_type != NTFS_CKPT_NOTIFICATION) {
- TRACE(
- "FAILED data->header.ckpt_rec_type != NTFS_CKPT_NOTIFICATION");
- rc = NCSCC_RC_FAILURE;
- goto done;
- }
-
- num_rec = data->header.num_ckpt_records;
- TRACE_2("Cached Notifications num_rec: %u", num_rec);
- while (num_rec) {
- /* freed in NtfNotification destructor */
- ntfsv_send_not_req_t *notification_rec =
- calloc(1, sizeof(ntfsv_send_not_req_t));
-
- if (notification_rec == NULL) {
- TRACE("calloc FAILED");
- rc = NCSCC_RC_FAILURE;
- goto done;
- }
-
- rc = ntfsv_dec_not_msg(&cbk_arg->info.decode.i_uba,
- notification_rec);
- if (rc != NCSCC_RC_SUCCESS) {
- TRACE("decode_subscribe_msg FAILED");
- goto done;
- }
- cachedNotificationReceivedColdSync(notification_rec->client_id,
- notification_rec->notificationType,
- notification_rec);
-
- num_rec--;
- }
- }
+
/* Get the async update count */
ptr = ncs_dec_flatten_space(&cbk_arg->info.decode.i_uba, data_cnt,
sizeof(uint32_t));
--
2.7.4
------------------------------------------------------------------------------
Check out the vibrant tech community on one of the world's most
engaging tech sites, Slashdot.org! http://sdm.link/slashdot
_______________________________________________
Opensaf-devel mailing list
Opensaf-devel@lists.sourceforge.net
https://lists.sourceforge.net/lists/listinfo/opensaf-devel
