Whamcloud - gitweb
LU-6840 target: update reply data after update replay
[fs/lustre-release.git] / lustre / target / update_recovery.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2014, Intel Corporation.
24  */
25
26 /*
27  * lustre/target/update_recovery.c
28  *
29  * This file implement the methods to handle the update recovery.
30  *
31  * During DNE recovery, the recovery thread will redo the operation according
32  * to the transaction no, and these replay are either from client replay req
33  * or update replay records(for distribute transaction) in the update log.
34  * For distribute transaction replay, the replay thread will call
35  * distribute_txn_replay_handle() to handle the updates.
36  *
37  * After the Master MDT restarts, it will retrieve the update records from all
38  * of MDTs, for each distributed operation, it will check updates on all MDTs,
39  * if some updates records are missing on some MDTs, the replay thread will redo
40  * updates on these MDTs.
41  *
42  * Author: Di Wang <di.wang@intel.com>
43  */
44 #define DEBUG_SUBSYSTEM S_CLASS
45
46 #include <lu_target.h>
47 #include <md_object.h>
48 #include <lustre_update.h>
49 #include <obd.h>
50 #include <obd_class.h>
51 #include "tgt_internal.h"
52
53 /**
54  * Lookup distribute_txn_replay req
55  *
56  * Lookup distribute_txn_replay in the replay list by batchid.
57  * It is assumed the list has been locked before calling this function.
58  *
59  * \param[in] tdtd      distribute_txn_data, which holds the replay
60  *                      list.
61  * \param[in] batchid   batchid used by lookup.
62  *
63  * \retval              pointer of the replay if succeeds.
64  * \retval              NULL if can not find it.
65  */
66 static struct distribute_txn_replay_req *
67 dtrq_lookup(struct target_distribute_txn_data *tdtd, __u64 batchid)
68 {
69         struct distribute_txn_replay_req        *tmp;
70         struct distribute_txn_replay_req        *dtrq = NULL;
71
72         list_for_each_entry(tmp, &tdtd->tdtd_replay_list, dtrq_list) {
73                 if (tmp->dtrq_batchid == batchid) {
74                         dtrq = tmp;
75                         break;
76                 }
77         }
78         return dtrq;
79 }
80
81 /**
82  * insert distribute txn replay req
83  *
84  * Insert distribute txn replay to the replay list, and it assumes the
85  * list has been looked. Note: the replay list is a sorted list, which
86  * is sorted by master transno. It is assumed the replay list has been
87  * locked before calling this function.
88  *
89  * \param[in] tdtd      target distribute txn data where replay list is
90  * \param[in] new       distribute txn replay to be inserted
91  *
92  * \retval              0 if insertion succeeds
93  * \retval              EEXIST if the dtrq already exists
94  */
95 static int dtrq_insert(struct target_distribute_txn_data *tdtd,
96                         struct distribute_txn_replay_req *new)
97 {
98         struct distribute_txn_replay_req *iter;
99
100         /* Check if the dtrq has been added to the list */
101         iter = dtrq_lookup(tdtd, new->dtrq_batchid);
102         if (iter != NULL)
103                 return -EEXIST;
104
105         list_for_each_entry_reverse(iter, &tdtd->tdtd_replay_list, dtrq_list) {
106                 if (iter->dtrq_master_transno > new->dtrq_master_transno)
107                         continue;
108
109                 /* If there are mulitple replay req with same transno, then
110                  * sort them with batchid */
111                 if (iter->dtrq_master_transno == new->dtrq_master_transno &&
112                     iter->dtrq_batchid > new->dtrq_batchid)
113                         continue;
114
115                 list_add(&new->dtrq_list, &iter->dtrq_list);
116                 break;
117         }
118
119         if (list_empty(&new->dtrq_list))
120                 list_add(&new->dtrq_list, &tdtd->tdtd_replay_list);
121
122         return 0;
123 }
124
125 /**
126  * create distribute txn replay req
127  *
128  * Allocate distribute txn replay req according to the update records.
129  *
130  * \param[in] tdtd      target distribute txn data where replay list is.
131  * \param[in] record    update records from the update log.
132  *
133  * \retval              the pointer of distribute txn replay req if
134  *                      the creation succeeds.
135  * \retval              NULL if the creation fails.
136  */
137 static struct distribute_txn_replay_req *
138 dtrq_create(struct target_distribute_txn_data *tdtd,
139             struct llog_update_record *lur)
140 {
141         struct distribute_txn_replay_req *new;
142
143         OBD_ALLOC_PTR(new);
144         if (new == NULL)
145                 RETURN(ERR_PTR(-ENOMEM));
146
147         new->dtrq_lur_size = llog_update_record_size(lur);
148         OBD_ALLOC_LARGE(new->dtrq_lur, new->dtrq_lur_size);
149         if (new->dtrq_lur == NULL) {
150                 OBD_FREE_PTR(new);
151                 RETURN(ERR_PTR(-ENOMEM));
152         }
153
154         memcpy(new->dtrq_lur, lur, new->dtrq_lur_size);
155
156         /* If the transno in the update record is 0, it means the
157          * update are from master MDT, and it will use the master
158          * last committed transno as its master transno. Later, if
159          * the update records are gotten from slave MDTs, then these
160          * transno will be replaced.
161          * See insert_update_records_to_replay_list(). */
162         if (lur->lur_update_rec.ur_master_transno == 0) {
163                 new->dtrq_lur->lur_update_rec.ur_master_transno =
164                                 tdtd->tdtd_lut->lut_obd->obd_last_committed;
165                 new->dtrq_master_transno =
166                                 tdtd->tdtd_lut->lut_obd->obd_last_committed;
167         } else {
168                 new->dtrq_master_transno =
169                                 lur->lur_update_rec.ur_master_transno;
170         }
171
172         new->dtrq_batchid = lur->lur_update_rec.ur_batchid;
173
174         spin_lock_init(&new->dtrq_sub_list_lock);
175         INIT_LIST_HEAD(&new->dtrq_sub_list);
176         INIT_LIST_HEAD(&new->dtrq_list);
177
178         RETURN(new);
179 }
180
181 /**
182  * Lookup distribute sub replay
183  *
184  * Lookup distribute sub replay in the sub list of distribute_txn_replay by
185  * mdt_index.
186  *
187  * \param[in] distribute_txn_replay_req the distribute txn replay req to lookup
188  * \param[in] mdt_index                 the mdt_index as the key of lookup
189  *
190  * \retval              the pointer of sub replay if it can be found.
191  * \retval              NULL if it can not find.
192  */
193 struct distribute_txn_replay_req_sub *
194 dtrq_sub_lookup(struct distribute_txn_replay_req *dtrq, __u32 mdt_index)
195 {
196         struct distribute_txn_replay_req_sub *dtrqs = NULL;
197         struct distribute_txn_replay_req_sub *tmp;
198
199         list_for_each_entry(tmp, &dtrq->dtrq_sub_list, dtrqs_list) {
200                 if (tmp->dtrqs_mdt_index == mdt_index) {
201                         dtrqs = tmp;
202                         break;
203                 }
204         }
205         return dtrqs;
206 }
207
208 /**
209  * Try to add cookie to sub distribute txn request
210  *
211  * Check if the update log cookie has been added to the request, if not,
212  * add it to the dtrqs_cookie_list.
213  *
214  * \param[in] dtrqs     sub replay req where cookies to be added.
215  * \param[in] cookie    cookie to be added.
216  *
217  * \retval              0 if the cookie is adding succeeds.
218  * \retval              negative errno if adding fails.
219  */
220 static int dtrq_sub_add_cookie(struct distribute_txn_replay_req_sub *dtrqs,
221                                struct llog_cookie *cookie)
222 {
223         struct sub_thandle_cookie *new;
224
225         OBD_ALLOC_PTR(new);
226         if (new == NULL)
227                 return -ENOMEM;
228
229         INIT_LIST_HEAD(&new->stc_list);
230         new->stc_cookie = *cookie;
231         /* Note: only single thread will access one sub_request each time,
232          * so no need lock here */
233         list_add(&new->stc_list, &dtrqs->dtrqs_cookie_list);
234
235         return 0;
236 }
237
238 /**
239  * Insert distribute txn sub req replay
240  *
241  * Allocate sub replay req and insert distribute txn replay list.
242  *
243  * \param[in] dtrq      d to be added
244  * \param[in] cookie    the cookie of the update record
245  * \param[in] mdt_index the mdt_index of the update record
246  *
247  * \retval              0 if the adding succeeds.
248  * \retval              negative errno if the adding fails.
249  */
250 static int
251 dtrq_sub_create_and_insert(struct distribute_txn_replay_req *dtrq,
252                            struct llog_cookie *cookie,
253                            __u32 mdt_index)
254 {
255         struct distribute_txn_replay_req_sub    *dtrqs = NULL;
256         struct distribute_txn_replay_req_sub    *new;
257         int                                     rc;
258         ENTRY;
259
260         spin_lock(&dtrq->dtrq_sub_list_lock);
261         dtrqs = dtrq_sub_lookup(dtrq, mdt_index);
262         spin_unlock(&dtrq->dtrq_sub_list_lock);
263         if (dtrqs != NULL) {
264                 rc = dtrq_sub_add_cookie(dtrqs, cookie);
265                 RETURN(0);
266         }
267
268         OBD_ALLOC_PTR(new);
269         if (new == NULL)
270                 RETURN(-ENOMEM);
271
272         INIT_LIST_HEAD(&new->dtrqs_list);
273         INIT_LIST_HEAD(&new->dtrqs_cookie_list);
274         new->dtrqs_mdt_index = mdt_index;
275         spin_lock(&dtrq->dtrq_sub_list_lock);
276         dtrqs = dtrq_sub_lookup(dtrq, mdt_index);
277         if (dtrqs == NULL) {
278                 list_add(&new->dtrqs_list, &dtrq->dtrq_sub_list);
279                 dtrqs = new;
280         } else {
281                 OBD_FREE_PTR(new);
282         }
283         spin_unlock(&dtrq->dtrq_sub_list_lock);
284
285         rc = dtrq_sub_add_cookie(dtrqs, cookie);
286
287         RETURN(rc);
288 }
289
290 /**
291  * append updates to the current replay updates
292  *
293  * Append more updates to the existent replay update. And this is only
294  * used when combining mulitple updates into one large updates during
295  * replay.
296  *
297  * \param[in] dtrq      the update replay request where the new update
298  *                      records will be added.
299  * \param[in] lur       the new update record.
300  *
301  * \retval              0 if appending succeeds.
302  * \retval              negative errno if appending fails.
303  */
304 static int dtrq_append_updates(struct distribute_txn_replay_req *dtrq,
305                                struct update_records *record)
306 {
307         struct llog_update_record *new_lur;
308         size_t lur_size = dtrq->dtrq_lur_size;
309         void *ptr;
310         ENTRY;
311
312         /* Because several threads might retrieve the same records from
313          * different targets, and we only need one copy of records. So
314          * we will check if the records is in the next one, if not, just
315          * skip it */
316         spin_lock(&dtrq->dtrq_sub_list_lock);
317         if (dtrq->dtrq_lur->lur_update_rec.ur_index + 1 != record->ur_index) {
318                 spin_unlock(&dtrq->dtrq_sub_list_lock);
319                 RETURN(0);
320         }
321         dtrq->dtrq_lur->lur_update_rec.ur_index++;
322         spin_unlock(&dtrq->dtrq_sub_list_lock);
323
324         lur_size += update_records_size(record);
325         OBD_ALLOC_LARGE(new_lur, lur_size);
326         if (new_lur == NULL) {
327                 spin_lock(&dtrq->dtrq_sub_list_lock);
328                 dtrq->dtrq_lur->lur_update_rec.ur_index--;
329                 spin_unlock(&dtrq->dtrq_sub_list_lock);
330                 RETURN(-ENOMEM);
331         }
332
333         /* Copy the old and new records to the new allocated buffer */
334         memcpy(new_lur, dtrq->dtrq_lur, dtrq->dtrq_lur_size);
335         ptr = (char *)&new_lur->lur_update_rec +
336                 update_records_size(&new_lur->lur_update_rec);
337         memcpy(ptr, &record->ur_ops,
338                update_records_size(record) -
339                offsetof(struct update_records, ur_ops));
340
341         new_lur->lur_update_rec.ur_update_count += record->ur_update_count;
342         new_lur->lur_update_rec.ur_param_count += record->ur_param_count;
343         new_lur->lur_hdr.lrh_len = llog_update_record_size(new_lur);
344
345         /* Replace the records */
346         OBD_FREE_LARGE(dtrq->dtrq_lur, dtrq->dtrq_lur_size);
347         dtrq->dtrq_lur = new_lur;
348         dtrq->dtrq_lur_size = lur_size;
349         dtrq->dtrq_lur->lur_update_rec.ur_flags = record->ur_flags;
350         update_records_dump(&new_lur->lur_update_rec, D_INFO, true);
351         RETURN(0);
352 }
353
354 /**
355  * Insert update records to the replay list.
356  *
357  * Allocate distribute txn replay req and insert it into the replay
358  * list, then insert the update records into the replay req.
359  *
360  * \param[in] tdtd      distribute txn replay data where the replay list
361  *                      is.
362  * \param[in] record    the update record
363  * \param[in] cookie    cookie of the record
364  * \param[in] index     mdt index of the record
365  *
366  * \retval              0 if the adding succeeds.
367  * \retval              negative errno if the adding fails.
368  */
369 int
370 insert_update_records_to_replay_list(struct target_distribute_txn_data *tdtd,
371                                      struct llog_update_record *lur,
372                                      struct llog_cookie *cookie,
373                                      __u32 mdt_index)
374 {
375         struct distribute_txn_replay_req *dtrq;
376         struct update_records *record = &lur->lur_update_rec;
377         bool replace_record = false;
378         int rc = 0;
379         ENTRY;
380
381         CDEBUG(D_HA, "%s: insert record batchid = "LPU64" transno = "LPU64
382                " mdt_index %u\n", tdtd->tdtd_lut->lut_obd->obd_name,
383                record->ur_batchid, record->ur_master_transno, mdt_index);
384
385         /* Update batchid if necessary */
386         spin_lock(&tdtd->tdtd_batchid_lock);
387         if (record->ur_batchid >= tdtd->tdtd_batchid) {
388                 CDEBUG(D_HA, "%s update batchid from "LPU64 " to "LPU64"\n",
389                        tdtd->tdtd_lut->lut_obd->obd_name,
390                        tdtd->tdtd_batchid, record->ur_batchid);
391                 tdtd->tdtd_batchid = record->ur_batchid + 1;
392         }
393         spin_unlock(&tdtd->tdtd_batchid_lock);
394
395 again:
396         spin_lock(&tdtd->tdtd_replay_list_lock);
397         /* First try to build the replay update request with the records */
398         dtrq = dtrq_lookup(tdtd, record->ur_batchid);
399         if (dtrq == NULL) {
400                 spin_unlock(&tdtd->tdtd_replay_list_lock);
401                 dtrq = dtrq_create(tdtd, lur);
402                 if (IS_ERR(dtrq))
403                         RETURN(PTR_ERR(dtrq));
404
405                 spin_lock(&tdtd->tdtd_replay_list_lock);
406                 rc = dtrq_insert(tdtd, dtrq);
407                 if (rc < 0) {
408                         spin_unlock(&tdtd->tdtd_replay_list_lock);
409                         dtrq_destroy(dtrq);
410                         if (rc == -EEXIST)
411                                 goto again;
412                         return rc;
413                 }
414         } else {
415                 /* If the master transno in update header is not
416                 * matched with the one in the record, then it means
417                 * the dtrq is originally created by master record,
418                 * so we need update master transno and reposition
419                 * the dtrq(by master transno) in the list and also
420                 * replace update record */
421                 if (record->ur_master_transno != 0 &&
422                     dtrq->dtrq_master_transno != record->ur_master_transno &&
423                     dtrq->dtrq_lur != NULL) {
424                         list_del_init(&dtrq->dtrq_list);
425                         dtrq->dtrq_lur->lur_update_rec.ur_master_transno =
426                                                 record->ur_master_transno;
427
428                         dtrq->dtrq_master_transno = record->ur_master_transno;
429                         replace_record = true;
430                         /* try to insert again */
431                         rc = dtrq_insert(tdtd, dtrq);
432                         if (rc < 0) {
433                                 spin_unlock(&tdtd->tdtd_replay_list_lock);
434                                 dtrq_destroy(dtrq);
435                                 return rc;
436                         }
437                 }
438         }
439         spin_unlock(&tdtd->tdtd_replay_list_lock);
440
441         /* Because there should be only thread access the update record, so
442          * we do not need lock here */
443         if (replace_record) {
444                 /* Replace the update record and master transno */
445                 OBD_FREE(dtrq->dtrq_lur, dtrq->dtrq_lur_size);
446                 dtrq->dtrq_lur = NULL;
447                 dtrq->dtrq_lur_size = llog_update_record_size(lur);
448                 OBD_ALLOC_LARGE(dtrq->dtrq_lur, dtrq->dtrq_lur_size);
449                 if (dtrq->dtrq_lur == NULL)
450                         return -ENOMEM;
451
452                 memcpy(dtrq->dtrq_lur, lur, dtrq->dtrq_lur_size);
453         }
454
455         /* This is a partial update records, let's try to append
456          * the record to the current replay request */
457         if (record->ur_flags & UPDATE_RECORD_CONTINUE)
458                 rc = dtrq_append_updates(dtrq, record);
459
460         /* Then create and add sub update request */
461         rc = dtrq_sub_create_and_insert(dtrq, cookie, mdt_index);
462
463         RETURN(rc);
464 }
465 EXPORT_SYMBOL(insert_update_records_to_replay_list);
466
467 /**
468  * Dump updates of distribute txns.
469  *
470  * Output all of recovery updates in the distribute txn list to the
471  * debug log.
472  *
473  * \param[in] tdtd      distribute txn data where all of distribute txn
474  *                      are listed.
475  * \param[in] mask      debug mask
476  */
477 void dtrq_list_dump(struct target_distribute_txn_data *tdtd, unsigned int mask)
478 {
479         struct distribute_txn_replay_req *dtrq;
480
481         spin_lock(&tdtd->tdtd_replay_list_lock);
482         list_for_each_entry(dtrq, &tdtd->tdtd_replay_list, dtrq_list)
483                 update_records_dump(&dtrq->dtrq_lur->lur_update_rec, mask,
484                                     false);
485         spin_unlock(&tdtd->tdtd_replay_list_lock);
486 }
487 EXPORT_SYMBOL(dtrq_list_dump);
488
489 /**
490  * Destroy distribute txn replay req
491  *
492  * Destroy distribute txn replay req and all of subs.
493  *
494  * \param[in] dtrq      distribute txn replqy req to be destroyed.
495  */
496 void dtrq_destroy(struct distribute_txn_replay_req *dtrq)
497 {
498         struct distribute_txn_replay_req_sub    *dtrqs;
499         struct distribute_txn_replay_req_sub    *tmp;
500
501         LASSERT(list_empty(&dtrq->dtrq_list));
502         spin_lock(&dtrq->dtrq_sub_list_lock);
503         list_for_each_entry_safe(dtrqs, tmp, &dtrq->dtrq_sub_list, dtrqs_list) {
504                 struct sub_thandle_cookie *stc;
505                 struct sub_thandle_cookie *tmp;
506
507                 list_del(&dtrqs->dtrqs_list);
508                 list_for_each_entry_safe(stc, tmp, &dtrqs->dtrqs_cookie_list,
509                                          stc_list) {
510                         list_del(&stc->stc_list);
511                         OBD_FREE_PTR(stc);
512                 }
513                 OBD_FREE_PTR(dtrqs);
514         }
515         spin_unlock(&dtrq->dtrq_sub_list_lock);
516
517         if (dtrq->dtrq_lur != NULL)
518                 OBD_FREE_LARGE(dtrq->dtrq_lur, dtrq->dtrq_lur_size);
519
520         OBD_FREE_PTR(dtrq);
521 }
522 EXPORT_SYMBOL(dtrq_destroy);
523
524 /**
525  * Destroy all of replay req.
526  *
527  * Destroy all of replay req in the replay list.
528  *
529  * \param[in] tdtd      target distribute txn data where the replay list is.
530  */
531 void dtrq_list_destroy(struct target_distribute_txn_data *tdtd)
532 {
533         struct distribute_txn_replay_req *dtrq;
534         struct distribute_txn_replay_req *tmp;
535
536         spin_lock(&tdtd->tdtd_replay_list_lock);
537         list_for_each_entry_safe(dtrq, tmp, &tdtd->tdtd_replay_list,
538                                  dtrq_list) {
539                 list_del_init(&dtrq->dtrq_list);
540                 dtrq_destroy(dtrq);
541         }
542         spin_unlock(&tdtd->tdtd_replay_list_lock);
543 }
544 EXPORT_SYMBOL(dtrq_list_destroy);
545
546 /**
547  * Get next req in the replay list
548  *
549  * Get next req needs to be replayed, since it is a sorted list
550  * (by master MDT transno)
551  *
552  * \param[in] tdtd      distribute txn data where the replay list is
553  *
554  * \retval              the pointer of update recovery header
555  */
556 struct distribute_txn_replay_req *
557 distribute_txn_get_next_req(struct target_distribute_txn_data *tdtd)
558 {
559         struct distribute_txn_replay_req *dtrq = NULL;
560
561         spin_lock(&tdtd->tdtd_replay_list_lock);
562         if (!list_empty(&tdtd->tdtd_replay_list)) {
563                 dtrq = list_entry(tdtd->tdtd_replay_list.next,
564                                  struct distribute_txn_replay_req, dtrq_list);
565                 list_del_init(&dtrq->dtrq_list);
566         }
567         spin_unlock(&tdtd->tdtd_replay_list_lock);
568
569         return dtrq;
570 }
571 EXPORT_SYMBOL(distribute_txn_get_next_req);
572
573 /**
574  * Get next transno in the replay list, because this is the sorted
575  * list, so it will return the transno of next req in the list.
576  *
577  * \param[in] tdtd      distribute txn data where the replay list is
578  *
579  * \retval              the transno of next update in the list
580  */
581 __u64 distribute_txn_get_next_transno(struct target_distribute_txn_data *tdtd)
582 {
583         struct distribute_txn_replay_req        *dtrq = NULL;
584         __u64                                   transno = 0;
585
586         spin_lock(&tdtd->tdtd_replay_list_lock);
587         if (!list_empty(&tdtd->tdtd_replay_list)) {
588                 dtrq = list_entry(tdtd->tdtd_replay_list.next,
589                                  struct distribute_txn_replay_req, dtrq_list);
590                 transno = dtrq->dtrq_master_transno;
591         }
592         spin_unlock(&tdtd->tdtd_replay_list_lock);
593
594         CDEBUG(D_HA, "%s: Next update transno "LPU64"\n",
595                tdtd->tdtd_lut->lut_obd->obd_name, transno);
596         return transno;
597 }
598 EXPORT_SYMBOL(distribute_txn_get_next_transno);
599
600 /**
601  * Check if the update of one object is committed
602  *
603  * Check whether the update for the object is committed by checking whether
604  * the correspondent sub exists in the replay req. If it is committed, mark
605  * the committed flag in correspondent the sub thandle.
606  *
607  * \param[in] env       execution environment
608  * \param[in] dtrq      replay request
609  * \param[in] dt_obj    object for the update
610  * \param[in] top_th    top thandle
611  * \param[in] sub_th    sub thandle which the update belongs to
612  *
613  * \retval              1 if the update is not committed.
614  * \retval              0 if the update is committed.
615  * \retval              negative errno if some other failures happen.
616  */
617 static int update_is_committed(const struct lu_env *env,
618                                struct distribute_txn_replay_req *dtrq,
619                                struct dt_object *dt_obj,
620                                struct top_thandle *top_th,
621                                struct sub_thandle *st)
622 {
623         struct seq_server_site  *seq_site;
624         const struct lu_fid     *fid = lu_object_fid(&dt_obj->do_lu);
625         struct distribute_txn_replay_req_sub    *dtrqs;
626         __u32                   mdt_index;
627         ENTRY;
628
629         if (st->st_sub_th != NULL)
630                 RETURN(1);
631
632         if (st->st_committed)
633                 RETURN(0);
634
635         seq_site = lu_site2seq(dt_obj->do_lu.lo_dev->ld_site);
636         if (fid_is_update_log(fid) || fid_is_update_log_dir(fid)) {
637                 mdt_index = fid_oid(fid);
638         } else if (!fid_seq_in_fldb(fid_seq(fid))) {
639                 mdt_index = seq_site->ss_node_id;
640         } else {
641                 struct lu_server_fld *fld;
642                 struct lu_seq_range range = {0};
643                 int rc;
644
645                 fld = seq_site->ss_server_fld;
646                 fld_range_set_type(&range, LU_SEQ_RANGE_MDT);
647                 LASSERT(fld->lsf_seq_lookup != NULL);
648                 rc = fld->lsf_seq_lookup(env, fld, fid_seq(fid),
649                                          &range);
650                 if (rc < 0)
651                         RETURN(rc);
652                 mdt_index = range.lsr_index;
653         }
654
655         dtrqs = dtrq_sub_lookup(dtrq, mdt_index);
656         if (dtrqs != NULL || top_th->tt_multiple_thandle->tmt_committed) {
657                 st->st_committed = 1;
658                 if (dtrqs != NULL) {
659                         struct sub_thandle_cookie *stc;
660                         struct sub_thandle_cookie *tmp;
661
662                         list_for_each_entry_safe(stc, tmp,
663                                                  &dtrqs->dtrqs_cookie_list,
664                                                  stc_list)
665                                 list_move(&stc->stc_list, &st->st_cookie_list);
666                 }
667                 RETURN(0);
668         }
669
670         CDEBUG(D_HA, "Update of "DFID "on MDT%u is not committed\n", PFID(fid),
671                mdt_index);
672
673         RETURN(1);
674 }
675
676 /**
677  * Implementation of different update methods for update recovery.
678  *
679  * These following functions update_recovery_$(update_name) implement
680  * different updates recovery methods. They will extract the parameters
681  * from the common parameters area and call correspondent dt API to redo
682  * the update.
683  *
684  * \param[in] env       execution environment
685  * \param[in] op        update operation to be replayed
686  * \param[in] params    common update parameters which holds all parameters
687  *                      of the operation
688  * \param[in] th        transaction handle
689  * \param[in] declare   indicate it will do declare or real execution, true
690  *                      means declare, false means real execution
691  *
692  * \retval              0 if it succeeds.
693  * \retval              negative errno if it fails.
694  */
695 static int update_recovery_create(const struct lu_env *env,
696                                   struct dt_object *dt_obj,
697                                   const struct update_op *op,
698                                   const struct update_params *params,
699                                   struct thandle_exec_args *ta,
700                                   struct thandle *th)
701 {
702         struct update_thread_info *uti = update_env_info(env);
703         struct llog_update_record *lur = uti->uti_dtrq->dtrq_lur;
704         struct lu_attr          *attr = &uti->uti_attr;
705         struct obdo             *wobdo;
706         struct obdo             *lobdo = &uti->uti_obdo;
707         struct dt_object_format dof;
708         __u16                   size;
709         unsigned int            param_count;
710         int rc;
711         ENTRY;
712
713         if (dt_object_exists(dt_obj))
714                 RETURN(-EEXIST);
715
716         param_count = lur->lur_update_rec.ur_param_count;
717         wobdo = update_params_get_param_buf(params, op->uop_params_off[0],
718                                             param_count, &size);
719         if (wobdo == NULL)
720                 RETURN(-EIO);
721         if (size != sizeof(*wobdo))
722                 RETURN(-EIO);
723
724         if (LLOG_REC_HDR_NEEDS_SWABBING(&lur->lur_hdr))
725                 lustre_swab_obdo(wobdo);
726
727         lustre_get_wire_obdo(NULL, lobdo, wobdo);
728         la_from_obdo(attr, lobdo, lobdo->o_valid);
729
730         dof.dof_type = dt_mode_to_dft(attr->la_mode);
731
732         rc = out_tx_create(env, dt_obj, attr, NULL, &dof,
733                            ta, th, NULL, 0);
734
735         RETURN(rc);
736 }
737
738 static int update_recovery_destroy(const struct lu_env *env,
739                                    struct dt_object *dt_obj,
740                                    const struct update_op *op,
741                                    const struct update_params *params,
742                                    struct thandle_exec_args *ta,
743                                    struct thandle *th)
744 {
745         int rc;
746         ENTRY;
747
748         rc = out_tx_destroy(env, dt_obj, ta, th, NULL, 0);
749
750         RETURN(rc);
751 }
752
753 static int update_recovery_ref_add(const struct lu_env *env,
754                                    struct dt_object *dt_obj,
755                                    const struct update_op *op,
756                                    const struct update_params *params,
757                                    struct thandle_exec_args *ta,
758                                    struct thandle *th)
759 {
760         int rc;
761         ENTRY;
762
763         rc = out_tx_ref_add(env, dt_obj, ta, th, NULL, 0);
764
765         RETURN(rc);
766 }
767
768 static int update_recovery_ref_del(const struct lu_env *env,
769                                    struct dt_object *dt_obj,
770                                    const struct update_op *op,
771                                    const struct update_params *params,
772                                    struct thandle_exec_args *ta,
773                                    struct thandle *th)
774 {
775         int rc;
776         ENTRY;
777
778         rc = out_tx_ref_del(env, dt_obj, ta, th, NULL, 0);
779
780         RETURN(rc);
781 }
782
783 static int update_recovery_attr_set(const struct lu_env *env,
784                                     struct dt_object *dt_obj,
785                                     const struct update_op *op,
786                                     const struct update_params *params,
787                                     struct thandle_exec_args *ta,
788                                     struct thandle *th)
789 {
790         struct update_thread_info *uti = update_env_info(env);
791         struct llog_update_record *lur = uti->uti_dtrq->dtrq_lur;
792         struct obdo     *wobdo;
793         struct obdo     *lobdo = &uti->uti_obdo;
794         struct lu_attr  *attr = &uti->uti_attr;
795         __u16           size;
796         unsigned int    param_count;
797         int             rc;
798         ENTRY;
799
800         param_count = lur->lur_update_rec.ur_param_count;
801         wobdo = update_params_get_param_buf(params, op->uop_params_off[0],
802                                             param_count, &size);
803         if (wobdo == NULL)
804                 RETURN(-EIO);
805         if (size != sizeof(*wobdo))
806                 RETURN(-EIO);
807
808         if (LLOG_REC_HDR_NEEDS_SWABBING(&lur->lur_hdr))
809                 lustre_swab_obdo(wobdo);
810
811         lustre_get_wire_obdo(NULL, lobdo, wobdo);
812         la_from_obdo(attr, lobdo, lobdo->o_valid);
813
814         rc = out_tx_attr_set(env, dt_obj, attr, ta, th, NULL, 0);
815
816         RETURN(rc);
817 }
818
819 static int update_recovery_xattr_set(const struct lu_env *env,
820                                      struct dt_object *dt_obj,
821                                      const struct update_op *op,
822                                      const struct update_params *params,
823                                      struct thandle_exec_args *ta,
824                                      struct thandle *th)
825 {
826         struct update_thread_info *uti = update_env_info(env);
827         char            *buf;
828         char            *name;
829         int             fl;
830         __u16           size;
831         __u32           param_count;
832         int             rc;
833         ENTRY;
834
835         param_count = uti->uti_dtrq->dtrq_lur->lur_update_rec.ur_param_count;
836         name = update_params_get_param_buf(params,
837                                            op->uop_params_off[0],
838                                            param_count, &size);
839         if (name == NULL)
840                 RETURN(-EIO);
841
842         buf = update_params_get_param_buf(params,
843                                           op->uop_params_off[1],
844                                           param_count, &size);
845         if (buf == NULL)
846                 RETURN(-EIO);
847
848         uti->uti_buf.lb_buf = buf;
849         uti->uti_buf.lb_len = (size_t)size;
850
851         buf = update_params_get_param_buf(params, op->uop_params_off[2],
852                                           param_count, &size);
853         if (buf == NULL)
854                 RETURN(-EIO);
855         if (size != sizeof(fl))
856                 RETURN(-EIO);
857
858         fl = le32_to_cpu(*(int *)buf);
859
860         rc = out_tx_xattr_set(env, dt_obj, &uti->uti_buf, name, fl, ta, th,
861                               NULL, 0);
862
863         RETURN(rc);
864 }
865
866 static int update_recovery_index_insert(const struct lu_env *env,
867                                         struct dt_object *dt_obj,
868                                         const struct update_op *op,
869                                         const struct update_params *params,
870                                         struct thandle_exec_args *ta,
871                                         struct thandle *th)
872 {
873         struct update_thread_info *uti = update_env_info(env);
874         struct lu_fid           *fid;
875         char                    *name;
876         __u32                   param_count;
877         __u32                   *ptype;
878         __u32                   type;
879         __u16                   size;
880         int rc;
881         ENTRY;
882
883         param_count = uti->uti_dtrq->dtrq_lur->lur_update_rec.ur_param_count;
884         name = update_params_get_param_buf(params, op->uop_params_off[0],
885                                            param_count, &size);
886         if (name == NULL)
887                 RETURN(-EIO);
888
889         fid = update_params_get_param_buf(params, op->uop_params_off[1],
890                                           param_count, &size);
891         if (fid == NULL)
892                 RETURN(-EIO);
893         if (size != sizeof(*fid))
894                 RETURN(-EIO);
895
896         fid_le_to_cpu(fid, fid);
897
898         ptype = update_params_get_param_buf(params, op->uop_params_off[2],
899                                             param_count, &size);
900         if (ptype == NULL)
901                 RETURN(-EIO);
902         if (size != sizeof(*ptype))
903                 RETURN(-EIO);
904         type = le32_to_cpu(*ptype);
905
906         if (dt_try_as_dir(env, dt_obj) == 0)
907                 RETURN(-ENOTDIR);
908
909         uti->uti_rec.rec_fid = fid;
910         uti->uti_rec.rec_type = type;
911
912         rc = out_tx_index_insert(env, dt_obj,
913                                  (const struct dt_rec *)&uti->uti_rec,
914                                  (const struct dt_key *)name, ta, th,
915                                  NULL, 0);
916
917         RETURN(rc);
918 }
919
920 static int update_recovery_index_delete(const struct lu_env *env,
921                                         struct dt_object *dt_obj,
922                                         const struct update_op *op,
923                                         const struct update_params *params,
924                                         struct thandle_exec_args *ta,
925                                         struct thandle *th)
926 {
927         struct update_thread_info *uti = update_env_info(env);
928         __u32   param_count;
929         char    *name;
930         __u16   size;
931         int     rc;
932         ENTRY;
933
934         param_count = uti->uti_dtrq->dtrq_lur->lur_update_rec.ur_param_count;
935         name = update_params_get_param_buf(params, op->uop_params_off[0],
936                                            param_count, &size);
937         if (name == NULL)
938                 RETURN(-EIO);
939
940         if (dt_try_as_dir(env, dt_obj) == 0)
941                 RETURN(-ENOTDIR);
942
943         rc = out_tx_index_delete(env, dt_obj,
944                                  (const struct dt_key *)name, ta, th, NULL, 0);
945
946         RETURN(rc);
947 }
948
949 static int update_recovery_write(const struct lu_env *env,
950                                  struct dt_object *dt_obj,
951                                  const struct update_op *op,
952                                  const struct update_params *params,
953                                  struct thandle_exec_args *ta,
954                                  struct thandle *th)
955 {
956         struct update_thread_info *uti = update_env_info(env);
957         char            *buf;
958         __u32           param_count;
959         __u64           pos;
960         __u16           size;
961         int rc;
962         ENTRY;
963
964         param_count = uti->uti_dtrq->dtrq_lur->lur_update_rec.ur_param_count;
965         buf = update_params_get_param_buf(params, op->uop_params_off[0],
966                                           param_count, &size);
967         if (buf == NULL)
968                 RETURN(-EIO);
969
970         uti->uti_buf.lb_buf = buf;
971         uti->uti_buf.lb_len = size;
972
973         buf = update_params_get_param_buf(params, op->uop_params_off[1],
974                                           param_count, &size);
975         if (buf == NULL)
976                 RETURN(-EIO);
977
978         pos = le64_to_cpu(*(__u64 *)buf);
979
980         rc = out_tx_write(env, dt_obj, &uti->uti_buf, pos,
981                           ta, th, NULL, 0);
982
983         RETURN(rc);
984 }
985
986 static int update_recovery_xattr_del(const struct lu_env *env,
987                                      struct dt_object *dt_obj,
988                                      const struct update_op *op,
989                                      const struct update_params *params,
990                                      struct thandle_exec_args *ta,
991                                      struct thandle *th)
992 {
993         struct update_thread_info *uti = update_env_info(env);
994         __u32   param_count;
995         char    *name;
996         __u16   size;
997         int     rc;
998         ENTRY;
999
1000         param_count = uti->uti_dtrq->dtrq_lur->lur_update_rec.ur_param_count;
1001         name = update_params_get_param_buf(params, op->uop_params_off[0],
1002                                            param_count, &size);
1003         if (name == NULL)
1004                 RETURN(-EIO);
1005
1006         rc = out_tx_xattr_del(env, dt_obj, name, ta, th, NULL, 0);
1007
1008         RETURN(rc);
1009 }
1010
1011 /**
1012  * Update session information
1013  *
1014  * Update session information so tgt_txn_stop_cb()->tgt_last_rcvd_update()
1015  * can be called correctly during update replay.
1016  *
1017  * \param[in] env       execution environment.
1018  * \param[in] tdtd      distribute data structure of the recovering tgt.
1019  * \param[in] th        thandle of this update replay.
1020  * \param[in] master_th master sub thandle.
1021  * \param[in] ta_arg    the tx arg structure to hold the update for updating
1022  *                      reply data.
1023  */
1024 static void update_recovery_update_ses(struct lu_env *env,
1025                                       struct target_distribute_txn_data *tdtd,
1026                                       struct thandle *th,
1027                                       struct thandle *master_th,
1028                                       struct tx_arg *ta_arg)
1029 {
1030         struct tgt_session_info *tsi;
1031         struct lu_target        *lut = tdtd->tdtd_lut;
1032         struct obd_export       *export;
1033         struct cfs_hash         *hash;
1034         struct top_thandle      *top_th;
1035         struct lsd_reply_data   *lrd;
1036         size_t                  size;
1037
1038         tsi = tgt_ses_info(env);
1039         if (tsi->tsi_exp != NULL)
1040                 return;
1041
1042         size = ta_arg->u.write.buf.lb_len;
1043         lrd = ta_arg->u.write.buf.lb_buf;
1044         if (size != sizeof(*lrd) || lrd == NULL)
1045                 return;
1046
1047         lrd->lrd_transno         = le64_to_cpu(lrd->lrd_transno);
1048         lrd->lrd_xid             = le64_to_cpu(lrd->lrd_xid);
1049         lrd->lrd_data            = le64_to_cpu(lrd->lrd_data);
1050         lrd->lrd_result          = le32_to_cpu(lrd->lrd_result);
1051         lrd->lrd_client_gen      = le32_to_cpu(lrd->lrd_client_gen);
1052
1053         if (lrd->lrd_transno != tgt_th_info(env)->tti_transno)
1054                 return;
1055
1056         hash = cfs_hash_getref(lut->lut_obd->obd_gen_hash);
1057         if (hash == NULL)
1058                 return;
1059
1060         export = cfs_hash_lookup(hash, &lrd->lrd_client_gen);
1061         if (export == NULL) {
1062                 cfs_hash_putref(hash);
1063                 return;
1064         }
1065
1066         tsi->tsi_exp = export;
1067         tsi->tsi_xid = lrd->lrd_xid;
1068         tsi->tsi_opdata = lrd->lrd_data;
1069         tsi->tsi_result = lrd->lrd_result;
1070         tsi->tsi_client_gen = lrd->lrd_client_gen;
1071         top_th = container_of(th, struct top_thandle, tt_super);
1072         top_th->tt_master_sub_thandle = master_th;
1073         cfs_hash_putref(hash);
1074 }
1075
1076 /**
1077  * Execute updates in the update replay records
1078  *
1079  * Declare distribute txn replay by update records and add the updates
1080  * to the execution list. Note: it will check if the update has been
1081  * committed, and only execute the updates if it is not committed to
1082  * disk.
1083  *
1084  * \param[in] env       execution environment
1085  * \param[in] tdtd      distribute txn replay data which hold all of replay
1086  *                      reqs and all replay parameters.
1087  * \param[in] dtrq      distribute transaction replay req.
1088  * \param[in] ta        thandle execute args.
1089  *
1090  * \retval              0 if declare succeeds.
1091  * \retval              negative errno if declare fails.
1092  */
1093 static int update_recovery_exec(const struct lu_env *env,
1094                                 struct target_distribute_txn_data *tdtd,
1095                                 struct distribute_txn_replay_req *dtrq,
1096                                 struct thandle_exec_args *ta)
1097 {
1098         struct llog_update_record *lur = dtrq->dtrq_lur;
1099         struct update_records   *records = &lur->lur_update_rec;
1100         struct update_ops       *ops = &records->ur_ops;
1101         struct update_params    *params = update_records_get_params(records);
1102         struct top_thandle      *top_th = container_of(ta->ta_handle,
1103                                                        struct top_thandle,
1104                                                        tt_super);
1105         struct top_multiple_thandle *tmt = top_th->tt_multiple_thandle;
1106         struct update_op        *op;
1107         unsigned int            i;
1108         int                     rc = 0;
1109         ENTRY;
1110
1111         /* These records have been swabbed in llog_cat_process() */
1112         for (i = 0, op = &ops->uops_op[0]; i < records->ur_update_count;
1113              i++, op = update_op_next_op(op)) {
1114                 struct lu_fid           *fid = &op->uop_fid;
1115                 struct dt_object        *dt_obj;
1116                 struct dt_object        *sub_dt_obj;
1117                 struct dt_device        *sub_dt;
1118                 struct sub_thandle      *st;
1119
1120                 if (op->uop_type == OUT_NOOP)
1121                         continue;
1122
1123                 dt_obj = dt_locate(env, tdtd->tdtd_dt, fid);
1124                 if (IS_ERR(dt_obj)) {
1125                         rc = PTR_ERR(dt_obj);
1126                         break;
1127                 }
1128                 sub_dt_obj = dt_object_child(dt_obj);
1129
1130                 /* Create sub thandle if not */
1131                 sub_dt = lu2dt_dev(sub_dt_obj->do_lu.lo_dev);
1132                 st = lookup_sub_thandle(tmt, sub_dt);
1133                 if (st == NULL) {
1134                         st = create_sub_thandle(tmt, sub_dt);
1135                         if (IS_ERR(st))
1136                                 GOTO(next, rc = PTR_ERR(st));
1137                 }
1138
1139                 /* check if updates on the OSD/OSP are committed */
1140                 rc = update_is_committed(env, dtrq, dt_obj, top_th, st);
1141                 if (rc == 0)
1142                         /* If this is committed, goto next */
1143                         goto next;
1144
1145                 if (rc < 0)
1146                         GOTO(next, rc);
1147
1148                 /* Create thandle for sub thandle if needed */
1149                 if (st->st_sub_th == NULL) {
1150                         rc = sub_thandle_trans_create(env, top_th, st);
1151                         if (rc != 0)
1152                                 GOTO(next, rc);
1153                 }
1154
1155                 CDEBUG(D_HA, "replay %uth update\n", i);
1156                 switch (op->uop_type) {
1157                 case OUT_CREATE:
1158                         rc = update_recovery_create(env, sub_dt_obj,
1159                                                     op, params, ta,
1160                                                     st->st_sub_th);
1161                         break;
1162                 case OUT_DESTROY:
1163                         rc = update_recovery_destroy(env, sub_dt_obj,
1164                                                      op, params, ta,
1165                                                      st->st_sub_th);
1166                         break;
1167                 case OUT_REF_ADD:
1168                         rc = update_recovery_ref_add(env, sub_dt_obj,
1169                                                      op, params, ta,
1170                                                      st->st_sub_th);
1171                         break;
1172                 case OUT_REF_DEL:
1173                         rc = update_recovery_ref_del(env, sub_dt_obj,
1174                                                      op, params, ta,
1175                                                      st->st_sub_th);
1176                         break;
1177                 case OUT_ATTR_SET:
1178                         rc = update_recovery_attr_set(env, sub_dt_obj,
1179                                                       op, params, ta,
1180                                                       st->st_sub_th);
1181                         break;
1182                 case OUT_XATTR_SET:
1183                         rc = update_recovery_xattr_set(env, sub_dt_obj,
1184                                                        op, params, ta,
1185                                                        st->st_sub_th);
1186                         break;
1187                 case OUT_INDEX_INSERT:
1188                         rc = update_recovery_index_insert(env, sub_dt_obj,
1189                                                           op, params, ta,
1190                                                           st->st_sub_th);
1191                         break;
1192                 case OUT_INDEX_DELETE:
1193                         rc = update_recovery_index_delete(env, sub_dt_obj,
1194                                                           op, params, ta,
1195                                                           st->st_sub_th);
1196                         break;
1197                 case OUT_WRITE:
1198                         rc = update_recovery_write(env, sub_dt_obj,
1199                                                    op, params, ta,
1200                                                    st->st_sub_th);
1201                         break;
1202                 case OUT_XATTR_DEL:
1203                         rc = update_recovery_xattr_del(env, sub_dt_obj,
1204                                                        op, params, ta,
1205                                                        st->st_sub_th);
1206                         break;
1207                 default:
1208                         CERROR("Unknown update type %u\n", (__u32)op->uop_type);
1209                         rc = -EINVAL;
1210                         break;
1211                 }
1212 next:
1213                 lu_object_put(env, &dt_obj->do_lu);
1214                 if (rc < 0)
1215                         break;
1216         }
1217
1218         ta->ta_handle->th_result = rc;
1219         RETURN(rc);
1220 }
1221
1222 /**
1223  * redo updates on MDT if needed.
1224  *
1225  * During DNE recovery, the recovery thread (target_recovery_thread) will call
1226  * this function to replay distribute txn updates on all MDTs. It only replay
1227  * updates on the MDT where the update record is missing.
1228  *
1229  * If the update already exists on the MDT, then it does not need replay the
1230  * updates on that MDT, and only mark the sub transaction has been committed
1231  * there.
1232  *
1233  * \param[in] env       execution environment
1234  * \param[in] tdtd      target distribute txn data, which holds the replay list
1235  *                      and all parameters needed by replay process.
1236  * \param[in] dtrq      distribute txn replay req.
1237  *
1238  * \retval              0 if replay succeeds.
1239  * \retval              negative errno if replay failes.
1240  */
1241 int distribute_txn_replay_handle(struct lu_env *env,
1242                                  struct target_distribute_txn_data *tdtd,
1243                                  struct distribute_txn_replay_req *dtrq)
1244 {
1245         struct update_records   *records = &dtrq->dtrq_lur->lur_update_rec;
1246         struct thandle_exec_args *ta;
1247         struct lu_context       session_env;
1248         struct thandle          *th = NULL;
1249         struct top_thandle      *top_th;
1250         struct top_multiple_thandle *tmt;
1251         struct thandle_update_records *tur = NULL;
1252         int                     i;
1253         int                     rc = 0;
1254         ENTRY;
1255
1256         /* initialize session, it is needed for the handler of target */
1257         rc = lu_context_init(&session_env, LCT_SERVER_SESSION | LCT_NOREF);
1258         if (rc) {
1259                 CERROR("%s: failure to initialize session: rc = %d\n",
1260                        tdtd->tdtd_lut->lut_obd->obd_name, rc);
1261                 RETURN(rc);
1262         }
1263         lu_context_enter(&session_env);
1264         env->le_ses = &session_env;
1265         lu_env_refill(env);
1266         update_records_dump(records, D_HA, true);
1267         th = top_trans_create(env, NULL);
1268         if (IS_ERR(th))
1269                 GOTO(exit_session, rc = PTR_ERR(th));
1270
1271         ta = &update_env_info(env)->uti_tea;
1272         ta->ta_argno = 0;
1273
1274         update_env_info(env)->uti_dtrq = dtrq;
1275         /* Create distribute transaction structure for this top thandle */
1276         top_th = container_of(th, struct top_thandle, tt_super);
1277         rc = top_trans_create_tmt(env, top_th);
1278         if (rc < 0)
1279                 GOTO(stop_trans, rc);
1280
1281         th->th_dev = tdtd->tdtd_dt;
1282         ta->ta_handle = th;
1283
1284         /* check if the distribute transaction has been committed */
1285         tmt = top_th->tt_multiple_thandle;
1286         tmt->tmt_master_sub_dt = tdtd->tdtd_lut->lut_bottom;
1287         tmt->tmt_batchid = dtrq->dtrq_batchid;
1288         tgt_th_info(env)->tti_transno = dtrq->dtrq_master_transno;
1289
1290         if (tmt->tmt_batchid <= tdtd->tdtd_committed_batchid)
1291                 tmt->tmt_committed = 1;
1292
1293         rc = update_recovery_exec(env, tdtd, dtrq, ta);
1294         if (rc < 0)
1295                 GOTO(stop_trans, rc);
1296
1297         /* If no updates are needed to be replayed, then mark this records as
1298          * committed, so commit thread distribute_txn_commit_thread() will
1299          * delete the record */
1300         if (ta->ta_argno == 0)
1301                 tmt->tmt_committed = 1;
1302
1303         tur = &update_env_info(env)->uti_tur;
1304         tur->tur_update_records = dtrq->dtrq_lur;
1305         tur->tur_update_records_buf_size = dtrq->dtrq_lur_size;
1306         tur->tur_update_params = NULL;
1307         tur->tur_update_param_count = 0;
1308         tmt->tmt_update_records = tur;
1309
1310         distribute_txn_insert_by_batchid(tmt);
1311         rc = top_trans_start(env, NULL, th);
1312         if (rc < 0)
1313                 GOTO(stop_trans, rc);
1314
1315         for (i = 0; i < ta->ta_argno; i++) {
1316                 struct tx_arg           *ta_arg;
1317                 struct dt_object        *dt_obj;
1318                 struct dt_device        *sub_dt;
1319                 struct sub_thandle      *st;
1320
1321                 ta_arg = ta->ta_args[i];
1322                 dt_obj = ta_arg->object;
1323
1324                 LASSERT(tmt->tmt_committed == 0);
1325                 sub_dt = lu2dt_dev(dt_obj->do_lu.lo_dev);
1326                 st = lookup_sub_thandle(tmt, sub_dt);
1327
1328                 LASSERT(st != NULL);
1329                 LASSERT(st->st_sub_th != NULL);
1330                 rc = ta->ta_args[i]->exec_fn(env, st->st_sub_th,
1331                                              ta->ta_args[i]);
1332
1333                 /* If the update is to update the reply data, then
1334                  * we need set the session information, so
1335                  * tgt_last_rcvd_update() can be called correctly */
1336                 if (rc == 0 && dt_obj == tdtd->tdtd_lut->lut_reply_data)
1337                         update_recovery_update_ses(env, tdtd, th,
1338                                                    st->st_sub_th, ta_arg);
1339
1340                 if (unlikely(rc < 0)) {
1341                         CDEBUG(D_HA, "error during execution of #%u from"
1342                                " %s:%d: rc = %d\n", i, ta->ta_args[i]->file,
1343                                ta->ta_args[i]->line, rc);
1344                         while (--i > 0) {
1345                                 if (ta->ta_args[i]->undo_fn != NULL) {
1346                                         dt_obj = ta->ta_args[i]->object;
1347                                         sub_dt =
1348                                                 lu2dt_dev(dt_obj->do_lu.lo_dev);
1349                                         st = lookup_sub_thandle(tmt, sub_dt);
1350                                         LASSERT(st != NULL);
1351                                         LASSERT(st->st_sub_th != NULL);
1352
1353                                         ta->ta_args[i]->undo_fn(env,
1354                                                                st->st_sub_th,
1355                                                                ta->ta_args[i]);
1356                                 } else {
1357                                         CERROR("%s: undo for %s:%d: rc = %d\n",
1358                                              dt_obd_name(ta->ta_handle->th_dev),
1359                                                ta->ta_args[i]->file,
1360                                                ta->ta_args[i]->line, -ENOTSUPP);
1361                                 }
1362                         }
1363                         break;
1364                 }
1365                 CDEBUG(D_HA, "%s: executed %u/%u: rc = %d\n",
1366                        dt_obd_name(sub_dt), i, ta->ta_argno, rc);
1367         }
1368
1369 stop_trans:
1370         if (rc < 0)
1371                 th->th_result = rc;
1372         rc = top_trans_stop(env, tdtd->tdtd_dt, th);
1373         for (i = 0; i < ta->ta_argno; i++) {
1374                 if (ta->ta_args[i]->object != NULL) {
1375                         lu_object_put(env, &ta->ta_args[i]->object->do_lu);
1376                         ta->ta_args[i]->object = NULL;
1377                 }
1378         }
1379
1380         if (tur != NULL)
1381                 tur->tur_update_records = NULL;
1382
1383         if (tgt_ses_info(env)->tsi_exp != NULL) {
1384                 class_export_put(tgt_ses_info(env)->tsi_exp);
1385                 tgt_ses_info(env)->tsi_exp = NULL;
1386         }
1387 exit_session:
1388         lu_context_exit(&session_env);
1389         lu_context_fini(&session_env);
1390         RETURN(rc);
1391 }
1392 EXPORT_SYMBOL(distribute_txn_replay_handle);