Whamcloud - gitweb
LU-14393 recovery: reply reconstruction for batched RPCs
[fs/lustre-release.git] / lustre / target / out_handler.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) 2013, 2017, Intel Corporation.
24  *
25  * lustre/target/out_handler.c
26  *
27  * Object update handler between targets.
28  *
29  * Author: di.wang <di.wang@intel.com>
30  */
31
32 #define DEBUG_SUBSYSTEM S_CLASS
33
34 #include <llog_swab.h>
35 #include <lustre_obdo.h>
36 #include <lustre_swab.h>
37 #include <lustre_update.h>
38 #include <md_object.h>
39 #include <obd_class.h>
40 #include "tgt_internal.h"
41
42 static inline void orr_cpu_to_le(struct out_read_reply *orr_dst,
43                                  const struct out_read_reply *orr_src)
44 {
45         orr_dst->orr_size = cpu_to_le32(orr_src->orr_size);
46         orr_dst->orr_padding = cpu_to_le32(orr_src->orr_padding);
47         orr_dst->orr_offset = cpu_to_le64(orr_dst->orr_offset);
48 }
49
50 static void out_reconstruct(const struct lu_env *env, struct dt_device *dt,
51                             struct dt_object *obj,
52                             struct object_update_reply *reply,
53                             int index)
54 {
55         CDEBUG(D_HA, "%s: fork reply reply %p index %d: rc = %d\n",
56                dt_obd_name(dt), reply, index, 0);
57
58         object_update_result_insert(reply, NULL, 0, index, 0);
59 }
60
61 typedef void (*out_reconstruct_t)(const struct lu_env *env,
62                                   struct dt_device *dt,
63                                   struct dt_object *obj,
64                                   struct object_update_reply *reply,
65                                   int index);
66
67 static int out_create(struct tgt_session_info *tsi)
68 {
69         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
70         struct object_update    *update = tti->tti_u.update.tti_update;
71         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
72         struct dt_object_format *dof = &tti->tti_u.update.tti_update_dof;
73         struct obdo             *lobdo = &tti->tti_u.update.tti_obdo;
74         struct lu_attr          *attr = &tti->tti_attr;
75         struct lu_fid           *fid = NULL;
76         struct obdo             *wobdo;
77         size_t                  size;
78         int                     rc;
79
80         ENTRY;
81
82         wobdo = object_update_param_get(update, 0, &size);
83         if (IS_ERR(wobdo) || size != sizeof(*wobdo)) {
84                 CERROR("%s: obdo is NULL, invalid RPC: rc = %ld\n",
85                        tgt_name(tsi->tsi_tgt), PTR_ERR(wobdo));
86                 RETURN(PTR_ERR(wobdo));
87         }
88
89         if (req_capsule_req_need_swab(tsi->tsi_pill))
90                 lustre_swab_obdo(wobdo);
91         lustre_get_wire_obdo(NULL, lobdo, wobdo);
92         la_from_obdo(attr, lobdo, lobdo->o_valid);
93
94         dof->dof_type = dt_mode_to_dft(attr->la_mode);
95         if (update->ou_params_count > 1) {
96                 fid = object_update_param_get(update, 1, &size);
97                 if (IS_ERR(fid) || size != sizeof(*fid)) {
98                         CERROR("%s: invalid fid: rc = %ld\n",
99                                tgt_name(tsi->tsi_tgt), PTR_ERR(fid));
100                         RETURN(PTR_ERR(fid));
101                 }
102                 if (req_capsule_req_need_swab(tsi->tsi_pill))
103                         lustre_swab_lu_fid(fid);
104                 if (!fid_is_sane(fid)) {
105                         CERROR("%s: invalid fid "DFID": rc = %d\n",
106                                tgt_name(tsi->tsi_tgt), PFID(fid), -EPROTO);
107                         RETURN(-EPROTO);
108                 }
109         }
110
111         if (lu_object_exists(&obj->do_lu))
112                 RETURN(-EEXIST);
113
114         rc = out_tx_create(tsi->tsi_env, obj, attr, fid, dof,
115                            &tti->tti_tea, tti->tti_tea.ta_handle,
116                            tti->tti_u.update.tti_update_reply,
117                            tti->tti_u.update.tti_update_reply_index);
118
119         RETURN(rc);
120 }
121
122 static int out_attr_set(struct tgt_session_info *tsi)
123 {
124         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
125         struct object_update    *update = tti->tti_u.update.tti_update;
126         struct lu_attr          *attr = &tti->tti_attr;
127         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
128         struct obdo             *lobdo = &tti->tti_u.update.tti_obdo;
129         struct obdo             *wobdo;
130         size_t                   size;
131         int                      rc;
132
133         ENTRY;
134
135         wobdo = object_update_param_get(update, 0, &size);
136         if (IS_ERR(wobdo) || size != sizeof(*wobdo)) {
137                 CERROR("%s: empty obdo in the update: rc = %ld\n",
138                        tgt_name(tsi->tsi_tgt), PTR_ERR(wobdo));
139                 RETURN(PTR_ERR(wobdo));
140         }
141
142         attr->la_valid = 0;
143         attr->la_valid = 0;
144
145         if (req_capsule_req_need_swab(tsi->tsi_pill))
146                 lustre_swab_obdo(wobdo);
147         lustre_get_wire_obdo(NULL, lobdo, wobdo);
148         la_from_obdo(attr, lobdo, lobdo->o_valid);
149
150         rc = out_tx_attr_set(tsi->tsi_env, obj, attr, &tti->tti_tea,
151                              tti->tti_tea.ta_handle,
152                              tti->tti_u.update.tti_update_reply,
153                              tti->tti_u.update.tti_update_reply_index);
154
155         RETURN(rc);
156 }
157
158 static int out_attr_get(struct tgt_session_info *tsi)
159 {
160         const struct lu_env     *env = tsi->tsi_env;
161         struct tgt_thread_info  *tti = tgt_th_info(env);
162         struct object_update    *update = tti->tti_u.update.tti_update;
163         struct obdo             *obdo = &tti->tti_u.update.tti_obdo;
164         struct lu_attr          *la = &tti->tti_attr;
165         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
166         int                     idx = tti->tti_u.update.tti_update_reply_index;
167         int                     rc;
168
169         ENTRY;
170
171         if (unlikely(update->ou_result_size < sizeof(*obdo)))
172                 return -EPROTO;
173
174         if (!lu_object_exists(&obj->do_lu)) {
175                 /* Usually, this will be called when the master MDT try
176                  * to init a remote object(see osp_object_init), so if
177                  * the object does not exist on slave, we need set BANSHEE flag,
178                  * so the object can be removed from the cache immediately */
179                 set_bit(LU_OBJECT_HEARD_BANSHEE,
180                         &obj->do_lu.lo_header->loh_flags);
181                 RETURN(-ENOENT);
182         }
183
184         dt_read_lock(env, obj, DT_TGT_CHILD);
185         rc = dt_attr_get(env, obj, la);
186         if (rc)
187                 GOTO(out_unlock, rc);
188
189         obdo->o_valid = 0;
190         obdo_from_la(obdo, la, la->la_valid);
191
192 out_unlock:
193         dt_read_unlock(env, obj);
194
195         CDEBUG(D_INFO, "%s: insert attr get reply %p index %d: rc = %d\n",
196                tgt_name(tsi->tsi_tgt), tti->tti_u.update.tti_update_reply,
197                0, rc);
198
199         object_update_result_insert(tti->tti_u.update.tti_update_reply, obdo,
200                                     sizeof(*obdo), idx, rc);
201
202         RETURN(rc);
203 }
204
205 static int out_xattr_get(struct tgt_session_info *tsi)
206 {
207         const struct lu_env        *env = tsi->tsi_env;
208         struct tgt_thread_info     *tti = tgt_th_info(env);
209         struct object_update       *update = tti->tti_u.update.tti_update;
210         struct lu_buf              *lbuf = &tti->tti_buf;
211         struct object_update_reply *reply = tti->tti_u.update.tti_update_reply;
212         struct dt_object           *obj = tti->tti_u.update.tti_dt_object;
213         char                       *name;
214         struct object_update_result *update_result;
215         int                     idx = tti->tti_u.update.tti_update_reply_index;
216         int                        rc;
217
218         ENTRY;
219
220         if (!lu_object_exists(&obj->do_lu)) {
221                 set_bit(LU_OBJECT_HEARD_BANSHEE,
222                         &obj->do_lu.lo_header->loh_flags);
223                 RETURN(-ENOENT);
224         }
225
226         name = object_update_param_get(update, 0, NULL);
227         if (IS_ERR(name)) {
228                 CERROR("%s: empty name for xattr get: rc = %ld\n",
229                        tgt_name(tsi->tsi_tgt), PTR_ERR(name));
230                 RETURN(PTR_ERR(name));
231         }
232
233         update_result = object_update_result_get(reply, idx, NULL);
234         if (update_result == NULL) {
235                 CERROR("%s: empty name for xattr get: rc = %d\n",
236                        tgt_name(tsi->tsi_tgt), -EPROTO);
237                 RETURN(-EPROTO);
238         }
239
240         lbuf->lb_len = (int)tti->tti_u.update.tti_update->ou_result_size;
241         if (lbuf->lb_len == 0)
242                 lbuf->lb_buf = NULL;
243         else
244                 lbuf->lb_buf = update_result->our_data;
245
246         dt_read_lock(env, obj, DT_TGT_CHILD);
247         rc = dt_xattr_get(env, obj, lbuf, name);
248         dt_read_unlock(env, obj);
249         if (rc <= 0) {
250                 lbuf->lb_len = 0;
251                 if (unlikely(!rc))
252                         rc = -ENODATA;
253         } else if (lbuf->lb_buf) {
254                 lbuf->lb_len = rc;
255         }
256         CDEBUG(D_INFO, "%s: "DFID" get xattr %s len %d\n",
257                tgt_name(tsi->tsi_tgt), PFID(lu_object_fid(&obj->do_lu)),
258                name, rc);
259
260         GOTO(out, rc);
261
262 out:
263         object_update_result_insert(reply, lbuf->lb_buf, lbuf->lb_len, idx, rc);
264         RETURN(0);
265 }
266
267 static int out_xattr_list(struct tgt_session_info *tsi)
268 {
269         const struct lu_env *env = tsi->tsi_env;
270         struct tgt_thread_info *tti = tgt_th_info(env);
271         struct lu_buf *lbuf = &tti->tti_buf;
272         struct object_update_reply *reply = tti->tti_u.update.tti_update_reply;
273         struct dt_object *obj = tti->tti_u.update.tti_dt_object;
274         struct object_update_result *update_result;
275         int idx = tti->tti_u.update.tti_update_reply_index;
276         int rc;
277
278         ENTRY;
279
280         if (!lu_object_exists(&obj->do_lu)) {
281                 set_bit(LU_OBJECT_HEARD_BANSHEE,
282                         &obj->do_lu.lo_header->loh_flags);
283                 RETURN(-ENOENT);
284         }
285
286         update_result = object_update_result_get(reply, 0, NULL);
287         if (!update_result) {
288                 rc = -EPROTO;
289                 CERROR("%s: empty buf for xattr list: rc = %d\n",
290                        tgt_name(tsi->tsi_tgt), rc);
291                 RETURN(rc);
292         }
293
294         lbuf->lb_len = (int)tti->tti_u.update.tti_update->ou_result_size;
295         lbuf->lb_buf = update_result->our_data;
296         if (lbuf->lb_len == 0)
297                 lbuf->lb_buf = 0;
298
299         dt_read_lock(env, obj, DT_TGT_CHILD);
300         rc = dt_xattr_list(env, obj, lbuf);
301         dt_read_unlock(env, obj);
302         if (rc <= 0) {
303                 lbuf->lb_len = 0;
304                 if (unlikely(!rc))
305                         rc = -ENODATA;
306         } else if (lbuf->lb_buf) {
307                 lbuf->lb_len = rc;
308         }
309
310         CDEBUG(D_INFO, "%s: "DFID" list xattr len %d\n",
311                tgt_name(tsi->tsi_tgt), PFID(lu_object_fid(&obj->do_lu)), rc);
312
313         /* Since we directly use update_result->our_data as the lbuf->lb_buf,
314          * then use NULL for result_insert to avoid unnecessary memory copy. */
315         object_update_result_insert(reply, NULL, lbuf->lb_len, idx, rc);
316
317         RETURN(0);
318 }
319
320 static int out_index_lookup(struct tgt_session_info *tsi)
321 {
322         const struct lu_env     *env = tsi->tsi_env;
323         struct tgt_thread_info  *tti = tgt_th_info(env);
324         struct object_update    *update = tti->tti_u.update.tti_update;
325         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
326         char                    *name;
327         int                      rc;
328
329         ENTRY;
330
331         if (unlikely(update->ou_result_size < sizeof(tti->tti_fid1)))
332                 return -EPROTO;
333
334         if (!lu_object_exists(&obj->do_lu))
335                 RETURN(-ENOENT);
336
337         name = object_update_param_get(update, 0, NULL);
338         if (IS_ERR(name)) {
339                 CERROR("%s: empty name for lookup: rc = %ld\n",
340                        tgt_name(tsi->tsi_tgt), PTR_ERR(name));
341                 RETURN(PTR_ERR(name));
342         }
343
344         dt_read_lock(env, obj, DT_TGT_CHILD);
345         if (!dt_try_as_dir(env, obj, true))
346                 GOTO(out_unlock, rc = -ENOTDIR);
347
348         rc = dt_lookup(env, obj, (struct dt_rec *)&tti->tti_fid1,
349                        (struct dt_key *)name);
350
351         if (rc < 0)
352                 GOTO(out_unlock, rc);
353
354         if (rc == 0)
355                 rc += 1;
356
357 out_unlock:
358         dt_read_unlock(env, obj);
359
360         CDEBUG(D_INFO, "lookup "DFID" %s get "DFID" rc %d\n",
361                PFID(lu_object_fid(&obj->do_lu)), name,
362                PFID(&tti->tti_fid1), rc);
363
364         CDEBUG(D_INFO, "%s: insert lookup reply %p index %d: rc = %d\n",
365                tgt_name(tsi->tsi_tgt), tti->tti_u.update.tti_update_reply,
366                0, rc);
367
368         object_update_result_insert(tti->tti_u.update.tti_update_reply,
369                             &tti->tti_fid1, sizeof(tti->tti_fid1),
370                             tti->tti_u.update.tti_update_reply_index, rc);
371         RETURN(rc);
372 }
373
374 static int out_xattr_set(struct tgt_session_info *tsi)
375 {
376         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
377         struct object_update    *update = tti->tti_u.update.tti_update;
378         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
379         struct lu_buf           *lbuf = &tti->tti_buf;
380         char                    *name;
381         char                    *buf;
382         __u32                   *tmp;
383         size_t                   buf_len = 0;
384         int                      flag;
385         size_t                   size = 0;
386         int                      rc;
387         ENTRY;
388
389         name = object_update_param_get(update, 0, NULL);
390         if (IS_ERR(name)) {
391                 CERROR("%s: empty name for xattr set: rc = %ld\n",
392                        tgt_name(tsi->tsi_tgt), PTR_ERR(name));
393                 RETURN(PTR_ERR(name));
394         }
395
396         /* If buffer == NULL (-ENODATA), then it might mean delete xattr */
397         buf = object_update_param_get(update, 1, &buf_len);
398         if (IS_ERR(buf) && PTR_ERR(buf) != -ENODATA)
399                 RETURN(PTR_ERR(buf));
400
401         lbuf->lb_buf = buf;
402         lbuf->lb_len = buf_len;
403
404         tmp = object_update_param_get(update, 2, &size);
405         if (IS_ERR(tmp) || size != sizeof(*tmp)) {
406                 CERROR("%s: emptry or wrong size %zu flag: rc = %ld\n",
407                        tgt_name(tsi->tsi_tgt), size, PTR_ERR(tmp));
408                 RETURN(PTR_ERR(tmp));
409         }
410
411         if (req_capsule_req_need_swab(tsi->tsi_pill))
412                 __swab32s(tmp);
413         flag = *tmp;
414
415         rc = out_tx_xattr_set(tsi->tsi_env, obj, lbuf, name, flag,
416                               &tti->tti_tea, tti->tti_tea.ta_handle,
417                               tti->tti_u.update.tti_update_reply,
418                               tti->tti_u.update.tti_update_reply_index);
419         RETURN(rc);
420 }
421
422 static int out_xattr_del(struct tgt_session_info *tsi)
423 {
424         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
425         struct object_update    *update = tti->tti_u.update.tti_update;
426         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
427         char                    *name;
428         int                      rc;
429         ENTRY;
430
431         name = object_update_param_get(update, 0, NULL);
432         if (IS_ERR(name)) {
433                 CERROR("%s: empty name for xattr set: rc = %ld\n",
434                        tgt_name(tsi->tsi_tgt), PTR_ERR(name));
435                 RETURN(PTR_ERR(name));
436         }
437
438         rc = out_tx_xattr_del(tsi->tsi_env, obj, name, &tti->tti_tea,
439                               tti->tti_tea.ta_handle,
440                               tti->tti_u.update.tti_update_reply,
441                               tti->tti_u.update.tti_update_reply_index);
442         RETURN(rc);
443 }
444
445 /**
446  * increase ref of the object
447  **/
448 static int out_ref_add(struct tgt_session_info *tsi)
449 {
450         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
451         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
452         int                      rc;
453
454         ENTRY;
455
456         rc = out_tx_ref_add(tsi->tsi_env, obj, &tti->tti_tea,
457                             tti->tti_tea.ta_handle,
458                             tti->tti_u.update.tti_update_reply,
459                             tti->tti_u.update.tti_update_reply_index);
460         RETURN(rc);
461 }
462
463 static int out_ref_del(struct tgt_session_info *tsi)
464 {
465         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
466         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
467         int                      rc;
468
469         ENTRY;
470
471         if (!lu_object_exists(&obj->do_lu))
472                 RETURN(-ENOENT);
473
474         rc = out_tx_ref_del(tsi->tsi_env, obj, &tti->tti_tea,
475                             tti->tti_tea.ta_handle,
476                             tti->tti_u.update.tti_update_reply,
477                             tti->tti_u.update.tti_update_reply_index);
478         RETURN(rc);
479 }
480
481 static int out_index_insert(struct tgt_session_info *tsi)
482 {
483         struct tgt_thread_info  *tti    = tgt_th_info(tsi->tsi_env);
484         struct object_update    *update = tti->tti_u.update.tti_update;
485         struct dt_object        *obj    = tti->tti_u.update.tti_dt_object;
486         struct dt_insert_rec    *rec    = &tti->tti_rec;
487         struct lu_fid           *fid;
488         char                    *name;
489         __u32                   *ptype;
490         int                      rc     = 0;
491         size_t                   size;
492         ENTRY;
493
494         name = object_update_param_get(update, 0, NULL);
495         if (IS_ERR(name)) {
496                 CERROR("%s: empty name for index insert: rc = %ld\n",
497                        tgt_name(tsi->tsi_tgt), PTR_ERR(name));
498                 RETURN(PTR_ERR(name));
499         }
500
501         fid = object_update_param_get(update, 1, &size);
502         if (IS_ERR(fid) || size != sizeof(*fid)) {
503                 CERROR("%s: invalid fid: rc = %ld\n",
504                        tgt_name(tsi->tsi_tgt), PTR_ERR(fid));
505                 RETURN(PTR_ERR(fid));
506         }
507
508         if (req_capsule_req_need_swab(tsi->tsi_pill))
509                 lustre_swab_lu_fid(fid);
510
511         if (!fid_is_sane(fid)) {
512                 CERROR("%s: invalid FID "DFID": rc = %d\n",
513                        tgt_name(tsi->tsi_tgt), PFID(fid), -EPROTO);
514                 RETURN(-EPROTO);
515         }
516
517         ptype = object_update_param_get(update, 2, &size);
518         if (IS_ERR(ptype) || size != sizeof(*ptype)) {
519                 CERROR("%s: invalid type for index insert: rc = %ld\n",
520                        tgt_name(tsi->tsi_tgt), PTR_ERR(ptype));
521                 RETURN(PTR_ERR(ptype));
522         }
523
524         if (req_capsule_req_need_swab(tsi->tsi_pill))
525                 __swab32s(ptype);
526
527         rec->rec_fid = fid;
528         rec->rec_type = *ptype;
529
530         rc = out_tx_index_insert(tsi->tsi_env, obj, (const struct dt_rec *)rec,
531                                  (const struct dt_key *)name, &tti->tti_tea,
532                                  tti->tti_tea.ta_handle,
533                                  tti->tti_u.update.tti_update_reply,
534                                  tti->tti_u.update.tti_update_reply_index);
535
536         CDEBUG(D_INFO, "%s: "DFID" index insert %s: rc = %d\n",
537                tgt_name(tsi->tsi_tgt), PFID(lu_object_fid(&obj->do_lu)),
538                name, rc);
539
540         RETURN(rc);
541 }
542
543 static int out_index_delete(struct tgt_session_info *tsi)
544 {
545         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
546         struct object_update    *update = tti->tti_u.update.tti_update;
547         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
548         char                    *name;
549         int                      rc = 0;
550
551         if (!lu_object_exists(&obj->do_lu))
552                 RETURN(-ENOENT);
553
554         name = object_update_param_get(update, 0, NULL);
555         if (IS_ERR(name)) {
556                 CERROR("%s: empty name for index delete: rc = %ld\n",
557                        tgt_name(tsi->tsi_tgt), PTR_ERR(name));
558                 RETURN(PTR_ERR(name));
559         }
560
561         rc = out_tx_index_delete(tsi->tsi_env, obj, (const struct dt_key *)name,
562                                  &tti->tti_tea, tti->tti_tea.ta_handle,
563                                  tti->tti_u.update.tti_update_reply,
564                                  tti->tti_u.update.tti_update_reply_index);
565         RETURN(rc);
566 }
567
568 static int out_destroy(struct tgt_session_info *tsi)
569 {
570         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
571         struct object_update    *update = tti->tti_u.update.tti_update;
572         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
573         struct lu_fid           *fid;
574         int                      rc;
575         ENTRY;
576
577         fid = &update->ou_fid;
578         if (!fid_is_sane(fid)) {
579                 CERROR("%s: invalid FID "DFID": rc = %d\n",
580                        tgt_name(tsi->tsi_tgt), PFID(fid), -EPROTO);
581                 RETURN(-EPROTO);
582         }
583
584         if (!lu_object_exists(&obj->do_lu))
585                 RETURN(-ENOENT);
586
587         rc = out_tx_destroy(tsi->tsi_env, obj, &tti->tti_tea,
588                             tti->tti_tea.ta_handle,
589                             tti->tti_u.update.tti_update_reply,
590                             tti->tti_u.update.tti_update_reply_index);
591
592         RETURN(rc);
593 }
594
595 static int out_write(struct tgt_session_info *tsi)
596 {
597         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
598         struct object_update    *update = tti->tti_u.update.tti_update;
599         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
600         struct lu_buf           *lbuf = &tti->tti_buf;
601         char                    *buf;
602         __u64                   *tmp;
603         size_t                  size = 0;
604         size_t                  buf_len = 0;
605         loff_t                  pos;
606         int                      rc;
607         ENTRY;
608
609         buf = object_update_param_get(update, 0, &buf_len);
610         if (IS_ERR(buf) || buf_len == 0) {
611                 CERROR("%s: empty buf for xattr set: rc = %ld\n",
612                        tgt_name(tsi->tsi_tgt), PTR_ERR(buf));
613                 RETURN(PTR_ERR(buf));
614         }
615         lbuf->lb_buf = buf;
616         lbuf->lb_len = buf_len;
617
618         tmp = object_update_param_get(update, 1, &size);
619         if (IS_ERR(tmp) || size != sizeof(*tmp)) {
620                 CERROR("%s: empty or wrong size %zu pos: rc = %ld\n",
621                        tgt_name(tsi->tsi_tgt), size, PTR_ERR(tmp));
622                 RETURN(PTR_ERR(tmp));
623         }
624
625         if (req_capsule_req_need_swab(tsi->tsi_pill))
626                 __swab64s(tmp);
627         pos = *tmp;
628
629         rc = out_tx_write(tsi->tsi_env, obj, lbuf, pos,
630                           &tti->tti_tea, tti->tti_tea.ta_handle,
631                           tti->tti_u.update.tti_update_reply,
632                           tti->tti_u.update.tti_update_reply_index);
633         RETURN(rc);
634 }
635
636 static int out_read(struct tgt_session_info *tsi)
637 {
638         const struct lu_env     *env = tsi->tsi_env;
639         struct tgt_thread_info  *tti = tgt_th_info(env);
640         struct object_update    *update = tti->tti_u.update.tti_update;
641         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
642         struct object_update_reply *reply = tti->tti_u.update.tti_update_reply;
643         int index = tti->tti_u.update.tti_update_reply_index;
644         struct lu_rdbuf *rdbuf;
645         struct object_update_result *update_result;
646         struct out_read_reply   *orr;
647         void *tmp;
648         size_t size;
649         size_t total_size = 0;
650         __u64 pos;
651         unsigned int i;
652         unsigned int nbufs;
653         int rc = 0;
654         ENTRY;
655
656         update_result = object_update_result_get(reply, index, NULL);
657         LASSERT(update_result != NULL);
658         update_result->our_datalen = sizeof(*orr);
659
660         if (!lu_object_exists(&obj->do_lu))
661                 GOTO(out, rc = -ENOENT);
662
663         tmp = object_update_param_get(update, 0, NULL);
664         if (IS_ERR(tmp)) {
665                 CERROR("%s: empty size for read: rc = %ld\n",
666                        tgt_name(tsi->tsi_tgt), PTR_ERR(tmp));
667                 GOTO(out, rc = PTR_ERR(tmp));
668         }
669         size = le64_to_cpu(*(size_t *)(tmp));
670
671         tmp = object_update_param_get(update, 1, NULL);
672         if (IS_ERR(tmp)) {
673                 CERROR("%s: empty pos for read: rc = %ld\n",
674                        tgt_name(tsi->tsi_tgt), PTR_ERR(tmp));
675                 GOTO(out, rc = PTR_ERR(tmp));
676         }
677         pos = le64_to_cpu(*(__u64 *)(tmp));
678
679         /* Put the offset into the begining of the buffer in reply */
680         orr = (struct out_read_reply *)update_result->our_data;
681
682         nbufs = (size + OUT_BULK_BUFFER_SIZE - 1) / OUT_BULK_BUFFER_SIZE;
683         OBD_ALLOC(rdbuf, sizeof(*rdbuf) + nbufs * sizeof(rdbuf->rb_bufs[0]));
684         if (rdbuf == NULL)
685                 GOTO(out, rc = -ENOMEM);
686
687         rdbuf->rb_nbufs = 0;
688         total_size = 0;
689         for (i = 0; i < nbufs; i++) {
690                 __u32 read_size;
691
692                 read_size = size > OUT_BULK_BUFFER_SIZE ?
693                             OUT_BULK_BUFFER_SIZE : size;
694                 OBD_ALLOC(rdbuf->rb_bufs[i].lb_buf, read_size);
695                 if (rdbuf->rb_bufs[i].lb_buf == NULL)
696                         GOTO(out_free, rc = -ENOMEM);
697
698                 rdbuf->rb_bufs[i].lb_len = read_size;
699                 dt_read_lock(env, obj, DT_TGT_CHILD);
700                 rc = dt_read(env, obj, &rdbuf->rb_bufs[i], &pos);
701                 dt_read_unlock(env, obj);
702
703                 total_size += rc < 0 ? 0 : rc;
704                 if (rc <= 0)
705                         break;
706
707                 rdbuf->rb_nbufs++;
708                 size -= read_size;
709         }
710
711         /* send pages to client */
712         rc = tgt_send_buffer(tsi, rdbuf);
713         if (rc < 0)
714                 GOTO(out_free, rc);
715
716         orr->orr_size = total_size;
717         orr->orr_offset = pos;
718
719         orr_cpu_to_le(orr, orr);
720         update_result->our_datalen += orr->orr_size;
721 out_free:
722         for (i = 0; i < nbufs; i++) {
723                 if (rdbuf->rb_bufs[i].lb_buf != NULL) {
724                         OBD_FREE(rdbuf->rb_bufs[i].lb_buf,
725                                  rdbuf->rb_bufs[i].lb_len);
726                 }
727         }
728         OBD_FREE(rdbuf, sizeof(*rdbuf) +
729                         nbufs * sizeof(rdbuf->rb_bufs[0]));
730 out:
731         /* Insert read buffer */
732         update_result->our_rc = ptlrpc_status_hton(rc);
733         reply->ourp_lens[index] = cfs_size_round(update_result->our_datalen +
734                                                  sizeof(*update_result));
735         RETURN(rc);
736 }
737
738 static int out_noop(struct tgt_session_info *tsi)
739 {
740         return 0;
741 }
742
743 #define DEF_OUT_HNDL(opc, name, flags, fn)     \
744 [opc - OUT_CREATE] = {                                  \
745         .th_name    = name,                             \
746         .th_fail_id = 0,                                \
747         .th_opc     = opc,                              \
748         .th_flags   = flags,                            \
749         .th_act     = fn,                               \
750         .th_fmt     = NULL,                             \
751         .th_version = 0,                                \
752 }
753
754 static struct tgt_handler out_update_ops[] = {
755         DEF_OUT_HNDL(OUT_CREATE, "out_create", IS_MUTABLE | HAS_REPLY,
756                      out_create),
757         DEF_OUT_HNDL(OUT_DESTROY, "out_create", IS_MUTABLE | HAS_REPLY,
758                      out_destroy),
759         DEF_OUT_HNDL(OUT_REF_ADD, "out_ref_add", IS_MUTABLE | HAS_REPLY,
760                      out_ref_add),
761         DEF_OUT_HNDL(OUT_REF_DEL, "out_ref_del", IS_MUTABLE | HAS_REPLY,
762                      out_ref_del),
763         DEF_OUT_HNDL(OUT_ATTR_SET, "out_attr_set",  IS_MUTABLE | HAS_REPLY,
764                      out_attr_set),
765         DEF_OUT_HNDL(OUT_ATTR_GET, "out_attr_get",  HAS_REPLY,
766                      out_attr_get),
767         DEF_OUT_HNDL(OUT_XATTR_SET, "out_xattr_set", IS_MUTABLE | HAS_REPLY,
768                      out_xattr_set),
769         DEF_OUT_HNDL(OUT_XATTR_DEL, "out_xattr_del", IS_MUTABLE | HAS_REPLY,
770                      out_xattr_del),
771         DEF_OUT_HNDL(OUT_XATTR_GET, "out_xattr_get", HAS_REPLY,
772                      out_xattr_get),
773         DEF_OUT_HNDL(OUT_INDEX_LOOKUP, "out_index_lookup", HAS_REPLY,
774                      out_index_lookup),
775         DEF_OUT_HNDL(OUT_INDEX_INSERT, "out_index_insert",
776                      IS_MUTABLE | HAS_REPLY, out_index_insert),
777         DEF_OUT_HNDL(OUT_INDEX_DELETE, "out_index_delete",
778                      IS_MUTABLE | HAS_REPLY, out_index_delete),
779         DEF_OUT_HNDL(OUT_WRITE, "out_write", IS_MUTABLE | HAS_REPLY, out_write),
780         DEF_OUT_HNDL(OUT_READ, "out_read", HAS_REPLY, out_read),
781         DEF_OUT_HNDL(OUT_NOOP, "out_noop", HAS_REPLY, out_noop),
782         DEF_OUT_HNDL(OUT_XATTR_LIST, "out_xattr_list", HAS_REPLY,
783                      out_xattr_list),
784 };
785
786 static struct tgt_handler *out_handler_find(__u32 opc)
787 {
788         struct tgt_handler *h;
789
790         h = NULL;
791         if (OUT_CREATE <= opc && opc < OUT_LAST) {
792                 h = &out_update_ops[opc - OUT_CREATE];
793                 LASSERTF(h->th_opc == opc, "opcode mismatch %d != %d\n",
794                          h->th_opc, opc);
795         } else {
796                 h = NULL; /* unsupported opc */
797         }
798         return h;
799 }
800
801 static int out_tx_start(const struct lu_env *env, struct dt_device *dt,
802                         struct thandle_exec_args *ta, struct obd_export *exp)
803 {
804         ta->ta_argno = 0;
805         ta->ta_handle = dt_trans_create(env, dt);
806         if (IS_ERR(ta->ta_handle)) {
807                 int rc;
808
809                 rc = PTR_ERR(ta->ta_handle);
810                 ta->ta_handle = NULL;
811                 CERROR("%s: start handle error: rc = %d\n", dt_obd_name(dt),
812                        rc);
813                 return rc;
814         }
815         if (exp->exp_need_sync)
816                 ta->ta_handle->th_sync = 1;
817
818         return 0;
819 }
820
821 static int out_trans_start(const struct lu_env *env,
822                            struct thandle_exec_args *ta)
823 {
824         return dt_trans_start(env, ta->ta_handle->th_dev, ta->ta_handle);
825 }
826
827 static int out_trans_stop(const struct lu_env *env,
828                           struct thandle_exec_args *ta, int err)
829 {
830         int i;
831         int rc;
832
833         ta->ta_handle->th_result = err;
834         rc = dt_trans_stop(env, ta->ta_handle->th_dev, ta->ta_handle);
835         for (i = 0; i < ta->ta_argno; i++) {
836                 if (ta->ta_args[i]->object != NULL) {
837                         dt_object_put(env, ta->ta_args[i]->object);
838                         ta->ta_args[i]->object = NULL;
839                 }
840         }
841         ta->ta_handle = NULL;
842         ta->ta_argno = 0;
843
844         return rc;
845 }
846
847 static int out_tx_end(const struct lu_env *env, struct thandle_exec_args *ta,
848                       int declare_ret)
849 {
850         struct tgt_session_info *tsi = tgt_ses_info(env);
851         int                     i;
852         int                     rc;
853         int                     rc1;
854         ENTRY;
855
856         if (ta->ta_handle == NULL)
857                 RETURN(0);
858
859         if (declare_ret != 0 || ta->ta_argno == 0)
860                 GOTO(stop, rc = declare_ret);
861
862         LASSERT(ta->ta_handle->th_dev != NULL);
863         rc = out_trans_start(env, ta);
864         if (unlikely(rc != 0))
865                 GOTO(stop, rc);
866
867         for (i = 0; i < ta->ta_argno; i++) {
868                 rc = ta->ta_args[i]->exec_fn(env, ta->ta_handle,
869                                              ta->ta_args[i]);
870                 if (unlikely(rc != 0)) {
871                         CDEBUG(D_INFO, "error during execution of #%u from"
872                                " %s:%d: rc = %d\n", i, ta->ta_args[i]->file,
873                                ta->ta_args[i]->line, rc);
874                         while (--i >= 0) {
875                                 if (ta->ta_args[i]->undo_fn != NULL)
876                                         ta->ta_args[i]->undo_fn(env,
877                                                                ta->ta_handle,
878                                                                ta->ta_args[i]);
879                                 else
880                                         CERROR("%s: undo for %s:%d: rc = %d\n",
881                                              dt_obd_name(ta->ta_handle->th_dev),
882                                                ta->ta_args[i]->file,
883                                                ta->ta_args[i]->line, -ENOTSUPP);
884                         }
885                         break;
886                 }
887                 CDEBUG(D_INFO, "%s: executed %u/%u: rc = %d\n",
888                        dt_obd_name(ta->ta_handle->th_dev), i, ta->ta_argno, rc);
889         }
890
891         /* Only fail for real updates, XXX right now llog updates will be
892         * ignore, whose updates count is usually 1, so failover test
893         * case will spot this FAIL_UPDATE_NET_REP precisely, and it will
894         * be removed after async update patch is landed. */
895         if (ta->ta_argno > 1)
896                 tsi->tsi_reply_fail_id = OBD_FAIL_OUT_UPDATE_NET_REP;
897
898 stop:
899         rc1 = out_trans_stop(env, ta, rc);
900         if (rc == 0)
901                 rc = rc1;
902
903         ta->ta_handle = NULL;
904         ta->ta_argno = 0;
905
906         RETURN(rc);
907 }
908
909 /**
910  * Object updates between Targets. Because all the updates has been
911  * dis-assemblied into object updates at sender side, so OUT will
912  * call OSD API directly to execute these updates.
913  *
914  * In DNE phase I all of the updates in the request need to be executed
915  * in one transaction, and the transaction has to be synchronously.
916  *
917  * Please refer to lustre/include/lustre/lustre_idl.h for req/reply
918  * format.
919  */
920 int out_handle(struct tgt_session_info *tsi)
921 {
922         const struct lu_env             *env = tsi->tsi_env;
923         struct tgt_thread_info          *tti = tgt_th_info(env);
924         struct thandle_exec_args        *ta = &tti->tti_tea;
925         struct req_capsule              *pill = tsi->tsi_pill;
926         struct dt_device                *dt = tsi->tsi_tgt->lut_bottom;
927         struct out_update_header        *ouh;
928         struct out_update_buffer        *oub = NULL;
929         struct object_update            *update;
930         struct object_update_reply      *reply;
931         struct ptlrpc_bulk_desc         *desc = NULL;
932         struct tg_reply_data *trd = NULL;
933         void                            **update_bufs;
934         int                             current_batchid = -1;
935         __u32                           update_buf_count;
936         unsigned int                    i;
937         unsigned int                    reply_index = 0;
938         int                             rc = 0;
939         int                             rc1 = 0;
940         int                             ouh_size, reply_size;
941         int                             updates;
942         bool need_reconstruct;
943
944         ENTRY;
945
946         req_capsule_set(pill, &RQF_OUT_UPDATE);
947         ouh_size = req_capsule_get_size(pill, &RMF_OUT_UPDATE_HEADER,
948                                         RCL_CLIENT);
949         if (ouh_size <= 0)
950                 RETURN(err_serious(-EPROTO));
951
952         ouh = req_capsule_client_get(pill, &RMF_OUT_UPDATE_HEADER);
953         if (ouh == NULL)
954                 RETURN(err_serious(-EPROTO));
955
956         if (ouh->ouh_magic != OUT_UPDATE_HEADER_MAGIC) {
957                 CERROR("%s: invalid update buffer magic %x expect %x: "
958                        "rc = %d\n", tgt_name(tsi->tsi_tgt), ouh->ouh_magic,
959                        UPDATE_REQUEST_MAGIC, -EPROTO);
960                 RETURN(err_serious(-EPROTO));
961         }
962
963         update_buf_count = ouh->ouh_count;
964         if (update_buf_count == 0)
965                 RETURN(err_serious(-EPROTO));
966
967         OBD_ALLOC_PTR_ARRAY(update_bufs, update_buf_count);
968         if (update_bufs == NULL)
969                 RETURN(err_serious(-ENOMEM));
970
971         if (ouh->ouh_inline_length > 0) {
972                 update_bufs[0] = ouh->ouh_inline_data;
973         } else {
974                 struct out_update_buffer *tmp;
975                 int page_count = 0;
976
977                 oub = req_capsule_client_get(pill, &RMF_OUT_UPDATE_BUF);
978                 if (oub == NULL)
979                         GOTO(out_free, rc = err_serious(-EPROTO));
980
981                 for (i = 0; i < update_buf_count; i++)
982                         /* First *and* last might be partial pages, hence +1 */
983                         page_count += DIV_ROUND_UP(oub[i].oub_size,
984                                                    PAGE_SIZE) + 1;
985
986                 desc = ptlrpc_prep_bulk_exp(pill->rc_req, page_count,
987                                             PTLRPC_BULK_OPS_COUNT,
988                                             PTLRPC_BULK_GET_SINK,
989                                             MDS_BULK_PORTAL,
990                                             &ptlrpc_bulk_kiov_nopin_ops);
991                 if (desc == NULL)
992                         GOTO(out_free, rc = err_serious(-ENOMEM));
993
994                 tmp = oub;
995                 for (i = 0; i < update_buf_count; i++, tmp++) {
996                         if (tmp->oub_size >= OUT_MAXREQSIZE)
997                                 GOTO(out_free, rc = err_serious(-EPROTO));
998
999                         OBD_ALLOC_LARGE(update_bufs[i], tmp->oub_size);
1000                         if (update_bufs[i] == NULL)
1001                                 GOTO(out_free, rc = err_serious(-ENOMEM));
1002
1003                         desc->bd_frag_ops->add_iov_frag(desc, update_bufs[i],
1004                                                         tmp->oub_size);
1005                 }
1006
1007                 pill->rc_req->rq_bulk_write = 1;
1008                 rc = sptlrpc_svc_prep_bulk(pill->rc_req, desc);
1009                 if (rc != 0)
1010                         GOTO(out_free, rc = err_serious(rc));
1011
1012                 rc = target_bulk_io(pill->rc_req->rq_export, desc);
1013                 if (rc < 0)
1014                         GOTO(out_free, rc = err_serious(rc));
1015         }
1016         /* validate the request and calculate the total update count and
1017          * set it to reply */
1018         reply_size = 0;
1019         updates = 0;
1020         for (i = 0; i < update_buf_count; i++) {
1021                 struct object_update_request    *our;
1022                 int                              j;
1023
1024                 our = update_bufs[i];
1025                 if (req_capsule_req_need_swab(pill))
1026                         lustre_swab_object_update_request(our, 0);
1027
1028                 if (our->ourq_magic != UPDATE_REQUEST_MAGIC) {
1029                         CERROR("%s: invalid update buffer magic %x"
1030                                " expect %x: rc = %d\n",
1031                                tgt_name(tsi->tsi_tgt), our->ourq_magic,
1032                                UPDATE_REQUEST_MAGIC, -EPROTO);
1033                         GOTO(out_free, rc = err_serious(-EPROTO));
1034                 }
1035                 updates += our->ourq_count;
1036
1037                 /* need to calculate reply size */
1038                 for (j = 0; j < our->ourq_count; j++) {
1039                         update = object_update_request_get(our, j, NULL);
1040                         if (update == NULL)
1041                                 GOTO(out, rc = err_serious(-EPROTO));
1042                         if (req_capsule_req_need_swab(pill))
1043                                 lustre_swab_object_update(update);
1044
1045                         if (!fid_is_sane(&update->ou_fid)) {
1046                                 CERROR("%s: invalid FID "DFID": rc = %d\n",
1047                                        tgt_name(tsi->tsi_tgt),
1048                                        PFID(&update->ou_fid), -EPROTO);
1049                                 GOTO(out, rc = err_serious(-EPROTO));
1050                         }
1051
1052                         /* XXX: what ou_result_size can be considered safe? */
1053
1054                         reply_size += sizeof(reply->ourp_lens[0]);
1055                         reply_size += sizeof(struct object_update_result);
1056                         reply_size += update->ou_result_size;
1057                 }
1058         }
1059         reply_size += sizeof(*reply);
1060
1061         if (unlikely(reply_size > ouh->ouh_reply_size)) {
1062                 CERROR("%s: too small reply buf %u for %u, need %u at least\n",
1063                        tgt_name(tsi->tsi_tgt), ouh->ouh_reply_size,
1064                        updates, reply_size);
1065                 GOTO(out_free, rc = err_serious(-EPROTO));
1066         }
1067
1068         req_capsule_set_size(pill, &RMF_OUT_UPDATE_REPLY, RCL_SERVER,
1069                              ouh->ouh_reply_size);
1070         rc = req_capsule_server_pack(pill);
1071         if (rc != 0) {
1072                 CERROR("%s: Can't pack response: rc = %d\n",
1073                        tgt_name(tsi->tsi_tgt), rc);
1074                 GOTO(out_free, rc = err_serious(-EPROTO));
1075         }
1076
1077         /* Prepare the update reply buffer */
1078         reply = req_capsule_server_get(pill, &RMF_OUT_UPDATE_REPLY);
1079         if (reply == NULL)
1080                 GOTO(out_free, rc = -EPROTO);
1081         reply->ourp_magic = UPDATE_REPLY_MAGIC;
1082         reply->ourp_count = updates;
1083         tti->tti_u.update.tti_update_reply = reply;
1084         tsi->tsi_mult_trans = !req_is_replay(tgt_ses_req(tsi));
1085
1086         OBD_ALLOC_PTR(trd);
1087         if (!trd)
1088                 GOTO(out_free, rc = -ENOMEM);
1089
1090         need_reconstruct = tgt_check_resent(pill->rc_req, trd);
1091
1092         /* Walk through updates in the request to execute them */
1093         for (i = 0; i < update_buf_count; i++) {
1094                 struct tgt_handler      *h;
1095                 struct dt_object        *dt_obj;
1096                 int                     update_count;
1097                 struct object_update_request *our;
1098                 int                     j;
1099
1100                 our = update_bufs[i];
1101                 update_count = our->ourq_count;
1102                 for (j = 0; j < update_count; j++) {
1103                         struct lu_object_conf conf;
1104
1105                         update = object_update_request_get(our, j, NULL);
1106                         if (update->ou_type == OUT_CREATE)
1107                                 conf.loc_flags = LOC_F_NEW;
1108                         else
1109                                 conf.loc_flags = 0;
1110
1111                         dt_obj = dt_locate_at(env, dt, &update->ou_fid,
1112                                 dt->dd_lu_dev.ld_site->ls_top_dev, &conf);
1113                         if (IS_ERR(dt_obj))
1114                                 GOTO(out, rc = PTR_ERR(dt_obj));
1115
1116                         if (dt->dd_record_fid_accessed) {
1117                                 struct lfsck_req_local *lrl = &tti->tti_lrl;
1118
1119                                 lfsck_pack_rfa(lrl,
1120                                                lu_object_fid(&dt_obj->do_lu),
1121                                                LEL_FID_ACCESSED,
1122                                                LFSCK_TYPE_LAYOUT);
1123                                 tgt_lfsck_in_notify_local(env, dt, lrl, NULL);
1124                         }
1125
1126                         tti->tti_u.update.tti_dt_object = dt_obj;
1127                         tti->tti_u.update.tti_update = update;
1128                         tti->tti_u.update.tti_update_reply_index = reply_index;
1129
1130                         h = out_handler_find(update->ou_type);
1131                         if (unlikely(h == NULL)) {
1132                                 CERROR("%s: unsupported opc: 0x%x\n",
1133                                        tgt_name(tsi->tsi_tgt), update->ou_type);
1134                                 GOTO(next, rc = -ENOTSUPP);
1135                         }
1136
1137                         /* Check resend case only for modifying RPC */
1138                         if (h->th_flags & IS_MUTABLE) {
1139                                 /* sanity check for last XID changing */
1140                                 if (unlikely(!need_reconstruct &&
1141                                              req_xid_is_last(pill->rc_req))) {
1142                                         DEBUG_REQ(D_ERROR, pill->rc_req,
1143                                                   "unexpected last XID change");
1144                                         GOTO(next, rc = -EINVAL);
1145                                 }
1146
1147                                 if (need_reconstruct) {
1148                                         out_reconstruct(env, dt, dt_obj, reply,
1149                                                         reply_index);
1150                                         GOTO(next, rc = 0);
1151                                 }
1152
1153                                 if (dt->dd_rdonly)
1154                                         GOTO(next, rc = -EROFS);
1155                         }
1156
1157                         /* start transaction for modification RPC only */
1158                         if (h->th_flags & IS_MUTABLE && current_batchid == -1) {
1159                                 current_batchid = update->ou_batchid;
1160
1161                                 if (reply_index == 0)
1162                                         CFS_RACE(OBD_FAIL_PTLRPC_RESEND_RACE);
1163
1164                                 rc = out_tx_start(env, dt, ta, tsi->tsi_exp);
1165                                 if (rc != 0)
1166                                         GOTO(next, rc);
1167
1168                                 if (update->ou_flags & UPDATE_FL_SYNC)
1169                                         ta->ta_handle->th_sync = 1;
1170                         }
1171
1172                         /* Stop the current update transaction, if the update
1173                          * has different batchid, or read-only update */
1174                         if (((current_batchid != update->ou_batchid) ||
1175                              !(h->th_flags & IS_MUTABLE)) &&
1176                              ta->ta_handle != NULL) {
1177                                 rc = out_tx_end(env, ta, rc);
1178                                 current_batchid = -1;
1179                                 if (rc != 0)
1180                                         GOTO(next, rc);
1181
1182                                 /* start a new transaction if needed */
1183                                 if (h->th_flags & IS_MUTABLE) {
1184                                         rc = out_tx_start(env, dt, ta,
1185                                                           tsi->tsi_exp);
1186                                         if (rc != 0)
1187                                                 GOTO(next, rc);
1188                                         if (update->ou_flags & UPDATE_FL_SYNC)
1189                                                 ta->ta_handle->th_sync = 1;
1190                                         current_batchid = update->ou_batchid;
1191                                 }
1192                         }
1193
1194                         if (OBD_FAIL_CHECK(OBD_FAIL_OUT_EIO))
1195                                 rc = -EIO;
1196                         else
1197                                 rc = h->th_act(tsi);
1198 next:
1199                         reply_index++;
1200                         dt_object_put(env, dt_obj);
1201                         if (rc < 0)
1202                                 GOTO(out, rc);
1203                 }
1204         }
1205 out:
1206         if (current_batchid != -1) {
1207                 rc1 = out_tx_end(env, ta, rc);
1208                 if (rc == 0)
1209                         rc = rc1;
1210         }
1211
1212 out_free:
1213         if (update_bufs != NULL) {
1214                 if (oub != NULL) {
1215                         for (i = 0; i < update_buf_count; i++, oub++) {
1216                                 if (update_bufs[i] != NULL)
1217                                         OBD_FREE_LARGE(update_bufs[i],
1218                                                        oub->oub_size);
1219                         }
1220                 }
1221
1222                 OBD_FREE_PTR_ARRAY(update_bufs, update_buf_count);
1223         }
1224
1225         if (trd)
1226                 OBD_FREE_PTR(trd);
1227
1228         if (desc != NULL)
1229                 ptlrpc_free_bulk(desc);
1230
1231         RETURN(rc);
1232 }
1233
1234 struct tgt_handler tgt_out_handlers[] = {
1235 TGT_UPDATE_HDL(IS_MUTABLE,      OUT_UPDATE,     out_handle),
1236 };
1237 EXPORT_SYMBOL(tgt_out_handlers);
1238