Whamcloud - gitweb
LU-4343 tests: mkdir failing in sanity-hsm test 228
[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, Intel Corporation.
24  *
25  * lustre/mdt/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 <obd_class.h>
35 #include <md_object.h>
36 #include "tgt_internal.h"
37 #include <lustre_update.h>
38
39 struct tx_arg *tx_add_exec(struct thandle_exec_args *ta, tx_exec_func_t func,
40                            tx_exec_func_t undo, char *file, int line)
41 {
42         int i;
43
44         LASSERT(ta);
45         LASSERT(func);
46
47         i = ta->ta_argno;
48         LASSERT(i < UPDATE_MAX_OPS);
49
50         ta->ta_argno++;
51
52         ta->ta_args[i].exec_fn = func;
53         ta->ta_args[i].undo_fn = undo;
54         ta->ta_args[i].file    = file;
55         ta->ta_args[i].line    = line;
56
57         return &ta->ta_args[i];
58 }
59
60 static int out_tx_start(const struct lu_env *env, struct dt_device *dt,
61                         struct thandle_exec_args *ta)
62 {
63         memset(ta, 0, sizeof(*ta));
64         ta->ta_handle = dt_trans_create(env, dt);
65         if (IS_ERR(ta->ta_handle)) {
66                 CERROR("%s: start handle error: rc = %ld\n",
67                        dt_obd_name(dt), PTR_ERR(ta->ta_handle));
68                 return PTR_ERR(ta->ta_handle);
69         }
70         ta->ta_dev = dt;
71         /*For phase I, sync for cross-ref operation*/
72         ta->ta_handle->th_sync = 1;
73         return 0;
74 }
75
76 static int out_trans_start(const struct lu_env *env,
77                            struct thandle_exec_args *ta)
78 {
79         /* Always do sync commit for Phase I */
80         LASSERT(ta->ta_handle->th_sync != 0);
81         return dt_trans_start(env, ta->ta_dev, ta->ta_handle);
82 }
83
84 static int out_trans_stop(const struct lu_env *env,
85                           struct thandle_exec_args *ta, int err)
86 {
87         int i;
88         int rc;
89
90         ta->ta_handle->th_result = err;
91         LASSERT(ta->ta_handle->th_sync != 0);
92         rc = dt_trans_stop(env, ta->ta_dev, ta->ta_handle);
93         for (i = 0; i < ta->ta_argno; i++) {
94                 if (ta->ta_args[i].object != NULL) {
95                         lu_object_put(env, &ta->ta_args[i].object->do_lu);
96                         ta->ta_args[i].object = NULL;
97                 }
98         }
99
100         return rc;
101 }
102
103 int out_tx_end(const struct lu_env *env, struct thandle_exec_args *ta)
104 {
105         struct tgt_session_info *tsi = tgt_ses_info(env);
106         int i = 0, rc;
107
108         LASSERT(ta->ta_dev);
109         LASSERT(ta->ta_handle);
110
111         if (ta->ta_err != 0 || ta->ta_argno == 0)
112                 GOTO(stop, rc = ta->ta_err);
113
114         rc = out_trans_start(env, ta);
115         if (unlikely(rc))
116                 GOTO(stop, rc);
117
118         for (i = 0; i < ta->ta_argno; i++) {
119                 rc = ta->ta_args[i].exec_fn(env, ta->ta_handle,
120                                             &ta->ta_args[i]);
121                 if (unlikely(rc)) {
122                         CDEBUG(D_INFO, "error during execution of #%u from"
123                                " %s:%d: rc = %d\n", i, ta->ta_args[i].file,
124                                ta->ta_args[i].line, rc);
125                         while (--i >= 0) {
126                                 LASSERTF(ta->ta_args[i].undo_fn != NULL,
127                                     "can't undo changes, hope for failover!\n");
128                                 ta->ta_args[i].undo_fn(env, ta->ta_handle,
129                                                        &ta->ta_args[i]);
130                         }
131                         break;
132                 }
133         }
134
135         /* Only fail for real update */
136         tsi->tsi_reply_fail_id = OBD_FAIL_UPDATE_OBJ_NET_REP;
137 stop:
138         CDEBUG(D_INFO, "%s: executed %u/%u: rc = %d\n",
139                dt_obd_name(ta->ta_dev), i, ta->ta_argno, rc);
140         out_trans_stop(env, ta, rc);
141         ta->ta_handle = NULL;
142         ta->ta_argno = 0;
143         ta->ta_err = 0;
144
145         RETURN(rc);
146 }
147
148 static void out_reconstruct(const struct lu_env *env, struct dt_device *dt,
149                             struct dt_object *obj, struct update_reply *reply,
150                             int index)
151 {
152         CDEBUG(D_INFO, "%s: fork reply reply %p index %d: rc = %d\n",
153                dt_obd_name(dt), reply, index, 0);
154
155         update_insert_reply(reply, NULL, 0, index, 0);
156         return;
157 }
158
159 typedef void (*out_reconstruct_t)(const struct lu_env *env,
160                                   struct dt_device *dt,
161                                   struct dt_object *obj,
162                                   struct update_reply *reply,
163                                   int index);
164
165 static inline int out_check_resent(const struct lu_env *env,
166                                    struct dt_device *dt,
167                                    struct dt_object *obj,
168                                    struct ptlrpc_request *req,
169                                    out_reconstruct_t reconstruct,
170                                    struct update_reply *reply,
171                                    int index)
172 {
173         if (likely(!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT)))
174                 return 0;
175
176         if (req_xid_is_last(req)) {
177                 reconstruct(env, dt, obj, reply, index);
178                 return 1;
179         }
180         DEBUG_REQ(D_HA, req, "no reply for RESENT req (have "LPD64")",
181                  req->rq_export->exp_target_data.ted_lcd->lcd_last_xid);
182         return 0;
183 }
184
185 static int out_obj_destroy(const struct lu_env *env, struct dt_object *dt_obj,
186                            struct thandle *th)
187 {
188         int rc;
189
190         CDEBUG(D_INFO, "%s: destroy "DFID"\n", dt_obd_name(th->th_dev),
191                PFID(lu_object_fid(&dt_obj->do_lu)));
192
193         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
194         rc = dt_destroy(env, dt_obj, th);
195         dt_write_unlock(env, dt_obj);
196
197         return rc;
198 }
199
200 /**
201  * All of the xxx_undo will be used once execution failed,
202  * But because all of the required resource has been reserved in
203  * declare phase, i.e. if declare succeed, it should make sure
204  * the following executing phase succeed in anyway, so these undo
205  * should be useless for most of the time in Phase I
206  */
207 int out_tx_create_undo(const struct lu_env *env, struct thandle *th,
208                        struct tx_arg *arg)
209 {
210         int rc;
211
212         rc = out_obj_destroy(env, arg->object, th);
213         if (rc != 0)
214                 CERROR("%s: undo failure, we are doomed!: rc = %d\n",
215                        dt_obd_name(th->th_dev), rc);
216         return rc;
217 }
218
219 int out_tx_create_exec(const struct lu_env *env, struct thandle *th,
220                        struct tx_arg *arg)
221 {
222         struct dt_object        *dt_obj = arg->object;
223         int                      rc;
224
225         CDEBUG(D_OTHER, "%s: create "DFID": dof %u, mode %o\n",
226                dt_obd_name(th->th_dev),
227                PFID(lu_object_fid(&arg->object->do_lu)),
228                arg->u.create.dof.dof_type,
229                arg->u.create.attr.la_mode & S_IFMT);
230
231         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
232         rc = dt_create(env, dt_obj, &arg->u.create.attr,
233                        &arg->u.create.hint, &arg->u.create.dof, th);
234
235         dt_write_unlock(env, dt_obj);
236
237         CDEBUG(D_INFO, "%s: insert create reply %p index %d: rc = %d\n",
238                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
239
240         update_insert_reply(arg->reply, NULL, 0, arg->index, rc);
241
242         return rc;
243 }
244
245 static int __out_tx_create(const struct lu_env *env, struct dt_object *obj,
246                            struct lu_attr *attr, struct lu_fid *parent_fid,
247                            struct dt_object_format *dof,
248                            struct thandle_exec_args *ta,
249                            struct update_reply *reply,
250                            int index, char *file, int line)
251 {
252         struct tx_arg *arg;
253
254         LASSERT(ta->ta_handle != NULL);
255         ta->ta_err = dt_declare_create(env, obj, attr, NULL, dof,
256                                        ta->ta_handle);
257         if (ta->ta_err != 0)
258                 return ta->ta_err;
259
260         arg = tx_add_exec(ta, out_tx_create_exec, out_tx_create_undo, file,
261                           line);
262         LASSERT(arg);
263
264         /* release the object in out_trans_stop */
265         lu_object_get(&obj->do_lu);
266         arg->object = obj;
267         arg->u.create.attr = *attr;
268         if (parent_fid)
269                 arg->u.create.fid = *parent_fid;
270         memset(&arg->u.create.hint, 0, sizeof(arg->u.create.hint));
271         arg->u.create.dof  = *dof;
272         arg->reply = reply;
273         arg->index = index;
274
275         return 0;
276 }
277
278 static int out_create(struct tgt_session_info *tsi)
279 {
280         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
281         struct update           *update = tti->tti_u.update.tti_update;
282         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
283         struct dt_object_format *dof = &tti->tti_u.update.tti_update_dof;
284         struct obdo             *lobdo = &tti->tti_u.update.tti_obdo;
285         struct lu_attr          *attr = &tti->tti_attr;
286         struct lu_fid           *fid = NULL;
287         struct obdo             *wobdo;
288         int                     size;
289         int                     rc;
290
291         ENTRY;
292
293         wobdo = update_param_buf(update, 0, &size);
294         if (wobdo == NULL || size != sizeof(*wobdo)) {
295                 CERROR("%s: obdo is NULL, invalid RPC: rc = %d\n",
296                        tgt_name(tsi->tsi_tgt), -EPROTO);
297                 RETURN(err_serious(-EPROTO));
298         }
299
300         obdo_le_to_cpu(wobdo, wobdo);
301         lustre_get_wire_obdo(NULL, lobdo, wobdo);
302         la_from_obdo(attr, lobdo, lobdo->o_valid);
303
304         dof->dof_type = dt_mode_to_dft(attr->la_mode);
305         if (S_ISDIR(attr->la_mode)) {
306                 int size;
307
308                 fid = update_param_buf(update, 1, &size);
309                 if (fid == NULL || size != sizeof(*fid)) {
310                         CERROR("%s: invalid fid: rc = %d\n",
311                                tgt_name(tsi->tsi_tgt), -EPROTO);
312                         RETURN(err_serious(-EPROTO));
313                 }
314                 fid_le_to_cpu(fid, fid);
315                 if (!fid_is_sane(fid)) {
316                         CERROR("%s: invalid fid "DFID": rc = %d\n",
317                                tgt_name(tsi->tsi_tgt), PFID(fid), -EPROTO);
318                         RETURN(err_serious(-EPROTO));
319                 }
320         }
321
322         if (lu_object_exists(&obj->do_lu))
323                 RETURN(-EEXIST);
324
325         rc = out_tx_create(tsi->tsi_env, obj, attr, fid, dof,
326                            &tti->tti_tea,
327                            tti->tti_u.update.tti_update_reply,
328                            tti->tti_u.update.tti_update_reply_index);
329
330         RETURN(rc);
331 }
332
333 static int out_tx_attr_set_undo(const struct lu_env *env,
334                                 struct thandle *th, struct tx_arg *arg)
335 {
336         CERROR("%s: attr set undo "DFID" unimplemented yet!: rc = %d\n",
337                dt_obd_name(th->th_dev),
338                PFID(lu_object_fid(&arg->object->do_lu)), -ENOTSUPP);
339
340         return -ENOTSUPP;
341 }
342
343 static int out_tx_attr_set_exec(const struct lu_env *env, struct thandle *th,
344                                 struct tx_arg *arg)
345 {
346         struct dt_object        *dt_obj = arg->object;
347         int                     rc;
348
349         CDEBUG(D_OTHER, "%s: attr set "DFID"\n", dt_obd_name(th->th_dev),
350                PFID(lu_object_fid(&dt_obj->do_lu)));
351
352         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
353         rc = dt_attr_set(env, dt_obj, &arg->u.attr_set.attr, th, NULL);
354         dt_write_unlock(env, dt_obj);
355
356         CDEBUG(D_INFO, "%s: insert attr_set reply %p index %d: rc = %d\n",
357                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
358
359         update_insert_reply(arg->reply, NULL, 0, arg->index, rc);
360
361         return rc;
362 }
363
364 static int __out_tx_attr_set(const struct lu_env *env,
365                              struct dt_object *dt_obj,
366                              const struct lu_attr *attr,
367                              struct thandle_exec_args *th,
368                              struct update_reply *reply, int index,
369                              char *file, int line)
370 {
371         struct tx_arg           *arg;
372
373         LASSERT(th->ta_handle != NULL);
374         th->ta_err = dt_declare_attr_set(env, dt_obj, attr, th->ta_handle);
375         if (th->ta_err != 0)
376                 return th->ta_err;
377
378         arg = tx_add_exec(th, out_tx_attr_set_exec, out_tx_attr_set_undo,
379                           file, line);
380         LASSERT(arg);
381         lu_object_get(&dt_obj->do_lu);
382         arg->object = dt_obj;
383         arg->u.attr_set.attr = *attr;
384         arg->reply = reply;
385         arg->index = index;
386         return 0;
387 }
388
389 static int out_attr_set(struct tgt_session_info *tsi)
390 {
391         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
392         struct update           *update = tti->tti_u.update.tti_update;
393         struct lu_attr          *attr = &tti->tti_attr;
394         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
395         struct obdo             *lobdo = &tti->tti_u.update.tti_obdo;
396         struct obdo             *wobdo;
397         int                      size;
398         int                      rc;
399
400         ENTRY;
401
402         wobdo = update_param_buf(update, 0, &size);
403         if (wobdo == NULL || size != sizeof(*wobdo)) {
404                 CERROR("%s: empty obdo in the update: rc = %d\n",
405                        tgt_name(tsi->tsi_tgt), -EPROTO);
406                 RETURN(err_serious(-EPROTO));
407         }
408
409         attr->la_valid = 0;
410         attr->la_valid = 0;
411         obdo_le_to_cpu(wobdo, wobdo);
412         lustre_get_wire_obdo(NULL, lobdo, wobdo);
413         la_from_obdo(attr, lobdo, lobdo->o_valid);
414
415         rc = out_tx_attr_set(tsi->tsi_env, obj, attr, &tti->tti_tea,
416                              tti->tti_u.update.tti_update_reply,
417                              tti->tti_u.update.tti_update_reply_index);
418
419         RETURN(rc);
420 }
421
422 static int out_attr_get(struct tgt_session_info *tsi)
423 {
424         const struct lu_env     *env = tsi->tsi_env;
425         struct tgt_thread_info  *tti = tgt_th_info(env);
426         struct obdo             *obdo = &tti->tti_u.update.tti_obdo;
427         struct lu_attr          *la = &tti->tti_attr;
428         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
429         int                     rc;
430
431         ENTRY;
432
433         if (!lu_object_exists(&obj->do_lu))
434                 RETURN(-ENOENT);
435
436         dt_read_lock(env, obj, MOR_TGT_CHILD);
437         rc = dt_attr_get(env, obj, la, NULL);
438         if (rc)
439                 GOTO(out_unlock, rc);
440         /*
441          * If it is a directory, we will also check whether the
442          * directory is empty.
443          * la_flags = 0 : Empty.
444          *          = 1 : Not empty.
445          */
446         la->la_flags = 0;
447         if (S_ISDIR(la->la_mode)) {
448                 struct dt_it            *it;
449                 const struct dt_it_ops  *iops;
450
451                 if (!dt_try_as_dir(env, obj))
452                         GOTO(out_unlock, rc = -ENOTDIR);
453
454                 iops = &obj->do_index_ops->dio_it;
455                 it = iops->init(env, obj, LUDA_64BITHASH, BYPASS_CAPA);
456                 if (!IS_ERR(it)) {
457                         int  result;
458                         result = iops->get(env, it, (const void *)"");
459                         if (result > 0) {
460                                 int i;
461                                 for (result = 0, i = 0; result == 0 && i < 3;
462                                      ++i)
463                                         result = iops->next(env, it);
464                                 if (result == 0)
465                                         la->la_flags = 1;
466                         } else if (result == 0)
467                                 /*
468                                  * Huh? Index contains no zero key?
469                                  */
470                                 rc = -EIO;
471
472                         iops->put(env, it);
473                         iops->fini(env, it);
474                 }
475         }
476
477         obdo->o_valid = 0;
478         obdo_from_la(obdo, la, la->la_valid);
479         obdo_cpu_to_le(obdo, obdo);
480         lustre_set_wire_obdo(NULL, obdo, obdo);
481
482 out_unlock:
483         dt_read_unlock(env, obj);
484
485         CDEBUG(D_INFO, "%s: insert attr get reply %p index %d: rc = %d\n",
486                tgt_name(tsi->tsi_tgt), tti->tti_u.update.tti_update_reply,
487                0, rc);
488
489         update_insert_reply(tti->tti_u.update.tti_update_reply, obdo,
490                             sizeof(*obdo), 0, rc);
491         RETURN(rc);
492 }
493
494 static int out_xattr_get(struct tgt_session_info *tsi)
495 {
496         const struct lu_env     *env = tsi->tsi_env;
497         struct tgt_thread_info  *tti = tgt_th_info(env);
498         struct update           *update = tti->tti_u.update.tti_update;
499         struct lu_buf           *lbuf = &tti->tti_buf;
500         struct update_reply     *reply = tti->tti_u.update.tti_update_reply;
501         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
502         char                    *name;
503         void                    *ptr;
504         int                      rc;
505
506         ENTRY;
507
508         name = (char *)update_param_buf(update, 0, NULL);
509         if (name == NULL) {
510                 CERROR("%s: empty name for xattr get: rc = %d\n",
511                        tgt_name(tsi->tsi_tgt), -EPROTO);
512                 RETURN(err_serious(-EPROTO));
513         }
514
515         ptr = update_get_buf_internal(reply, 0, NULL);
516         LASSERT(ptr != NULL);
517
518         /* The first 4 bytes(int) are used to store the result */
519         lbuf->lb_buf = (char *)ptr + sizeof(int);
520         lbuf->lb_len = UPDATE_BUFFER_SIZE - sizeof(struct update_reply);
521         dt_read_lock(env, obj, MOR_TGT_CHILD);
522         rc = dt_xattr_get(env, obj, lbuf, name, NULL);
523         dt_read_unlock(env, obj);
524         if (rc < 0) {
525                 lbuf->lb_len = 0;
526                 GOTO(out, rc);
527         }
528         if (rc == 0) {
529                 lbuf->lb_len = 0;
530                 GOTO(out, rc = -ENOENT);
531         }
532         lbuf->lb_len = rc;
533         rc = 0;
534         CDEBUG(D_INFO, "%s: "DFID" get xattr %s len %d\n",
535                tgt_name(tsi->tsi_tgt), PFID(lu_object_fid(&obj->do_lu)),
536                name, (int)lbuf->lb_len);
537 out:
538         *(int *)ptr = rc;
539         reply->ur_lens[0] = lbuf->lb_len + sizeof(int);
540         RETURN(rc);
541 }
542
543 static int out_index_lookup(struct tgt_session_info *tsi)
544 {
545         const struct lu_env     *env = tsi->tsi_env;
546         struct tgt_thread_info  *tti = tgt_th_info(env);
547         struct update           *update = tti->tti_u.update.tti_update;
548         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
549         char                    *name;
550         int                      rc;
551
552         ENTRY;
553
554         if (!lu_object_exists(&obj->do_lu))
555                 RETURN(-ENOENT);
556
557         name = (char *)update_param_buf(update, 0, NULL);
558         if (name == NULL) {
559                 CERROR("%s: empty name for lookup: rc = %d\n",
560                        tgt_name(tsi->tsi_tgt), -EPROTO);
561                 RETURN(err_serious(-EPROTO));
562         }
563
564         dt_read_lock(env, obj, MOR_TGT_CHILD);
565         if (!dt_try_as_dir(env, obj))
566                 GOTO(out_unlock, rc = -ENOTDIR);
567
568         rc = dt_lookup(env, obj, (struct dt_rec *)&tti->tti_fid1,
569                 (struct dt_key *)name, NULL);
570
571         if (rc < 0)
572                 GOTO(out_unlock, rc);
573
574         if (rc == 0)
575                 rc += 1;
576
577         CDEBUG(D_INFO, "lookup "DFID" %s get "DFID" rc %d\n",
578                PFID(lu_object_fid(&obj->do_lu)), name,
579                PFID(&tti->tti_fid1), rc);
580         fid_cpu_to_le(&tti->tti_fid1, &tti->tti_fid1);
581
582 out_unlock:
583         dt_read_unlock(env, obj);
584
585         CDEBUG(D_INFO, "%s: insert lookup reply %p index %d: rc = %d\n",
586                tgt_name(tsi->tsi_tgt), tti->tti_u.update.tti_update_reply,
587                0, rc);
588
589         update_insert_reply(tti->tti_u.update.tti_update_reply,
590                             &tti->tti_fid1, sizeof(tti->tti_fid1), 0, rc);
591         RETURN(rc);
592 }
593
594 static int out_tx_xattr_set_exec(const struct lu_env *env,
595                                  struct thandle *th,
596                                  struct tx_arg *arg)
597 {
598         struct dt_object *dt_obj = arg->object;
599         int rc;
600
601         CDEBUG(D_INFO, "%s: set xattr buf %p name %s flag %d\n",
602                dt_obd_name(th->th_dev), arg->u.xattr_set.buf.lb_buf,
603                arg->u.xattr_set.name, arg->u.xattr_set.flags);
604
605         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
606         rc = dt_xattr_set(env, dt_obj, &arg->u.xattr_set.buf,
607                           arg->u.xattr_set.name, arg->u.xattr_set.flags,
608                           th, NULL);
609         dt_write_unlock(env, dt_obj);
610         /**
611          * Ignore errors if this is LINK EA
612          **/
613         if (unlikely(rc && !strncmp(arg->u.xattr_set.name, XATTR_NAME_LINK,
614                                     strlen(XATTR_NAME_LINK))))
615                 rc = 0;
616
617         CDEBUG(D_INFO, "%s: insert xattr set reply %p index %d: rc = %d\n",
618                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
619
620         update_insert_reply(arg->reply, NULL, 0, arg->index, rc);
621
622         return rc;
623 }
624
625 static int __out_tx_xattr_set(const struct lu_env *env,
626                               struct dt_object *dt_obj,
627                               const struct lu_buf *buf,
628                               const char *name, int flags,
629                               struct thandle_exec_args *ta,
630                               struct update_reply *reply, int index,
631                               char *file, int line)
632 {
633         struct tx_arg           *arg;
634
635         LASSERT(ta->ta_handle != NULL);
636         ta->ta_err = dt_declare_xattr_set(env, dt_obj, buf, name,
637                                           flags, ta->ta_handle);
638         if (ta->ta_err != 0)
639                 return ta->ta_err;
640
641         arg = tx_add_exec(ta, out_tx_xattr_set_exec, NULL, file, line);
642         LASSERT(arg);
643         lu_object_get(&dt_obj->do_lu);
644         arg->object = dt_obj;
645         arg->u.xattr_set.name = name;
646         arg->u.xattr_set.flags = flags;
647         arg->u.xattr_set.buf = *buf;
648         arg->reply = reply;
649         arg->index = index;
650         arg->u.xattr_set.csum = 0;
651         return 0;
652 }
653
654 static int out_xattr_set(struct tgt_session_info *tsi)
655 {
656         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
657         struct update           *update = tti->tti_u.update.tti_update;
658         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
659         struct lu_buf           *lbuf = &tti->tti_buf;
660         char                    *name;
661         char                    *buf;
662         char                    *tmp;
663         int                      buf_len = 0;
664         int                      flag;
665         int                      rc;
666         ENTRY;
667
668         name = update_param_buf(update, 0, NULL);
669         if (name == NULL) {
670                 CERROR("%s: empty name for xattr set: rc = %d\n",
671                        tgt_name(tsi->tsi_tgt), -EPROTO);
672                 RETURN(err_serious(-EPROTO));
673         }
674
675         buf = (char *)update_param_buf(update, 1, &buf_len);
676         if (buf == NULL || buf_len == 0) {
677                 CERROR("%s: empty buf for xattr set: rc = %d\n",
678                        tgt_name(tsi->tsi_tgt), -EPROTO);
679                 RETURN(err_serious(-EPROTO));
680         }
681
682         lbuf->lb_buf = buf;
683         lbuf->lb_len = buf_len;
684
685         tmp = (char *)update_param_buf(update, 2, NULL);
686         if (tmp == NULL) {
687                 CERROR("%s: empty flag for xattr set: rc = %d\n",
688                        tgt_name(tsi->tsi_tgt), -EPROTO);
689                 RETURN(err_serious(-EPROTO));
690         }
691
692         flag = le32_to_cpu(*(int *)tmp);
693
694         rc = out_tx_xattr_set(tsi->tsi_env, obj, lbuf, name, flag,
695                               &tti->tti_tea,
696                               tti->tti_u.update.tti_update_reply,
697                               tti->tti_u.update.tti_update_reply_index);
698         RETURN(rc);
699 }
700
701 static int out_obj_ref_add(const struct lu_env *env,
702                            struct dt_object *dt_obj,
703                            struct thandle *th)
704 {
705         int rc;
706
707         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
708         rc = dt_ref_add(env, dt_obj, th);
709         dt_write_unlock(env, dt_obj);
710
711         return rc;
712 }
713
714 static int out_obj_ref_del(const struct lu_env *env,
715                            struct dt_object *dt_obj,
716                            struct thandle *th)
717 {
718         int rc;
719
720         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
721         rc = dt_ref_del(env, dt_obj, th);
722         dt_write_unlock(env, dt_obj);
723
724         return rc;
725 }
726
727 static int out_tx_ref_add_exec(const struct lu_env *env, struct thandle *th,
728                                struct tx_arg *arg)
729 {
730         struct dt_object *dt_obj = arg->object;
731         int rc;
732
733         rc = out_obj_ref_add(env, dt_obj, th);
734
735         CDEBUG(D_INFO, "%s: insert ref_add reply %p index %d: rc = %d\n",
736                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
737
738         update_insert_reply(arg->reply, NULL, 0, arg->index, rc);
739         return rc;
740 }
741
742 static int out_tx_ref_add_undo(const struct lu_env *env, struct thandle *th,
743                                struct tx_arg *arg)
744 {
745         return out_obj_ref_del(env, arg->object, th);
746 }
747
748 static int __out_tx_ref_add(const struct lu_env *env,
749                             struct dt_object *dt_obj,
750                             struct thandle_exec_args *ta,
751                             struct update_reply *reply,
752                             int index, char *file, int line)
753 {
754         struct tx_arg   *arg;
755
756         LASSERT(ta->ta_handle != NULL);
757         ta->ta_err = dt_declare_ref_add(env, dt_obj, ta->ta_handle);
758         if (ta->ta_err != 0)
759                 return ta->ta_err;
760
761         arg = tx_add_exec(ta, out_tx_ref_add_exec, out_tx_ref_add_undo, file,
762                           line);
763         LASSERT(arg);
764         lu_object_get(&dt_obj->do_lu);
765         arg->object = dt_obj;
766         arg->reply = reply;
767         arg->index = index;
768         return 0;
769 }
770
771 /**
772  * increase ref of the object
773  **/
774 static int out_ref_add(struct tgt_session_info *tsi)
775 {
776         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
777         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
778         int                      rc;
779
780         ENTRY;
781
782         rc = out_tx_ref_add(tsi->tsi_env, obj, &tti->tti_tea,
783                             tti->tti_u.update.tti_update_reply,
784                             tti->tti_u.update.tti_update_reply_index);
785         RETURN(rc);
786 }
787
788 static int out_tx_ref_del_exec(const struct lu_env *env, struct thandle *th,
789                                struct tx_arg *arg)
790 {
791         struct dt_object        *dt_obj = arg->object;
792         int                      rc;
793
794         rc = out_obj_ref_del(env, dt_obj, th);
795
796         CDEBUG(D_INFO, "%s: insert ref_del reply %p index %d: rc = %d\n",
797                dt_obd_name(th->th_dev), arg->reply, arg->index, 0);
798
799         update_insert_reply(arg->reply, NULL, 0, arg->index, rc);
800
801         return rc;
802 }
803
804 static int out_tx_ref_del_undo(const struct lu_env *env, struct thandle *th,
805                                struct tx_arg *arg)
806 {
807         return out_obj_ref_add(env, arg->object, th);
808 }
809
810 static int __out_tx_ref_del(const struct lu_env *env,
811                             struct dt_object *dt_obj,
812                             struct thandle_exec_args *ta,
813                             struct update_reply *reply,
814                             int index, char *file, int line)
815 {
816         struct tx_arg   *arg;
817
818         LASSERT(ta->ta_handle != NULL);
819         ta->ta_err = dt_declare_ref_del(env, dt_obj, ta->ta_handle);
820         if (ta->ta_err != 0)
821                 return ta->ta_err;
822
823         arg = tx_add_exec(ta, out_tx_ref_del_exec, out_tx_ref_del_undo, file,
824                           line);
825         LASSERT(arg);
826         lu_object_get(&dt_obj->do_lu);
827         arg->object = dt_obj;
828         arg->reply = reply;
829         arg->index = index;
830         return 0;
831 }
832
833 static int out_ref_del(struct tgt_session_info *tsi)
834 {
835         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
836         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
837         int                      rc;
838
839         ENTRY;
840
841         if (!lu_object_exists(&obj->do_lu))
842                 RETURN(-ENOENT);
843
844         rc = out_tx_ref_del(tsi->tsi_env, obj, &tti->tti_tea,
845                             tti->tti_u.update.tti_update_reply,
846                             tti->tti_u.update.tti_update_reply_index);
847         RETURN(rc);
848 }
849
850 static int out_obj_index_insert(const struct lu_env *env,
851                                 struct dt_object *dt_obj,
852                                 const struct dt_rec *rec,
853                                 const struct dt_key *key,
854                                 struct thandle *th)
855 {
856         int rc;
857
858         CDEBUG(D_INFO, "%s: index insert "DFID" name: %s fid "DFID"\n",
859                dt_obd_name(th->th_dev), PFID(lu_object_fid(&dt_obj->do_lu)),
860                (char *)key, PFID((struct lu_fid *)rec));
861
862         if (dt_try_as_dir(env, dt_obj) == 0)
863                 return -ENOTDIR;
864
865         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
866         rc = dt_insert(env, dt_obj, rec, key, th, NULL, 0);
867         dt_write_unlock(env, dt_obj);
868
869         return rc;
870 }
871
872 static int out_obj_index_delete(const struct lu_env *env,
873                                 struct dt_object *dt_obj,
874                                 const struct dt_key *key,
875                                 struct thandle *th)
876 {
877         int rc;
878
879         CDEBUG(D_INFO, "%s: index delete "DFID" name: %s\n",
880                dt_obd_name(th->th_dev), PFID(lu_object_fid(&dt_obj->do_lu)),
881                (char *)key);
882
883         if (dt_try_as_dir(env, dt_obj) == 0)
884                 return -ENOTDIR;
885
886         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
887         rc = dt_delete(env, dt_obj, key, th, NULL);
888         dt_write_unlock(env, dt_obj);
889
890         return rc;
891 }
892
893 static int out_tx_index_insert_exec(const struct lu_env *env,
894                                     struct thandle *th, struct tx_arg *arg)
895 {
896         struct dt_object *dt_obj = arg->object;
897         int rc;
898
899         rc = out_obj_index_insert(env, dt_obj, arg->u.insert.rec,
900                                   arg->u.insert.key, th);
901
902         CDEBUG(D_INFO, "%s: insert idx insert reply %p index %d: rc = %d\n",
903                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
904
905         update_insert_reply(arg->reply, NULL, 0, arg->index, rc);
906
907         return rc;
908 }
909
910 static int out_tx_index_insert_undo(const struct lu_env *env,
911                                     struct thandle *th, struct tx_arg *arg)
912 {
913         return out_obj_index_delete(env, arg->object, arg->u.insert.key, th);
914 }
915
916 static int __out_tx_index_insert(const struct lu_env *env,
917                                  struct dt_object *dt_obj,
918                                  char *name, struct lu_fid *fid,
919                                  struct thandle_exec_args *ta,
920                                  struct update_reply *reply,
921                                  int index, char *file, int line)
922 {
923         struct tx_arg *arg;
924
925         LASSERT(ta->ta_handle != NULL);
926
927         if (lu_object_exists(&dt_obj->do_lu)) {
928                 if (dt_try_as_dir(env, dt_obj) == 0) {
929                         ta->ta_err = -ENOTDIR;
930                         return ta->ta_err;
931                 }
932                 ta->ta_err = dt_declare_insert(env, dt_obj,
933                                                (struct dt_rec *)fid,
934                                                (struct dt_key *)name,
935                                                ta->ta_handle);
936         }
937
938         if (ta->ta_err != 0)
939                 return ta->ta_err;
940
941         arg = tx_add_exec(ta, out_tx_index_insert_exec,
942                           out_tx_index_insert_undo, file,
943                           line);
944         LASSERT(arg);
945         lu_object_get(&dt_obj->do_lu);
946         arg->object = dt_obj;
947         arg->reply = reply;
948         arg->index = index;
949         arg->u.insert.rec = (struct dt_rec *)fid;
950         arg->u.insert.key = (struct dt_key *)name;
951
952         return 0;
953 }
954
955 static int out_index_insert(struct tgt_session_info *tsi)
956 {
957         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
958         struct update     *update = tti->tti_u.update.tti_update;
959         struct dt_object  *obj = tti->tti_u.update.tti_dt_object;
960         struct lu_fid     *fid;
961         char              *name;
962         int                rc = 0;
963         int                size;
964
965         ENTRY;
966
967         name = (char *)update_param_buf(update, 0, NULL);
968         if (name == NULL) {
969                 CERROR("%s: empty name for index insert: rc = %d\n",
970                        tgt_name(tsi->tsi_tgt), -EPROTO);
971                 RETURN(err_serious(-EPROTO));
972         }
973
974         fid = (struct lu_fid *)update_param_buf(update, 1, &size);
975         if (fid == NULL || size != sizeof(*fid)) {
976                 CERROR("%s: invalid fid: rc = %d\n",
977                        tgt_name(tsi->tsi_tgt), -EPROTO);
978                        RETURN(err_serious(-EPROTO));
979         }
980
981         fid_le_to_cpu(fid, fid);
982         if (!fid_is_sane(fid)) {
983                 CERROR("%s: invalid FID "DFID": rc = %d\n",
984                        tgt_name(tsi->tsi_tgt), PFID(fid), -EPROTO);
985                 RETURN(err_serious(-EPROTO));
986         }
987
988         rc = out_tx_index_insert(tsi->tsi_env, obj, name, fid,
989                                  &tti->tti_tea,
990                                  tti->tti_u.update.tti_update_reply,
991                                  tti->tti_u.update.tti_update_reply_index);
992         RETURN(rc);
993 }
994
995 static int out_tx_index_delete_exec(const struct lu_env *env,
996                                     struct thandle *th,
997                                     struct tx_arg *arg)
998 {
999         int rc;
1000
1001         rc = out_obj_index_delete(env, arg->object, arg->u.insert.key, th);
1002
1003         CDEBUG(D_INFO, "%s: insert idx insert reply %p index %d: rc = %d\n",
1004                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
1005
1006         update_insert_reply(arg->reply, NULL, 0, arg->index, rc);
1007
1008         return rc;
1009 }
1010
1011 static int out_tx_index_delete_undo(const struct lu_env *env,
1012                                     struct thandle *th,
1013                                     struct tx_arg *arg)
1014 {
1015         CERROR("%s: Oops, can not rollback index_delete yet: rc = %d\n",
1016                dt_obd_name(th->th_dev), -ENOTSUPP);
1017         return -ENOTSUPP;
1018 }
1019
1020 static int __out_tx_index_delete(const struct lu_env *env,
1021                                  struct dt_object *dt_obj, char *name,
1022                                  struct thandle_exec_args *ta,
1023                                  struct update_reply *reply,
1024                                  int index, char *file, int line)
1025 {
1026         struct tx_arg *arg;
1027
1028         if (dt_try_as_dir(env, dt_obj) == 0) {
1029                 ta->ta_err = -ENOTDIR;
1030                 return ta->ta_err;
1031         }
1032
1033         LASSERT(ta->ta_handle != NULL);
1034         ta->ta_err = dt_declare_delete(env, dt_obj,
1035                                        (struct dt_key *)name,
1036                                        ta->ta_handle);
1037         if (ta->ta_err != 0)
1038                 return ta->ta_err;
1039
1040         arg = tx_add_exec(ta, out_tx_index_delete_exec,
1041                           out_tx_index_delete_undo, file,
1042                           line);
1043         LASSERT(arg);
1044         lu_object_get(&dt_obj->do_lu);
1045         arg->object = dt_obj;
1046         arg->reply = reply;
1047         arg->index = index;
1048         arg->u.insert.key = (struct dt_key *)name;
1049         return 0;
1050 }
1051
1052 static int out_index_delete(struct tgt_session_info *tsi)
1053 {
1054         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
1055         struct update           *update = tti->tti_u.update.tti_update;
1056         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
1057         char                    *name;
1058         int                      rc = 0;
1059
1060         if (!lu_object_exists(&obj->do_lu))
1061                 RETURN(-ENOENT);
1062
1063         name = (char *)update_param_buf(update, 0, NULL);
1064         if (name == NULL) {
1065                 CERROR("%s: empty name for index delete: rc = %d\n",
1066                        tgt_name(tsi->tsi_tgt), -EPROTO);
1067                 RETURN(err_serious(-EPROTO));
1068         }
1069
1070         rc = out_tx_index_delete(tsi->tsi_env, obj, name, &tti->tti_tea,
1071                                  tti->tti_u.update.tti_update_reply,
1072                                  tti->tti_u.update.tti_update_reply_index);
1073         RETURN(rc);
1074 }
1075
1076 static int out_tx_destroy_exec(const struct lu_env *env, struct thandle *th,
1077                                struct tx_arg *arg)
1078 {
1079         struct dt_object *dt_obj = arg->object;
1080         int rc;
1081
1082         rc = out_obj_destroy(env, dt_obj, th);
1083
1084         CDEBUG(D_INFO, "%s: insert destroy reply %p index %d: rc = %d\n",
1085                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
1086
1087         update_insert_reply(arg->reply, NULL, 0, arg->index, rc);
1088
1089         RETURN(rc);
1090 }
1091
1092 static int out_tx_destroy_undo(const struct lu_env *env, struct thandle *th,
1093                                struct tx_arg *arg)
1094 {
1095         CERROR("%s: not support destroy undo yet!: rc = %d\n",
1096                dt_obd_name(th->th_dev), -ENOTSUPP);
1097         return -ENOTSUPP;
1098 }
1099
1100 static int __out_tx_destroy(const struct lu_env *env, struct dt_object *dt_obj,
1101                              struct thandle_exec_args *ta,
1102                              struct update_reply *reply,
1103                              int index, char *file, int line)
1104 {
1105         struct tx_arg *arg;
1106
1107         LASSERT(ta->ta_handle != NULL);
1108         ta->ta_err = dt_declare_destroy(env, dt_obj, ta->ta_handle);
1109         if (ta->ta_err)
1110                 return ta->ta_err;
1111
1112         arg = tx_add_exec(ta, out_tx_destroy_exec, out_tx_destroy_undo,
1113                           file, line);
1114         LASSERT(arg);
1115         lu_object_get(&dt_obj->do_lu);
1116         arg->object = dt_obj;
1117         arg->reply = reply;
1118         arg->index = index;
1119         return 0;
1120 }
1121
1122 static int out_destroy(struct tgt_session_info *tsi)
1123 {
1124         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
1125         struct update           *update = tti->tti_u.update.tti_update;
1126         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
1127         struct lu_fid           *fid;
1128         int                      rc;
1129         ENTRY;
1130
1131         fid = &update->u_fid;
1132         fid_le_to_cpu(fid, fid);
1133         if (!fid_is_sane(fid)) {
1134                 CERROR("%s: invalid FID "DFID": rc = %d\n",
1135                        tgt_name(tsi->tsi_tgt), PFID(fid), -EPROTO);
1136                 RETURN(err_serious(-EPROTO));
1137         }
1138
1139         if (!lu_object_exists(&obj->do_lu))
1140                 RETURN(-ENOENT);
1141
1142         rc = out_tx_destroy(tsi->tsi_env, obj, &tti->tti_tea,
1143                             tti->tti_u.update.tti_update_reply,
1144                             tti->tti_u.update.tti_update_reply_index);
1145
1146         RETURN(rc);
1147 }
1148
1149 #define DEF_OUT_HNDL(opc, name, flags, fn)     \
1150 [opc - OBJ_CREATE] = {                                  \
1151         .th_name    = name,                             \
1152         .th_fail_id = 0,                                \
1153         .th_opc     = opc,                              \
1154         .th_flags   = flags,                            \
1155         .th_act     = fn,                               \
1156         .th_fmt     = NULL,                             \
1157         .th_version = 0,                                \
1158 }
1159
1160 #define out_handler mdt_handler
1161 static struct tgt_handler out_update_ops[] = {
1162         DEF_OUT_HNDL(OBJ_CREATE, "obj_create", MUTABOR | HABEO_REFERO,
1163                      out_create),
1164         DEF_OUT_HNDL(OBJ_DESTROY, "obj_create", MUTABOR | HABEO_REFERO,
1165                      out_destroy),
1166         DEF_OUT_HNDL(OBJ_REF_ADD, "obj_ref_add", MUTABOR | HABEO_REFERO,
1167                      out_ref_add),
1168         DEF_OUT_HNDL(OBJ_REF_DEL, "obj_ref_del", MUTABOR | HABEO_REFERO,
1169                      out_ref_del),
1170         DEF_OUT_HNDL(OBJ_ATTR_SET, "obj_attr_set",  MUTABOR | HABEO_REFERO,
1171                      out_attr_set),
1172         DEF_OUT_HNDL(OBJ_ATTR_GET, "obj_attr_get",  HABEO_REFERO,
1173                      out_attr_get),
1174         DEF_OUT_HNDL(OBJ_XATTR_SET, "obj_xattr_set", MUTABOR | HABEO_REFERO,
1175                      out_xattr_set),
1176         DEF_OUT_HNDL(OBJ_XATTR_GET, "obj_xattr_get", HABEO_REFERO,
1177                      out_xattr_get),
1178         DEF_OUT_HNDL(OBJ_INDEX_LOOKUP, "obj_index_lookup", HABEO_REFERO,
1179                      out_index_lookup),
1180         DEF_OUT_HNDL(OBJ_INDEX_INSERT, "obj_index_insert",
1181                      MUTABOR | HABEO_REFERO, out_index_insert),
1182         DEF_OUT_HNDL(OBJ_INDEX_DELETE, "obj_index_delete",
1183                      MUTABOR | HABEO_REFERO, out_index_delete),
1184 };
1185
1186 struct tgt_handler *out_handler_find(__u32 opc)
1187 {
1188         struct tgt_handler *h;
1189
1190         h = NULL;
1191         if (OBJ_CREATE <= opc && opc < OBJ_LAST) {
1192                 h = &out_update_ops[opc - OBJ_CREATE];
1193                 LASSERTF(h->th_opc == opc, "opcode mismatch %d != %d\n",
1194                          h->th_opc, opc);
1195         } else {
1196                 h = NULL; /* unsupported opc */
1197         }
1198         return h;
1199 }
1200
1201 /**
1202  * Object updates between Targets. Because all the updates has been
1203  * dis-assemblied into object updates at sender side, so OUT will
1204  * call OSD API directly to execute these updates.
1205  *
1206  * In DNE phase I all of the updates in the request need to be executed
1207  * in one transaction, and the transaction has to be synchronously.
1208  *
1209  * Please refer to lustre/include/lustre/lustre_idl.h for req/reply
1210  * format.
1211  */
1212 int out_handle(struct tgt_session_info *tsi)
1213 {
1214         const struct lu_env             *env = tsi->tsi_env;
1215         struct tgt_thread_info          *tti = tgt_th_info(env);
1216         struct thandle_exec_args        *ta = &tti->tti_tea;
1217         struct req_capsule              *pill = tsi->tsi_pill;
1218         struct dt_device                *dt = tsi->tsi_tgt->lut_bottom;
1219         struct update_buf               *ubuf;
1220         struct update                   *update;
1221         struct update_reply             *update_reply;
1222         int                              bufsize;
1223         int                              count;
1224         int                              old_batchid = -1;
1225         unsigned                         off;
1226         int                              i;
1227         int                              rc = 0;
1228         int                              rc1 = 0;
1229
1230         ENTRY;
1231
1232         req_capsule_set(pill, &RQF_UPDATE_OBJ);
1233         bufsize = req_capsule_get_size(pill, &RMF_UPDATE, RCL_CLIENT);
1234         if (bufsize != UPDATE_BUFFER_SIZE) {
1235                 CERROR("%s: invalid bufsize %d: rc = %d\n",
1236                        tgt_name(tsi->tsi_tgt), bufsize, -EPROTO);
1237                 RETURN(err_serious(-EPROTO));
1238         }
1239
1240         ubuf = req_capsule_client_get(pill, &RMF_UPDATE);
1241         if (ubuf == NULL) {
1242                 CERROR("%s: No buf!: rc = %d\n", tgt_name(tsi->tsi_tgt),
1243                        -EPROTO);
1244                 RETURN(err_serious(-EPROTO));
1245         }
1246
1247         if (le32_to_cpu(ubuf->ub_magic) != UPDATE_BUFFER_MAGIC) {
1248                 CERROR("%s: invalid magic %x expect %x: rc = %d\n",
1249                        tgt_name(tsi->tsi_tgt), le32_to_cpu(ubuf->ub_magic),
1250                        UPDATE_BUFFER_MAGIC, -EPROTO);
1251                 RETURN(err_serious(-EPROTO));
1252         }
1253
1254         count = le32_to_cpu(ubuf->ub_count);
1255         if (count <= 0) {
1256                 CERROR("%s: No update!: rc = %d\n",
1257                        tgt_name(tsi->tsi_tgt), -EPROTO);
1258                 RETURN(err_serious(-EPROTO));
1259         }
1260
1261         req_capsule_set_size(pill, &RMF_UPDATE_REPLY, RCL_SERVER,
1262                              UPDATE_BUFFER_SIZE);
1263         rc = req_capsule_server_pack(pill);
1264         if (rc != 0) {
1265                 CERROR("%s: Can't pack response: rc = %d\n",
1266                        tgt_name(tsi->tsi_tgt), rc);
1267                 RETURN(rc);
1268         }
1269
1270         /* Prepare the update reply buffer */
1271         update_reply = req_capsule_server_get(pill, &RMF_UPDATE_REPLY);
1272         if (update_reply == NULL)
1273                 RETURN(err_serious(-EPROTO));
1274         update_init_reply_buf(update_reply, count);
1275         tti->tti_u.update.tti_update_reply = update_reply;
1276
1277         rc = out_tx_start(env, dt, ta);
1278         if (rc != 0)
1279                 RETURN(rc);
1280
1281         tti->tti_mult_trans = !req_is_replay(tgt_ses_req(tsi));
1282
1283         /* Walk through updates in the request to execute them synchronously */
1284         off = cfs_size_round(offsetof(struct update_buf, ub_bufs[0]));
1285         for (i = 0; i < count; i++) {
1286                 struct tgt_handler      *h;
1287                 struct dt_object        *dt_obj;
1288
1289                 update = (struct update *)((char *)ubuf + off);
1290                 if (old_batchid == -1) {
1291                         old_batchid = update->u_batchid;
1292                 } else if (old_batchid != update->u_batchid) {
1293                         /* Stop the current update transaction,
1294                          * create a new one */
1295                         rc = out_tx_end(env, ta);
1296                         if (rc != 0)
1297                                 RETURN(rc);
1298
1299                         rc = out_tx_start(env, dt, ta);
1300                         if (rc != 0)
1301                                 RETURN(rc);
1302                         old_batchid = update->u_batchid;
1303                 }
1304
1305                 fid_le_to_cpu(&update->u_fid, &update->u_fid);
1306                 if (!fid_is_sane(&update->u_fid)) {
1307                         CERROR("%s: invalid FID "DFID": rc = %d\n",
1308                                tgt_name(tsi->tsi_tgt), PFID(&update->u_fid),
1309                                -EPROTO);
1310                         GOTO(out, rc = err_serious(-EPROTO));
1311                 }
1312
1313                 dt_obj = dt_locate(env, dt, &update->u_fid);
1314                 if (IS_ERR(dt_obj))
1315                         GOTO(out, rc = PTR_ERR(dt_obj));
1316
1317                 tti->tti_u.update.tti_dt_object = dt_obj;
1318                 tti->tti_u.update.tti_update = update;
1319                 tti->tti_u.update.tti_update_reply_index = i;
1320
1321                 h = out_handler_find(update->u_type);
1322                 if (likely(h != NULL)) {
1323                         /* For real modification RPC, check if the update
1324                          * has been executed */
1325                         if (h->th_flags & MUTABOR) {
1326                                 struct ptlrpc_request *req = tgt_ses_req(tsi);
1327
1328                                 if (out_check_resent(env, dt, dt_obj, req,
1329                                                      out_reconstruct,
1330                                                      update_reply, i))
1331                                         GOTO(next, rc);
1332                         }
1333
1334                         rc = h->th_act(tsi);
1335                 } else {
1336                         CERROR("%s: The unsupported opc: 0x%x\n",
1337                                tgt_name(tsi->tsi_tgt), update->u_type);
1338                         lu_object_put(env, &dt_obj->do_lu);
1339                         GOTO(out, rc = -ENOTSUPP);
1340                 }
1341 next:
1342                 lu_object_put(env, &dt_obj->do_lu);
1343                 if (rc < 0)
1344                         GOTO(out, rc);
1345                 off += cfs_size_round(update_size(update));
1346         }
1347 out:
1348         rc1 = out_tx_end(env, ta);
1349         if (rc == 0)
1350                 rc = rc1;
1351         RETURN(rc);
1352 }
1353
1354 struct tgt_handler tgt_out_handlers[] = {
1355 TGT_UPDATE_HDL(MUTABOR, UPDATE_OBJ,     out_handle),
1356 };
1357 EXPORT_SYMBOL(tgt_out_handlers);
1358