+ rc = dtrq_sub_add_cookie(dtrqs, cookie);
+
+ RETURN(rc);
+}
+
+/**
+ * append updates to the current replay updates
+ *
+ * Append more updates to the existent replay update. And this is only
+ * used when combining mulitple updates into one large updates during
+ * replay.
+ *
+ * \param[in] dtrq the update replay request where the new update
+ * records will be added.
+ * \param[in] lur the new update record.
+ *
+ * \retval 0 if appending succeeds.
+ * \retval negative errno if appending fails.
+ */
+static int dtrq_append_updates(struct distribute_txn_replay_req *dtrq,
+ struct update_records *record)
+{
+ struct llog_update_record *new_lur;
+ size_t lur_size = dtrq->dtrq_lur_size;
+ void *ptr;
+ ENTRY;
+
+ /* Because several threads might retrieve the same records from
+ * different targets, and we only need one copy of records. So
+ * we will check if the records is in the next one, if not, just
+ * skip it */
+ spin_lock(&dtrq->dtrq_sub_list_lock);
+ if (dtrq->dtrq_lur->lur_update_rec.ur_index + 1 != record->ur_index) {
+ spin_unlock(&dtrq->dtrq_sub_list_lock);
+ RETURN(0);
+ }
+ dtrq->dtrq_lur->lur_update_rec.ur_index++;
+ spin_unlock(&dtrq->dtrq_sub_list_lock);
+
+ lur_size += update_records_size(record);
+ OBD_ALLOC_LARGE(new_lur, lur_size);
+ if (new_lur == NULL) {
+ spin_lock(&dtrq->dtrq_sub_list_lock);
+ dtrq->dtrq_lur->lur_update_rec.ur_index--;
+ spin_unlock(&dtrq->dtrq_sub_list_lock);
+ RETURN(-ENOMEM);
+ }
+
+ /* Copy the old and new records to the new allocated buffer */
+ memcpy(new_lur, dtrq->dtrq_lur, dtrq->dtrq_lur_size);
+ ptr = (char *)&new_lur->lur_update_rec +
+ update_records_size(&new_lur->lur_update_rec);
+ memcpy(ptr, &record->ur_ops,
+ update_records_size(record) -
+ offsetof(struct update_records, ur_ops));
+
+ new_lur->lur_update_rec.ur_update_count += record->ur_update_count;
+ new_lur->lur_update_rec.ur_param_count += record->ur_param_count;
+ new_lur->lur_hdr.lrh_len = llog_update_record_size(new_lur);
+
+ /* Replace the records */
+ OBD_FREE_LARGE(dtrq->dtrq_lur, dtrq->dtrq_lur_size);
+ dtrq->dtrq_lur = new_lur;
+ dtrq->dtrq_lur_size = lur_size;
+ dtrq->dtrq_lur->lur_update_rec.ur_flags = record->ur_flags;
+ update_records_dump(&new_lur->lur_update_rec, D_INFO, true);