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, 2014, 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 <obd_class.h>
  35 #include <md_object.h>
  36 #include "tgt_internal.h"
  37 #include <lustre_update.h>
  38
  39 static int tx_extend_args(struct thandle_exec_args *ta, int new_alloc_ta)
  40 {
  41         struct tx_arg   **new_ta;
  42         int             i;
  43         int             rc = 0;
  44
  45         if (ta->ta_alloc_args >= new_alloc_ta)
  46                 return 0;
  47
  48         OBD_ALLOC(new_ta, sizeof(*new_ta) * new_alloc_ta);
  49         if (new_ta == NULL)
  50                 return -ENOMEM;
  51
  52         for (i = 0; i < new_alloc_ta; i++) {
  53                 if (i < ta->ta_alloc_args) {
  54                         /* copy the old args to new one */
  55                         new_ta[i] = ta->ta_args[i];
  56                 } else {
  57                         OBD_ALLOC_PTR(new_ta[i]);
  58                         if (new_ta[i] == NULL)
  59                                 GOTO(out, rc = -ENOMEM);
  60                 }
  61         }
  62
  63         /* free the old args */
  64         if (ta->ta_args != NULL)
  65                 OBD_FREE(ta->ta_args, sizeof(ta->ta_args[0]) *
  66                                       ta->ta_alloc_args);
  67
  68         ta->ta_args = new_ta;
  69         ta->ta_alloc_args = new_alloc_ta;
  70 out:
  71         if (rc != 0) {
  72                 for (i = 0; i < new_alloc_ta; i++) {
  73                         if (new_ta[i] != NULL)
  74                                 OBD_FREE_PTR(new_ta[i]);
  75                 }
  76                 OBD_FREE(new_ta, sizeof(*new_ta) * new_alloc_ta);
  77         }
  78         return rc;
  79 }
  80
  81 #define TX_ALLOC_STEP   8
  82 static struct tx_arg *tx_add_exec(struct thandle_exec_args *ta,
  83                                   tx_exec_func_t func, tx_exec_func_t undo,
  84                                   const char *file, int line)
  85 {
  86         int rc;
  87         int i;
  88
  89         LASSERT(ta != NULL);
  90         LASSERT(func != NULL);
  91
  92         if (ta->ta_argno + 1 >= ta->ta_alloc_args) {
  93                 rc = tx_extend_args(ta, ta->ta_alloc_args + TX_ALLOC_STEP);
  94                 if (rc != 0)
  95                         return ERR_PTR(rc);
  96         }
  97
  98         i = ta->ta_argno;
  99
 100         ta->ta_argno++;
 101
 102         ta->ta_args[i]->exec_fn = func;
 103         ta->ta_args[i]->undo_fn = undo;
 104         ta->ta_args[i]->file    = file;
 105         ta->ta_args[i]->line    = line;
 106
 107         return ta->ta_args[i];
 108 }
 109
 110 static void out_reconstruct(const struct lu_env *env, struct dt_device *dt,
 111                             struct dt_object *obj,
 112                             struct object_update_reply *reply,
 113                             int index)
 114 {
 115         CDEBUG(D_INFO, "%s: fork reply reply %p index %d: rc = %d\n",
 116                dt_obd_name(dt), reply, index, 0);
 117
 118         object_update_result_insert(reply, NULL, 0, index, 0);
 119         return;
 120 }
 121
 122 typedef void (*out_reconstruct_t)(const struct lu_env *env,
 123                                   struct dt_device *dt,
 124                                   struct dt_object *obj,
 125                                   struct object_update_reply *reply,
 126                                   int index);
 127
 128 static inline int out_check_resent(const struct lu_env *env,
 129                                    struct dt_device *dt,
 130                                    struct dt_object *obj,
 131                                    struct ptlrpc_request *req,
 132                                    out_reconstruct_t reconstruct,
 133                                    struct object_update_reply *reply,
 134                                    int index)
 135 {
 136         if (likely(!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT)))
 137                 return 0;
 138
 139         if (req_xid_is_last(req)) {
 140                 reconstruct(env, dt, obj, reply, index);
 141                 return 1;
 142         }
 143         DEBUG_REQ(D_HA, req, "no reply for RESENT req (have "LPD64")",
 144                  req->rq_export->exp_target_data.ted_lcd->lcd_last_xid);
 145         return 0;
 146 }
 147
 148 static int out_obj_destroy(const struct lu_env *env, struct dt_object *dt_obj,
 149                            struct thandle *th)
 150 {
 151         int rc;
 152
 153         CDEBUG(D_INFO, "%s: destroy "DFID"\n", dt_obd_name(th->th_dev),
 154                PFID(lu_object_fid(&dt_obj->do_lu)));
 155
 156         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 157         rc = dt_destroy(env, dt_obj, th);
 158         dt_write_unlock(env, dt_obj);
 159
 160         return rc;
 161 }
 162
 163 /**
 164  * All of the xxx_undo will be used once execution failed,
 165  * But because all of the required resource has been reserved in
 166  * declare phase, i.e. if declare succeed, it should make sure
 167  * the following executing phase succeed in anyway, so these undo
 168  * should be useless for most of the time in Phase I
 169  */
 170 static int out_tx_create_undo(const struct lu_env *env, struct thandle *th,
 171                               struct tx_arg *arg)
 172 {
 173         int rc;
 174
 175         rc = out_obj_destroy(env, arg->object, th);
 176         if (rc != 0)
 177                 CERROR("%s: undo failure, we are doomed!: rc = %d\n",
 178                        dt_obd_name(th->th_dev), rc);
 179         return rc;
 180 }
 181
 182 static int out_tx_create_exec(const struct lu_env *env, struct thandle *th,
 183                               struct tx_arg *arg)
 184 {
 185         struct dt_object        *dt_obj = arg->object;
 186         int                      rc;
 187
 188         CDEBUG(D_OTHER, "%s: create "DFID": dof %u, mode %o\n",
 189                dt_obd_name(th->th_dev),
 190                PFID(lu_object_fid(&arg->object->do_lu)),
 191                arg->u.create.dof.dof_type,
 192                arg->u.create.attr.la_mode & S_IFMT);
 193
 194         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 195         rc = dt_create(env, dt_obj, &arg->u.create.attr,
 196                        &arg->u.create.hint, &arg->u.create.dof, th);
 197
 198         dt_write_unlock(env, dt_obj);
 199
 200         CDEBUG(D_INFO, "%s: insert create reply %p index %d: rc = %d\n",
 201                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
 202
 203         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
 204
 205         return rc;
 206 }
 207
 208 static int __out_tx_create(const struct lu_env *env, struct dt_object *obj,
 209                            struct lu_attr *attr, struct lu_fid *parent_fid,
 210                            struct dt_object_format *dof,
 211                            struct thandle_exec_args *ta,
 212                            struct object_update_reply *reply,
 213                            int index, const char *file, int line)
 214 {
 215         struct tx_arg *arg;
 216         int rc;
 217
 218         LASSERT(ta->ta_handle != NULL);
 219         rc = dt_declare_create(env, obj, attr, NULL, dof,
 220                                        ta->ta_handle);
 221         if (rc != 0)
 222                 return rc;
 223
 224         arg = tx_add_exec(ta, out_tx_create_exec, out_tx_create_undo, file,
 225                           line);
 226         if (IS_ERR(arg))
 227                 return PTR_ERR(arg);
 228
 229         /* release the object in out_trans_stop */
 230         lu_object_get(&obj->do_lu);
 231         arg->object = obj;
 232         arg->u.create.attr = *attr;
 233         if (parent_fid != NULL)
 234                 arg->u.create.fid = *parent_fid;
 235         memset(&arg->u.create.hint, 0, sizeof(arg->u.create.hint));
 236         arg->u.create.dof  = *dof;
 237         arg->reply = reply;
 238         arg->index = index;
 239
 240         return 0;
 241 }
 242
 243 static int out_create(struct tgt_session_info *tsi)
 244 {
 245         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
 246         struct object_update    *update = tti->tti_u.update.tti_update;
 247         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
 248         struct dt_object_format *dof = &tti->tti_u.update.tti_update_dof;
 249         struct obdo             *lobdo = &tti->tti_u.update.tti_obdo;
 250         struct lu_attr          *attr = &tti->tti_attr;
 251         struct lu_fid           *fid = NULL;
 252         struct obdo             *wobdo;
 253         size_t                  size;
 254         int                     rc;
 255
 256         ENTRY;
 257
 258         wobdo = object_update_param_get(update, 0, &size);
 259         if (wobdo == NULL || IS_ERR(wobdo) || size != sizeof(*wobdo)) {
 260                 CERROR("%s: obdo is NULL, invalid RPC: rc = %d\n",
 261                        tgt_name(tsi->tsi_tgt), -EPROTO);
 262                 RETURN(err_serious(-EPROTO));
 263         }
 264
 265         if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req))
 266                 lustre_swab_obdo(wobdo);
 267         lustre_get_wire_obdo(NULL, lobdo, wobdo);
 268         la_from_obdo(attr, lobdo, lobdo->o_valid);
 269
 270         dof->dof_type = dt_mode_to_dft(attr->la_mode);
 271         if (update->ou_params_count > 1) {
 272                 fid = object_update_param_get(update, 1, &size);
 273                 if (fid == NULL || IS_ERR(fid) || size != sizeof(*fid)) {
 274                         CERROR("%s: invalid fid: rc = %d\n",
 275                                tgt_name(tsi->tsi_tgt), -EPROTO);
 276                         RETURN(err_serious(-EPROTO));
 277                 }
 278                 if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req))
 279                         lustre_swab_lu_fid(fid);
 280                 if (!fid_is_sane(fid)) {
 281                         CERROR("%s: invalid fid "DFID": rc = %d\n",
 282                                tgt_name(tsi->tsi_tgt), PFID(fid), -EPROTO);
 283                         RETURN(err_serious(-EPROTO));
 284                 }
 285         }
 286
 287         if (lu_object_exists(&obj->do_lu))
 288                 RETURN(-EEXIST);
 289
 290         rc = out_tx_create(tsi->tsi_env, obj, attr, fid, dof,
 291                            &tti->tti_tea,
 292                            tti->tti_u.update.tti_update_reply,
 293                            tti->tti_u.update.tti_update_reply_index);
 294
 295         RETURN(rc);
 296 }
 297
 298 static int out_tx_attr_set_undo(const struct lu_env *env,
 299                                 struct thandle *th, struct tx_arg *arg)
 300 {
 301         CERROR("%s: attr set undo "DFID" unimplemented yet!: rc = %d\n",
 302                dt_obd_name(th->th_dev),
 303                PFID(lu_object_fid(&arg->object->do_lu)), -ENOTSUPP);
 304
 305         return -ENOTSUPP;
 306 }
 307
 308 static int out_tx_attr_set_exec(const struct lu_env *env, struct thandle *th,
 309                                 struct tx_arg *arg)
 310 {
 311         struct dt_object        *dt_obj = arg->object;
 312         int                     rc;
 313
 314         CDEBUG(D_OTHER, "%s: attr set "DFID"\n", dt_obd_name(th->th_dev),
 315                PFID(lu_object_fid(&dt_obj->do_lu)));
 316
 317         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 318         rc = dt_attr_set(env, dt_obj, &arg->u.attr_set.attr, th);
 319         dt_write_unlock(env, dt_obj);
 320
 321         CDEBUG(D_INFO, "%s: insert attr_set reply %p index %d: rc = %d\n",
 322                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
 323
 324         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
 325
 326         return rc;
 327 }
 328
 329 static int __out_tx_attr_set(const struct lu_env *env,
 330                              struct dt_object *dt_obj,
 331                              const struct lu_attr *attr,
 332                              struct thandle_exec_args *th,
 333                              struct object_update_reply *reply,
 334                              int index, const char *file, int line)
 335 {
 336         struct tx_arg   *arg;
 337         int             rc;
 338
 339         LASSERT(th->ta_handle != NULL);
 340         rc = dt_declare_attr_set(env, dt_obj, attr, th->ta_handle);
 341         if (rc != 0)
 342                 return rc;
 343
 344         arg = tx_add_exec(th, out_tx_attr_set_exec, out_tx_attr_set_undo,
 345                           file, line);
 346         if (IS_ERR(arg))
 347                 return PTR_ERR(arg);
 348
 349         lu_object_get(&dt_obj->do_lu);
 350         arg->object = dt_obj;
 351         arg->u.attr_set.attr = *attr;
 352         arg->reply = reply;
 353         arg->index = index;
 354         return 0;
 355 }
 356
 357 static int out_attr_set(struct tgt_session_info *tsi)
 358 {
 359         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
 360         struct object_update    *update = tti->tti_u.update.tti_update;
 361         struct lu_attr          *attr = &tti->tti_attr;
 362         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
 363         struct obdo             *lobdo = &tti->tti_u.update.tti_obdo;
 364         struct obdo             *wobdo;
 365         size_t                   size;
 366         int                      rc;
 367
 368         ENTRY;
 369
 370         wobdo = object_update_param_get(update, 0, &size);
 371         if (wobdo == NULL || IS_ERR(wobdo) || size != sizeof(*wobdo)) {
 372                 CERROR("%s: empty obdo in the update: rc = %d\n",
 373                        tgt_name(tsi->tsi_tgt), -EPROTO);
 374                 RETURN(err_serious(-EPROTO));
 375         }
 376
 377         attr->la_valid = 0;
 378         attr->la_valid = 0;
 379
 380         if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req))
 381                 lustre_swab_obdo(wobdo);
 382         lustre_get_wire_obdo(NULL, lobdo, wobdo);
 383         la_from_obdo(attr, lobdo, lobdo->o_valid);
 384
 385         rc = out_tx_attr_set(tsi->tsi_env, obj, attr, &tti->tti_tea,
 386                              tti->tti_u.update.tti_update_reply,
 387                              tti->tti_u.update.tti_update_reply_index);
 388
 389         RETURN(rc);
 390 }
 391
 392 static int out_attr_get(struct tgt_session_info *tsi)
 393 {
 394         const struct lu_env     *env = tsi->tsi_env;
 395         struct tgt_thread_info  *tti = tgt_th_info(env);
 396         struct obdo             *obdo = &tti->tti_u.update.tti_obdo;
 397         struct lu_attr          *la = &tti->tti_attr;
 398         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
 399         int                     idx = tti->tti_u.update.tti_update_reply_index;
 400         int                     rc;
 401
 402         ENTRY;
 403
 404         if (!lu_object_exists(&obj->do_lu)) {
 405                 /* Usually, this will be called when the master MDT try
 406                  * to init a remote object(see osp_object_init), so if
 407                  * the object does not exist on slave, we need set BANSHEE flag,
 408                  * so the object can be removed from the cache immediately */
 409                 set_bit(LU_OBJECT_HEARD_BANSHEE,
 410                         &obj->do_lu.lo_header->loh_flags);
 411                 RETURN(-ENOENT);
 412         }
 413
 414         dt_read_lock(env, obj, MOR_TGT_CHILD);
 415         rc = dt_attr_get(env, obj, la);
 416         if (rc)
 417                 GOTO(out_unlock, rc);
 418
 419         obdo->o_valid = 0;
 420         obdo_from_la(obdo, la, la->la_valid);
 421         lustre_set_wire_obdo(NULL, obdo, obdo);
 422
 423 out_unlock:
 424         dt_read_unlock(env, obj);
 425
 426         CDEBUG(D_INFO, "%s: insert attr get reply %p index %d: rc = %d\n",
 427                tgt_name(tsi->tsi_tgt), tti->tti_u.update.tti_update_reply,
 428                0, rc);
 429
 430         object_update_result_insert(tti->tti_u.update.tti_update_reply, obdo,
 431                                     sizeof(*obdo), idx, rc);
 432
 433         RETURN(rc);
 434 }
 435
 436 static int out_xattr_get(struct tgt_session_info *tsi)
 437 {
 438         const struct lu_env        *env = tsi->tsi_env;
 439         struct tgt_thread_info     *tti = tgt_th_info(env);
 440         struct object_update       *update = tti->tti_u.update.tti_update;
 441         struct lu_buf              *lbuf = &tti->tti_buf;
 442         struct object_update_reply *reply = tti->tti_u.update.tti_update_reply;
 443         struct dt_object           *obj = tti->tti_u.update.tti_dt_object;
 444         char                       *name;
 445         struct object_update_result *update_result;
 446         int                     idx = tti->tti_u.update.tti_update_reply_index;
 447         int                        rc;
 448
 449         ENTRY;
 450
 451         if (!lu_object_exists(&obj->do_lu)) {
 452                 set_bit(LU_OBJECT_HEARD_BANSHEE,
 453                         &obj->do_lu.lo_header->loh_flags);
 454                 RETURN(-ENOENT);
 455         }
 456
 457         name = object_update_param_get(update, 0, NULL);
 458         if (name == NULL || IS_ERR(name)) {
 459                 CERROR("%s: empty name for xattr get: rc = %d\n",
 460                        tgt_name(tsi->tsi_tgt), -EPROTO);
 461                 RETURN(err_serious(-EPROTO));
 462         }
 463
 464         update_result = object_update_result_get(reply, 0, NULL);
 465         if (update_result == NULL) {
 466                 CERROR("%s: empty name for xattr get: rc = %d\n",
 467                        tgt_name(tsi->tsi_tgt), -EPROTO);
 468                 RETURN(err_serious(-EPROTO));
 469         }
 470
 471         lbuf->lb_buf = update_result->our_data;
 472         lbuf->lb_len = OUT_UPDATE_REPLY_SIZE -
 473                        cfs_size_round((unsigned long)update_result->our_data -
 474                                       (unsigned long)update_result);
 475         dt_read_lock(env, obj, MOR_TGT_CHILD);
 476         rc = dt_xattr_get(env, obj, lbuf, name);
 477         dt_read_unlock(env, obj);
 478         if (rc < 0) {
 479                 lbuf->lb_len = 0;
 480                 GOTO(out, rc);
 481         }
 482         lbuf->lb_len = rc;
 483         rc = 0;
 484         CDEBUG(D_INFO, "%s: "DFID" get xattr %s len %d\n",
 485                tgt_name(tsi->tsi_tgt), PFID(lu_object_fid(&obj->do_lu)),
 486                name, (int)lbuf->lb_len);
 487
 488         GOTO(out, rc);
 489
 490 out:
 491         object_update_result_insert(reply, lbuf->lb_buf, lbuf->lb_len, idx, rc);
 492         RETURN(rc);
 493 }
 494
 495 static int out_index_lookup(struct tgt_session_info *tsi)
 496 {
 497         const struct lu_env     *env = tsi->tsi_env;
 498         struct tgt_thread_info  *tti = tgt_th_info(env);
 499         struct object_update    *update = tti->tti_u.update.tti_update;
 500         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
 501         char                    *name;
 502         int                      rc;
 503
 504         ENTRY;
 505
 506         if (!lu_object_exists(&obj->do_lu))
 507                 RETURN(-ENOENT);
 508
 509         name = object_update_param_get(update, 0, NULL);
 510         if (name == NULL || IS_ERR(name)) {
 511                 CERROR("%s: empty name for lookup: rc = %d\n",
 512                        tgt_name(tsi->tsi_tgt), -EPROTO);
 513                 RETURN(err_serious(-EPROTO));
 514         }
 515
 516         dt_read_lock(env, obj, MOR_TGT_CHILD);
 517         if (!dt_try_as_dir(env, obj))
 518                 GOTO(out_unlock, rc = -ENOTDIR);
 519
 520         rc = dt_lookup(env, obj, (struct dt_rec *)&tti->tti_fid1,
 521                        (struct dt_key *)name);
 522
 523         if (rc < 0)
 524                 GOTO(out_unlock, rc);
 525
 526         if (rc == 0)
 527                 rc += 1;
 528
 529 out_unlock:
 530         dt_read_unlock(env, obj);
 531
 532         CDEBUG(D_INFO, "lookup "DFID" %s get "DFID" rc %d\n",
 533                PFID(lu_object_fid(&obj->do_lu)), name,
 534                PFID(&tti->tti_fid1), rc);
 535
 536         CDEBUG(D_INFO, "%s: insert lookup reply %p index %d: rc = %d\n",
 537                tgt_name(tsi->tsi_tgt), tti->tti_u.update.tti_update_reply,
 538                0, rc);
 539
 540         object_update_result_insert(tti->tti_u.update.tti_update_reply,
 541                             &tti->tti_fid1, sizeof(tti->tti_fid1),
 542                             tti->tti_u.update.tti_update_reply_index, rc);
 543         RETURN(rc);
 544 }
 545
 546 static int out_tx_xattr_set_exec(const struct lu_env *env,
 547                                  struct thandle *th,
 548                                  struct tx_arg *arg)
 549 {
 550         struct dt_object *dt_obj = arg->object;
 551         int rc;
 552
 553         CDEBUG(D_INFO, "%s: set xattr buf %p name %s flag %d\n",
 554                dt_obd_name(th->th_dev), arg->u.xattr_set.buf.lb_buf,
 555                arg->u.xattr_set.name, arg->u.xattr_set.flags);
 556
 557         if (!lu_object_exists(&dt_obj->do_lu))
 558                 GOTO(out, rc = -ENOENT);
 559
 560         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 561         rc = dt_xattr_set(env, dt_obj, &arg->u.xattr_set.buf,
 562                           arg->u.xattr_set.name, arg->u.xattr_set.flags, th);
 563         /**
 564          * Ignore errors if this is LINK EA
 565          **/
 566         if (unlikely(rc != 0 &&
 567                      strcmp(arg->u.xattr_set.name, XATTR_NAME_LINK) == 0)) {
 568                 /* XXX: If the linkEA is overflow, then we need to notify the
 569                  *      namespace LFSCK to skip "nlink" attribute verification
 570                  *      on this object to avoid the "nlink" to be shrinked by
 571                  *      wrong. It may be not good an interaction with LFSCK
 572                  *      like this. We will consider to replace it with other
 573                  *      mechanism in future. LU-5802. */
 574                 if (rc == -ENOSPC) {
 575                         struct lfsck_request *lr = &tgt_th_info(env)->tti_lr;
 576
 577                         lfsck_pack_rfa(lr, lu_object_fid(&dt_obj->do_lu),
 578                                        LE_SKIP_NLINK, LFSCK_TYPE_NAMESPACE);
 579                         tgt_lfsck_in_notify(env,
 580                                 tgt_ses_info(env)->tsi_tgt->lut_bottom, lr, th);
 581                 }
 582
 583                 rc = 0;
 584         }
 585         dt_write_unlock(env, dt_obj);
 586
 587 out:
 588         CDEBUG(D_INFO, "%s: insert xattr set reply %p index %d: rc = %d\n",
 589                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
 590
 591         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
 592
 593         return rc;
 594 }
 595
 596 static int __out_tx_xattr_set(const struct lu_env *env,
 597                               struct dt_object *dt_obj,
 598                               const struct lu_buf *buf,
 599                               const char *name, int flags,
 600                               struct thandle_exec_args *ta,
 601                               struct object_update_reply *reply,
 602                               int index, const char *file, int line)
 603 {
 604         struct tx_arg   *arg;
 605         int             rc;
 606
 607         LASSERT(ta->ta_handle != NULL);
 608         rc = dt_declare_xattr_set(env, dt_obj, buf, name, flags, ta->ta_handle);
 609         if (rc != 0)
 610                 return rc;
 611
 612         if (strcmp(name, XATTR_NAME_LINK) == 0) {
 613                 struct lfsck_request *lr = &tgt_th_info(env)->tti_lr;
 614
 615                 /* XXX: If the linkEA is overflow, then we need to notify the
 616                  *      namespace LFSCK to skip "nlink" attribute verification
 617                  *      on this object to avoid the "nlink" to be shrinked by
 618                  *      wrong. It may be not good an interaction with LFSCK
 619                  *      like this. We will consider to replace it with other
 620                  *      mechanism in future. LU-5802. */
 621                 lfsck_pack_rfa(lr, lu_object_fid(&dt_obj->do_lu),
 622                                LE_SKIP_NLINK_DECLARE, LFSCK_TYPE_NAMESPACE);
 623                 rc = tgt_lfsck_in_notify(env,
 624                                          tgt_ses_info(env)->tsi_tgt->lut_bottom,
 625                                          lr, ta->ta_handle);
 626                 if (rc != 0)
 627                         return rc;
 628         }
 629
 630         arg = tx_add_exec(ta, out_tx_xattr_set_exec, NULL, file, line);
 631         if (IS_ERR(arg))
 632                 return PTR_ERR(arg);
 633
 634         lu_object_get(&dt_obj->do_lu);
 635         arg->object = dt_obj;
 636         arg->u.xattr_set.name = name;
 637         arg->u.xattr_set.flags = flags;
 638         arg->u.xattr_set.buf = *buf;
 639         arg->reply = reply;
 640         arg->index = index;
 641         arg->u.xattr_set.csum = 0;
 642         return 0;
 643 }
 644
 645 static int out_xattr_set(struct tgt_session_info *tsi)
 646 {
 647         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
 648         struct object_update    *update = tti->tti_u.update.tti_update;
 649         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
 650         struct lu_buf           *lbuf = &tti->tti_buf;
 651         char                    *name;
 652         char                    *buf;
 653         __u32                   *tmp;
 654         size_t                   buf_len = 0;
 655         int                      flag;
 656         size_t                   size = 0;
 657         int                      rc;
 658         ENTRY;
 659
 660         name = object_update_param_get(update, 0, NULL);
 661         if (name == NULL || IS_ERR(name)) {
 662                 CERROR("%s: empty name for xattr set: rc = %d\n",
 663                        tgt_name(tsi->tsi_tgt), -EPROTO);
 664                 RETURN(err_serious(-EPROTO));
 665         }
 666
 667         buf = object_update_param_get(update, 1, &buf_len);
 668         if (IS_ERR(buf))
 669                 RETURN(err_serious(-EPROTO));
 670
 671         lbuf->lb_buf = buf;
 672         lbuf->lb_len = buf_len;
 673
 674         tmp = object_update_param_get(update, 2, &size);
 675         if (tmp == NULL || size != sizeof(*tmp)) {
 676                 CERROR("%s: emptry or wrong size %zu flag: rc = %d\n",
 677                        tgt_name(tsi->tsi_tgt), size, -EPROTO);
 678                 RETURN(err_serious(-EPROTO));
 679         }
 680
 681         if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req))
 682                 __swab32s(tmp);
 683         flag = *tmp;
 684
 685         rc = out_tx_xattr_set(tsi->tsi_env, obj, lbuf, name, flag,
 686                               &tti->tti_tea,
 687                               tti->tti_u.update.tti_update_reply,
 688                               tti->tti_u.update.tti_update_reply_index);
 689         RETURN(rc);
 690 }
 691
 692 static int out_tx_xattr_del_exec(const struct lu_env *env, struct thandle *th,
 693                                  struct tx_arg *arg)
 694 {
 695         struct dt_object *dt_obj = arg->object;
 696         int rc;
 697
 698         CDEBUG(D_INFO, "%s: del xattr name '%s' on "DFID"\n",
 699                dt_obd_name(th->th_dev), arg->u.xattr_set.name,
 700                PFID(lu_object_fid(&dt_obj->do_lu)));
 701
 702         if (!lu_object_exists(&dt_obj->do_lu))
 703                 GOTO(out, rc = -ENOENT);
 704
 705         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 706         rc = dt_xattr_del(env, dt_obj, arg->u.xattr_set.name, th);
 707         dt_write_unlock(env, dt_obj);
 708 out:
 709         CDEBUG(D_INFO, "%s: insert xattr del reply %p index %d: rc = %d\n",
 710                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
 711
 712         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
 713
 714         return rc;
 715 }
 716
 717 static int __out_tx_xattr_del(const struct lu_env *env,
 718                               struct dt_object *dt_obj, const char *name,
 719                               struct thandle_exec_args *ta,
 720                               struct object_update_reply *reply,
 721                               int index, const char *file, int line)
 722 {
 723         struct tx_arg   *arg;
 724         int             rc;
 725
 726         rc = dt_declare_xattr_del(env, dt_obj, name, ta->ta_handle);
 727         if (rc != 0)
 728                 return rc;
 729
 730         arg = tx_add_exec(ta, out_tx_xattr_del_exec, NULL, file, line);
 731         if (IS_ERR(arg))
 732                 return PTR_ERR(arg);
 733
 734         lu_object_get(&dt_obj->do_lu);
 735         arg->object = dt_obj;
 736         arg->u.xattr_set.name = name;
 737         arg->reply = reply;
 738         arg->index = index;
 739         return 0;
 740 }
 741
 742 static int out_xattr_del(struct tgt_session_info *tsi)
 743 {
 744         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
 745         struct object_update    *update = tti->tti_u.update.tti_update;
 746         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
 747         char                    *name;
 748         int                      rc;
 749         ENTRY;
 750
 751         name = object_update_param_get(update, 0, NULL);
 752         if (name == NULL || IS_ERR(name)) {
 753                 CERROR("%s: empty name for xattr set: rc = %d\n",
 754                        tgt_name(tsi->tsi_tgt), -EPROTO);
 755                 RETURN(err_serious(-EPROTO));
 756         }
 757
 758         rc = out_tx_xattr_del(tsi->tsi_env, obj, name, &tti->tti_tea,
 759                               tti->tti_u.update.tti_update_reply,
 760                               tti->tti_u.update.tti_update_reply_index);
 761         RETURN(rc);
 762 }
 763
 764 static int out_obj_ref_add(const struct lu_env *env,
 765                            struct dt_object *dt_obj,
 766                            struct thandle *th)
 767 {
 768         int rc;
 769
 770         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 771         rc = dt_ref_add(env, dt_obj, th);
 772         dt_write_unlock(env, dt_obj);
 773
 774         return rc;
 775 }
 776
 777 static int out_obj_ref_del(const struct lu_env *env,
 778                            struct dt_object *dt_obj,
 779                            struct thandle *th)
 780 {
 781         int rc;
 782
 783         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 784         rc = dt_ref_del(env, dt_obj, th);
 785         dt_write_unlock(env, dt_obj);
 786
 787         return rc;
 788 }
 789
 790 static int out_tx_ref_add_exec(const struct lu_env *env, struct thandle *th,
 791                                struct tx_arg *arg)
 792 {
 793         struct dt_object *dt_obj = arg->object;
 794         int rc;
 795
 796         rc = out_obj_ref_add(env, dt_obj, th);
 797
 798         CDEBUG(D_INFO, "%s: insert ref_add reply %p index %d: rc = %d\n",
 799                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
 800
 801         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
 802         return rc;
 803 }
 804
 805 static int out_tx_ref_add_undo(const struct lu_env *env, struct thandle *th,
 806                                struct tx_arg *arg)
 807 {
 808         return out_obj_ref_del(env, arg->object, th);
 809 }
 810
 811 static int __out_tx_ref_add(const struct lu_env *env,
 812                             struct dt_object *dt_obj,
 813                             struct thandle_exec_args *ta,
 814                             struct object_update_reply *reply,
 815                             int index, const char *file, int line)
 816 {
 817         struct tx_arg   *arg;
 818         int             rc;
 819
 820         LASSERT(ta->ta_handle != NULL);
 821         rc = dt_declare_ref_add(env, dt_obj, ta->ta_handle);
 822         if (rc != 0)
 823                 return rc;
 824
 825         arg = tx_add_exec(ta, out_tx_ref_add_exec, out_tx_ref_add_undo, file,
 826                           line);
 827         if (IS_ERR(arg))
 828                 return PTR_ERR(arg);
 829
 830         lu_object_get(&dt_obj->do_lu);
 831         arg->object = dt_obj;
 832         arg->reply = reply;
 833         arg->index = index;
 834         return 0;
 835 }
 836
 837 /**
 838  * increase ref of the object
 839  **/
 840 static int out_ref_add(struct tgt_session_info *tsi)
 841 {
 842         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
 843         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
 844         int                      rc;
 845
 846         ENTRY;
 847
 848         rc = out_tx_ref_add(tsi->tsi_env, obj, &tti->tti_tea,
 849                             tti->tti_u.update.tti_update_reply,
 850                             tti->tti_u.update.tti_update_reply_index);
 851         RETURN(rc);
 852 }
 853
 854 static int out_tx_ref_del_exec(const struct lu_env *env, struct thandle *th,
 855                                struct tx_arg *arg)
 856 {
 857         struct dt_object        *dt_obj = arg->object;
 858         int                      rc;
 859
 860         rc = out_obj_ref_del(env, dt_obj, th);
 861
 862         CDEBUG(D_INFO, "%s: insert ref_del reply %p index %d: rc = %d\n",
 863                dt_obd_name(th->th_dev), arg->reply, arg->index, 0);
 864
 865         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
 866
 867         return rc;
 868 }
 869
 870 static int out_tx_ref_del_undo(const struct lu_env *env, struct thandle *th,
 871                                struct tx_arg *arg)
 872 {
 873         return out_obj_ref_add(env, arg->object, th);
 874 }
 875
 876 static int __out_tx_ref_del(const struct lu_env *env,
 877                             struct dt_object *dt_obj,
 878                             struct thandle_exec_args *ta,
 879                             struct object_update_reply *reply,
 880                             int index, const char *file, int line)
 881 {
 882         struct tx_arg   *arg;
 883         int             rc;
 884
 885         LASSERT(ta->ta_handle != NULL);
 886         rc = dt_declare_ref_del(env, dt_obj, ta->ta_handle);
 887         if (rc != 0)
 888                 return rc;
 889
 890         arg = tx_add_exec(ta, out_tx_ref_del_exec, out_tx_ref_del_undo, file,
 891                           line);
 892         if (IS_ERR(arg))
 893                 return PTR_ERR(arg);
 894
 895         lu_object_get(&dt_obj->do_lu);
 896         arg->object = dt_obj;
 897         arg->reply = reply;
 898         arg->index = index;
 899         return 0;
 900 }
 901
 902 static int out_ref_del(struct tgt_session_info *tsi)
 903 {
 904         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
 905         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
 906         int                      rc;
 907
 908         ENTRY;
 909
 910         if (!lu_object_exists(&obj->do_lu))
 911                 RETURN(-ENOENT);
 912
 913         rc = out_tx_ref_del(tsi->tsi_env, obj, &tti->tti_tea,
 914                             tti->tti_u.update.tti_update_reply,
 915                             tti->tti_u.update.tti_update_reply_index);
 916         RETURN(rc);
 917 }
 918
 919 static int out_obj_index_insert(const struct lu_env *env,
 920                                 struct dt_object *dt_obj,
 921                                 const struct dt_rec *rec,
 922                                 const struct dt_key *key,
 923                                 struct thandle *th)
 924 {
 925         int rc;
 926
 927         CDEBUG(D_INFO, "%s: index insert "DFID" name: %s fid "DFID", type %u\n",
 928                dt_obd_name(th->th_dev), PFID(lu_object_fid(&dt_obj->do_lu)),
 929                (char *)key, PFID(((struct dt_insert_rec *)rec)->rec_fid),
 930                ((struct dt_insert_rec *)rec)->rec_type);
 931
 932         if (dt_try_as_dir(env, dt_obj) == 0)
 933                 return -ENOTDIR;
 934
 935         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 936         rc = dt_insert(env, dt_obj, rec, key, th, 0);
 937         dt_write_unlock(env, dt_obj);
 938
 939         return rc;
 940 }
 941
 942 static int out_obj_index_delete(const struct lu_env *env,
 943                                 struct dt_object *dt_obj,
 944                                 const struct dt_key *key,
 945                                 struct thandle *th)
 946 {
 947         int rc;
 948
 949         CDEBUG(D_INFO, "%s: index delete "DFID" name: %s\n",
 950                dt_obd_name(th->th_dev), PFID(lu_object_fid(&dt_obj->do_lu)),
 951                (char *)key);
 952
 953         if (dt_try_as_dir(env, dt_obj) == 0)
 954                 return -ENOTDIR;
 955
 956         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 957         rc = dt_delete(env, dt_obj, key, th);
 958         dt_write_unlock(env, dt_obj);
 959
 960         return rc;
 961 }
 962
 963 static int out_tx_index_insert_exec(const struct lu_env *env,
 964                                     struct thandle *th, struct tx_arg *arg)
 965 {
 966         struct dt_object *dt_obj = arg->object;
 967         int rc;
 968
 969         if (unlikely(!dt_object_exists(dt_obj)))
 970                 RETURN(-ESTALE);
 971
 972         rc = out_obj_index_insert(env, dt_obj,
 973                                   (const struct dt_rec *)&arg->u.insert.rec,
 974                                   arg->u.insert.key, th);
 975
 976         CDEBUG(D_INFO, "%s: insert idx insert reply %p index %d: rc = %d\n",
 977                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
 978
 979         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
 980
 981         return rc;
 982 }
 983
 984 static int out_tx_index_insert_undo(const struct lu_env *env,
 985                                     struct thandle *th, struct tx_arg *arg)
 986 {
 987         return out_obj_index_delete(env, arg->object, arg->u.insert.key, th);
 988 }
 989
 990 static int __out_tx_index_insert(const struct lu_env *env,
 991                                  struct dt_object *dt_obj,
 992                                  const struct dt_rec *rec,
 993                                  const struct dt_key *key,
 994                                  struct thandle_exec_args *ta,
 995                                  struct object_update_reply *reply,
 996                                  int index, const char *file, int line)
 997 {
 998         struct tx_arg   *arg;
 999         int             rc;
1000
1001         LASSERT(ta->ta_handle != NULL);
1002         if (dt_try_as_dir(env, dt_obj) == 0) {
1003                 rc = -ENOTDIR;
1004                 return rc;
1005         }
1006
1007         rc = dt_declare_insert(env, dt_obj, rec, key, ta->ta_handle);
1008         if (rc != 0)
1009                 return rc;
1010
1011         arg = tx_add_exec(ta, out_tx_index_insert_exec,
1012                           out_tx_index_insert_undo, file, line);
1013         if (IS_ERR(arg))
1014                 return PTR_ERR(arg);
1015
1016         lu_object_get(&dt_obj->do_lu);
1017         arg->object = dt_obj;
1018         arg->reply = reply;
1019         arg->index = index;
1020         arg->u.insert.rec = *(const struct dt_insert_rec *)rec;
1021         arg->u.insert.key = key;
1022
1023         return 0;
1024 }
1025
1026 static int out_index_insert(struct tgt_session_info *tsi)
1027 {
1028         struct tgt_thread_info  *tti    = tgt_th_info(tsi->tsi_env);
1029         struct object_update    *update = tti->tti_u.update.tti_update;
1030         struct dt_object        *obj    = tti->tti_u.update.tti_dt_object;
1031         struct dt_insert_rec    *rec    = &tti->tti_rec;
1032         struct lu_fid           *fid;
1033         char                    *name;
1034         __u32                   *ptype;
1035         int                      rc     = 0;
1036         size_t                   size;
1037         ENTRY;
1038
1039         name = object_update_param_get(update, 0, NULL);
1040         if (name == NULL || IS_ERR(name)) {
1041                 CERROR("%s: empty name for index insert: rc = %d\n",
1042                        tgt_name(tsi->tsi_tgt), -EPROTO);
1043                 RETURN(err_serious(-EPROTO));
1044         }
1045
1046         fid = object_update_param_get(update, 1, &size);
1047         if (fid == NULL || IS_ERR(fid) || size != sizeof(*fid)) {
1048                 CERROR("%s: invalid fid: rc = %d\n",
1049                        tgt_name(tsi->tsi_tgt), -EPROTO);
1050                 RETURN(err_serious(-EPROTO));
1051         }
1052
1053         if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req))
1054                 lustre_swab_lu_fid(fid);
1055
1056         if (!fid_is_sane(fid)) {
1057                 CERROR("%s: invalid FID "DFID": rc = %d\n",
1058                        tgt_name(tsi->tsi_tgt), PFID(fid), -EPROTO);
1059                 RETURN(err_serious(-EPROTO));
1060         }
1061
1062         ptype = object_update_param_get(update, 2, &size);
1063         if (ptype == NULL || IS_ERR(ptype) || size != sizeof(*ptype)) {
1064                 CERROR("%s: invalid type for index insert: rc = %d\n",
1065                        tgt_name(tsi->tsi_tgt), -EPROTO);
1066                 RETURN(err_serious(-EPROTO));
1067         }
1068
1069         if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req))
1070                 __swab32s(ptype);
1071
1072         rec->rec_fid = fid;
1073         rec->rec_type = *ptype;
1074
1075         rc = out_tx_index_insert(tsi->tsi_env, obj, (const struct dt_rec *)rec,
1076                                  (const struct dt_key *)name, &tti->tti_tea,
1077                                  tti->tti_u.update.tti_update_reply,
1078                                  tti->tti_u.update.tti_update_reply_index);
1079         RETURN(rc);
1080 }
1081
1082 static int out_tx_index_delete_exec(const struct lu_env *env,
1083                                     struct thandle *th,
1084                                     struct tx_arg *arg)
1085 {
1086         int rc;
1087
1088         rc = out_obj_index_delete(env, arg->object, arg->u.insert.key, th);
1089
1090         CDEBUG(D_INFO, "%s: delete idx insert reply %p index %d: rc = %d\n",
1091                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
1092
1093         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
1094
1095         return rc;
1096 }
1097
1098 static int out_tx_index_delete_undo(const struct lu_env *env,
1099                                     struct thandle *th,
1100                                     struct tx_arg *arg)
1101 {
1102         CERROR("%s: Oops, can not rollback index_delete yet: rc = %d\n",
1103                dt_obd_name(th->th_dev), -ENOTSUPP);
1104         return -ENOTSUPP;
1105 }
1106
1107 static int __out_tx_index_delete(const struct lu_env *env,
1108                                  struct dt_object *dt_obj,
1109                                  const struct dt_key *key,
1110                                  struct thandle_exec_args *ta,
1111                                  struct object_update_reply *reply,
1112                                  int index, const char *file, int line)
1113 {
1114         struct tx_arg   *arg;
1115         int             rc;
1116
1117         if (dt_try_as_dir(env, dt_obj) == 0) {
1118                 rc = -ENOTDIR;
1119                 return rc;
1120         }
1121
1122         LASSERT(ta->ta_handle != NULL);
1123         rc = dt_declare_delete(env, dt_obj, key, ta->ta_handle);
1124         if (rc != 0)
1125                 return rc;
1126
1127         arg = tx_add_exec(ta, out_tx_index_delete_exec,
1128                           out_tx_index_delete_undo, file, line);
1129         if (IS_ERR(arg))
1130                 return PTR_ERR(arg);
1131
1132         lu_object_get(&dt_obj->do_lu);
1133         arg->object = dt_obj;
1134         arg->reply = reply;
1135         arg->index = index;
1136         arg->u.insert.key = key;
1137         return 0;
1138 }
1139
1140 static int out_index_delete(struct tgt_session_info *tsi)
1141 {
1142         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
1143         struct object_update    *update = tti->tti_u.update.tti_update;
1144         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
1145         char                    *name;
1146         int                      rc = 0;
1147
1148         if (!lu_object_exists(&obj->do_lu))
1149                 RETURN(-ENOENT);
1150
1151         name = object_update_param_get(update, 0, NULL);
1152         if (name == NULL || IS_ERR(name)) {
1153                 CERROR("%s: empty name for index delete: rc = %d\n",
1154                        tgt_name(tsi->tsi_tgt), -EPROTO);
1155                 RETURN(err_serious(-EPROTO));
1156         }
1157
1158         rc = out_tx_index_delete(tsi->tsi_env, obj, (const struct dt_key *)name,
1159                                  &tti->tti_tea,
1160                                  tti->tti_u.update.tti_update_reply,
1161                                  tti->tti_u.update.tti_update_reply_index);
1162         RETURN(rc);
1163 }
1164
1165 static int out_tx_destroy_exec(const struct lu_env *env, struct thandle *th,
1166                                struct tx_arg *arg)
1167 {
1168         struct dt_object *dt_obj = arg->object;
1169         int rc;
1170
1171         rc = out_obj_destroy(env, dt_obj, th);
1172
1173         CDEBUG(D_INFO, "%s: insert destroy reply %p index %d: rc = %d\n",
1174                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
1175
1176         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
1177
1178         RETURN(rc);
1179 }
1180
1181 static int out_tx_destroy_undo(const struct lu_env *env, struct thandle *th,
1182                                struct tx_arg *arg)
1183 {
1184         CERROR("%s: not support destroy undo yet!: rc = %d\n",
1185                dt_obd_name(th->th_dev), -ENOTSUPP);
1186         return -ENOTSUPP;
1187 }
1188
1189 static int __out_tx_destroy(const struct lu_env *env, struct dt_object *dt_obj,
1190                              struct thandle_exec_args *ta,
1191                              struct object_update_reply *reply,
1192                              int index, const char *file, int line)
1193 {
1194         struct tx_arg   *arg;
1195         int             rc;
1196
1197         LASSERT(ta->ta_handle != NULL);
1198         rc = dt_declare_destroy(env, dt_obj, ta->ta_handle);
1199         if (rc != 0)
1200                 return rc;
1201
1202         arg = tx_add_exec(ta, out_tx_destroy_exec, out_tx_destroy_undo,
1203                           file, line);
1204         if (IS_ERR(arg))
1205                 return PTR_ERR(arg);
1206
1207         lu_object_get(&dt_obj->do_lu);
1208         arg->object = dt_obj;
1209         arg->reply = reply;
1210         arg->index = index;
1211         return 0;
1212 }
1213
1214 static int out_destroy(struct tgt_session_info *tsi)
1215 {
1216         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
1217         struct object_update    *update = tti->tti_u.update.tti_update;
1218         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
1219         struct lu_fid           *fid;
1220         int                      rc;
1221         ENTRY;
1222
1223         fid = &update->ou_fid;
1224         if (!fid_is_sane(fid)) {
1225                 CERROR("%s: invalid FID "DFID": rc = %d\n",
1226                        tgt_name(tsi->tsi_tgt), PFID(fid), -EPROTO);
1227                 RETURN(err_serious(-EPROTO));
1228         }
1229
1230         if (!lu_object_exists(&obj->do_lu))
1231                 RETURN(-ENOENT);
1232
1233         rc = out_tx_destroy(tsi->tsi_env, obj, &tti->tti_tea,
1234                             tti->tti_u.update.tti_update_reply,
1235                             tti->tti_u.update.tti_update_reply_index);
1236
1237         RETURN(rc);
1238 }
1239
1240 static int out_tx_write_exec(const struct lu_env *env, struct thandle *th,
1241                              struct tx_arg *arg)
1242 {
1243         struct dt_object *dt_obj = arg->object;
1244         int rc;
1245
1246         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
1247         rc = dt_record_write(env, dt_obj, &arg->u.write.buf,
1248                              &arg->u.write.pos, th);
1249         dt_write_unlock(env, dt_obj);
1250
1251         if (rc == 0)
1252                 rc = arg->u.write.buf.lb_len;
1253
1254         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
1255
1256         return rc > 0 ? 0 : rc;
1257 }
1258
1259 static int __out_tx_write(const struct lu_env *env,
1260                           struct dt_object *dt_obj,
1261                           const struct lu_buf *buf,
1262                           loff_t pos, struct thandle_exec_args *ta,
1263                           struct object_update_reply *reply,
1264                           int index, const char *file, int line)
1265 {
1266         struct tx_arg   *arg;
1267         int             rc;
1268
1269         LASSERT(ta->ta_handle != NULL);
1270         rc = dt_declare_record_write(env, dt_obj, buf, pos, ta->ta_handle);
1271         if (rc != 0)
1272                 return rc;
1273
1274         arg = tx_add_exec(ta, out_tx_write_exec, NULL, file, line);
1275         if (IS_ERR(arg))
1276                 return PTR_ERR(arg);
1277
1278         lu_object_get(&dt_obj->do_lu);
1279         arg->object = dt_obj;
1280         arg->u.write.buf = *buf;
1281         arg->u.write.pos = pos;
1282         arg->reply = reply;
1283         arg->index = index;
1284         return 0;
1285 }
1286
1287 static int out_write(struct tgt_session_info *tsi)
1288 {
1289         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
1290         struct object_update    *update = tti->tti_u.update.tti_update;
1291         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
1292         struct lu_buf           *lbuf = &tti->tti_buf;
1293         char                    *buf;
1294         __u64                   *tmp;
1295         size_t                  size = 0;
1296         size_t                  buf_len = 0;
1297         loff_t                  pos;
1298         int                      rc;
1299         ENTRY;
1300
1301         buf = object_update_param_get(update, 0, &buf_len);
1302         if (buf == NULL || IS_ERR(buf) || buf_len == 0) {
1303                 CERROR("%s: empty buf for xattr set: rc = %d\n",
1304                        tgt_name(tsi->tsi_tgt), -EPROTO);
1305                 RETURN(err_serious(-EPROTO));
1306         }
1307         lbuf->lb_buf = buf;
1308         lbuf->lb_len = buf_len;
1309
1310         tmp = object_update_param_get(update, 1, &size);
1311         if (tmp == NULL || IS_ERR(tmp) || size != sizeof(*tmp)) {
1312                 CERROR("%s: empty or wrong size %zu pos: rc = %d\n",
1313                        tgt_name(tsi->tsi_tgt), size, -EPROTO);
1314                 RETURN(err_serious(-EPROTO));
1315         }
1316
1317         if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req))
1318                 __swab64s(tmp);
1319         pos = *tmp;
1320
1321         rc = out_tx_write(tsi->tsi_env, obj, lbuf, pos,
1322                           &tti->tti_tea,
1323                           tti->tti_u.update.tti_update_reply,
1324                           tti->tti_u.update.tti_update_reply_index);
1325         RETURN(rc);
1326 }
1327
1328 #define DEF_OUT_HNDL(opc, name, flags, fn)     \
1329 [opc - OUT_CREATE] = {                                  \
1330         .th_name    = name,                             \
1331         .th_fail_id = 0,                                \
1332         .th_opc     = opc,                              \
1333         .th_flags   = flags,                            \
1334         .th_act     = fn,                               \
1335         .th_fmt     = NULL,                             \
1336         .th_version = 0,                                \
1337 }
1338
1339 static struct tgt_handler out_update_ops[] = {
1340         DEF_OUT_HNDL(OUT_CREATE, "out_create", MUTABOR | HABEO_REFERO,
1341                      out_create),
1342         DEF_OUT_HNDL(OUT_DESTROY, "out_create", MUTABOR | HABEO_REFERO,
1343                      out_destroy),
1344         DEF_OUT_HNDL(OUT_REF_ADD, "out_ref_add", MUTABOR | HABEO_REFERO,
1345                      out_ref_add),
1346         DEF_OUT_HNDL(OUT_REF_DEL, "out_ref_del", MUTABOR | HABEO_REFERO,
1347                      out_ref_del),
1348         DEF_OUT_HNDL(OUT_ATTR_SET, "out_attr_set",  MUTABOR | HABEO_REFERO,
1349                      out_attr_set),
1350         DEF_OUT_HNDL(OUT_ATTR_GET, "out_attr_get",  HABEO_REFERO,
1351                      out_attr_get),
1352         DEF_OUT_HNDL(OUT_XATTR_SET, "out_xattr_set", MUTABOR | HABEO_REFERO,
1353                      out_xattr_set),
1354         DEF_OUT_HNDL(OUT_XATTR_DEL, "out_xattr_del", MUTABOR | HABEO_REFERO,
1355                      out_xattr_del),
1356         DEF_OUT_HNDL(OUT_XATTR_GET, "out_xattr_get", HABEO_REFERO,
1357                      out_xattr_get),
1358         DEF_OUT_HNDL(OUT_INDEX_LOOKUP, "out_index_lookup", HABEO_REFERO,
1359                      out_index_lookup),
1360         DEF_OUT_HNDL(OUT_INDEX_INSERT, "out_index_insert",
1361                      MUTABOR | HABEO_REFERO, out_index_insert),
1362         DEF_OUT_HNDL(OUT_INDEX_DELETE, "out_index_delete",
1363                      MUTABOR | HABEO_REFERO, out_index_delete),
1364         DEF_OUT_HNDL(OUT_WRITE, "out_write", MUTABOR | HABEO_REFERO, out_write),
1365 };
1366
1367 static struct tgt_handler *out_handler_find(__u32 opc)
1368 {
1369         struct tgt_handler *h;
1370
1371         h = NULL;
1372         if (OUT_CREATE <= opc && opc < OUT_LAST) {
1373                 h = &out_update_ops[opc - OUT_CREATE];
1374                 LASSERTF(h->th_opc == opc, "opcode mismatch %d != %d\n",
1375                          h->th_opc, opc);
1376         } else {
1377                 h = NULL; /* unsupported opc */
1378         }
1379         return h;
1380 }
1381
1382 static int out_tx_start(const struct lu_env *env, struct dt_device *dt,
1383                         struct thandle_exec_args *ta, struct obd_export *exp)
1384 {
1385         ta->ta_argno = 0;
1386         ta->ta_handle = dt_trans_create(env, dt);
1387         if (IS_ERR(ta->ta_handle)) {
1388                 int rc;
1389
1390                 rc = PTR_ERR(ta->ta_handle);
1391                 ta->ta_handle = NULL;
1392                 CERROR("%s: start handle error: rc = %d\n", dt_obd_name(dt),
1393                        rc);
1394                 return rc;
1395         }
1396         if (exp->exp_need_sync)
1397                 ta->ta_handle->th_sync = 1;
1398
1399         return 0;
1400 }
1401
1402 static int out_trans_start(const struct lu_env *env,
1403                            struct thandle_exec_args *ta)
1404 {
1405         return dt_trans_start(env, ta->ta_handle->th_dev, ta->ta_handle);
1406 }
1407
1408 static int out_trans_stop(const struct lu_env *env,
1409                           struct thandle_exec_args *ta, int err)
1410 {
1411         int i;
1412         int rc;
1413
1414         ta->ta_handle->th_result = err;
1415         rc = dt_trans_stop(env, ta->ta_handle->th_dev, ta->ta_handle);
1416         for (i = 0; i < ta->ta_argno; i++) {
1417                 if (ta->ta_args[i]->object != NULL) {
1418                         struct dt_object *obj = ta->ta_args[i]->object;
1419
1420                         /* If the object is being created during this
1421                          * transaction, we need to remove them from the
1422                          * cache immediately, because a few layers are
1423                          * missing in OUT handler, i.e. the object might
1424                          * not be initialized in all layers */
1425                         if (ta->ta_args[i]->exec_fn == out_tx_create_exec)
1426                                 set_bit(LU_OBJECT_HEARD_BANSHEE,
1427                                         &obj->do_lu.lo_header->loh_flags);
1428                         lu_object_put(env, &ta->ta_args[i]->object->do_lu);
1429                         ta->ta_args[i]->object = NULL;
1430                 }
1431         }
1432         ta->ta_handle = NULL;
1433         ta->ta_argno = 0;
1434
1435         return rc;
1436 }
1437
1438 static int out_tx_end(const struct lu_env *env, struct thandle_exec_args *ta,
1439                       int declare_ret)
1440 {
1441         struct tgt_session_info *tsi = tgt_ses_info(env);
1442         int                     i;
1443         int                     rc;
1444         int                     rc1;
1445         ENTRY;
1446
1447         if (ta->ta_handle == NULL)
1448                 RETURN(0);
1449
1450         if (declare_ret != 0 || ta->ta_argno == 0)
1451                 GOTO(stop, rc = declare_ret);
1452
1453         LASSERT(ta->ta_handle->th_dev != NULL);
1454         rc = out_trans_start(env, ta);
1455         if (unlikely(rc != 0))
1456                 GOTO(stop, rc);
1457
1458         for (i = 0; i < ta->ta_argno; i++) {
1459                 rc = ta->ta_args[i]->exec_fn(env, ta->ta_handle,
1460                                              ta->ta_args[i]);
1461                 if (unlikely(rc != 0)) {
1462                         CDEBUG(D_INFO, "error during execution of #%u from"
1463                                " %s:%d: rc = %d\n", i, ta->ta_args[i]->file,
1464                                ta->ta_args[i]->line, rc);
1465                         while (--i >= 0) {
1466                                 if (ta->ta_args[i]->undo_fn != NULL)
1467                                         ta->ta_args[i]->undo_fn(env,
1468                                                                ta->ta_handle,
1469                                                                ta->ta_args[i]);
1470                                 else
1471                                         CERROR("%s: undo for %s:%d: rc = %d\n",
1472                                              dt_obd_name(ta->ta_handle->th_dev),
1473                                                ta->ta_args[i]->file,
1474                                                ta->ta_args[i]->line, -ENOTSUPP);
1475                         }
1476                         break;
1477                 }
1478                 CDEBUG(D_INFO, "%s: executed %u/%u: rc = %d\n",
1479                        dt_obd_name(ta->ta_handle->th_dev), i, ta->ta_argno, rc);
1480         }
1481
1482         /* Only fail for real update */
1483         tsi->tsi_reply_fail_id = OBD_FAIL_OUT_UPDATE_NET_REP;
1484 stop:
1485         rc1 = out_trans_stop(env, ta, rc);
1486         if (rc == 0)
1487                 rc = rc1;
1488
1489         ta->ta_handle = NULL;
1490         ta->ta_argno = 0;
1491
1492         RETURN(rc);
1493 }
1494
1495 /**
1496  * Object updates between Targets. Because all the updates has been
1497  * dis-assemblied into object updates at sender side, so OUT will
1498  * call OSD API directly to execute these updates.
1499  *
1500  * In DNE phase I all of the updates in the request need to be executed
1501  * in one transaction, and the transaction has to be synchronously.
1502  *
1503  * Please refer to lustre/include/lustre/lustre_idl.h for req/reply
1504  * format.
1505  */
1506 int out_handle(struct tgt_session_info *tsi)
1507 {
1508         const struct lu_env             *env = tsi->tsi_env;
1509         struct tgt_thread_info          *tti = tgt_th_info(env);
1510         struct thandle_exec_args        *ta = &tti->tti_tea;
1511         struct req_capsule              *pill = tsi->tsi_pill;
1512         struct dt_device                *dt = tsi->tsi_tgt->lut_bottom;
1513         struct object_update_request    *ureq;
1514         struct object_update            *update;
1515         struct object_update_reply      *reply;
1516         int                              bufsize;
1517         int                              count;
1518         int                              current_batchid = -1;
1519         int                              i;
1520         int                              rc = 0;
1521         int                              rc1 = 0;
1522
1523         ENTRY;
1524
1525         req_capsule_set(pill, &RQF_OUT_UPDATE);
1526         ureq = req_capsule_client_get(pill, &RMF_OUT_UPDATE);
1527         if (ureq == NULL) {
1528                 CERROR("%s: No buf!: rc = %d\n", tgt_name(tsi->tsi_tgt),
1529                        -EPROTO);
1530                 RETURN(err_serious(-EPROTO));
1531         }
1532
1533         bufsize = req_capsule_get_size(pill, &RMF_OUT_UPDATE, RCL_CLIENT);
1534         if (bufsize != object_update_request_size(ureq)) {
1535                 CERROR("%s: invalid bufsize %d: rc = %d\n",
1536                        tgt_name(tsi->tsi_tgt), bufsize, -EPROTO);
1537                 RETURN(err_serious(-EPROTO));
1538         }
1539
1540         if (ureq->ourq_magic != UPDATE_REQUEST_MAGIC) {
1541                 CERROR("%s: invalid update buffer magic %x expect %x: "
1542                        "rc = %d\n", tgt_name(tsi->tsi_tgt), ureq->ourq_magic,
1543                        UPDATE_REQUEST_MAGIC, -EPROTO);
1544                 RETURN(err_serious(-EPROTO));
1545         }
1546
1547         count = ureq->ourq_count;
1548         if (count <= 0) {
1549                 CERROR("%s: empty update: rc = %d\n", tgt_name(tsi->tsi_tgt),
1550                        -EPROTO);
1551                 RETURN(err_serious(-EPROTO));
1552         }
1553
1554         req_capsule_set_size(pill, &RMF_OUT_UPDATE_REPLY, RCL_SERVER,
1555                              OUT_UPDATE_REPLY_SIZE);
1556         rc = req_capsule_server_pack(pill);
1557         if (rc != 0) {
1558                 CERROR("%s: Can't pack response: rc = %d\n",
1559                        tgt_name(tsi->tsi_tgt), rc);
1560                 RETURN(rc);
1561         }
1562
1563         /* Prepare the update reply buffer */
1564         reply = req_capsule_server_get(pill, &RMF_OUT_UPDATE_REPLY);
1565         if (reply == NULL)
1566                 RETURN(err_serious(-EPROTO));
1567         object_update_reply_init(reply, count);
1568         tti->tti_u.update.tti_update_reply = reply;
1569         tti->tti_mult_trans = !req_is_replay(tgt_ses_req(tsi));
1570
1571         /* Walk through updates in the request to execute them synchronously */
1572         for (i = 0; i < count; i++) {
1573                 struct tgt_handler      *h;
1574                 struct dt_object        *dt_obj;
1575
1576                 update = object_update_request_get(ureq, i, NULL);
1577                 if (update == NULL)
1578                         GOTO(out, rc = -EPROTO);
1579
1580                 if (ptlrpc_req_need_swab(pill->rc_req))
1581                         lustre_swab_object_update(update);
1582
1583                 if (!fid_is_sane(&update->ou_fid)) {
1584                         CERROR("%s: invalid FID "DFID": rc = %d\n",
1585                                tgt_name(tsi->tsi_tgt), PFID(&update->ou_fid),
1586                                -EPROTO);
1587                         GOTO(out, rc = err_serious(-EPROTO));
1588                 }
1589
1590                 dt_obj = dt_locate(env, dt, &update->ou_fid);
1591                 if (IS_ERR(dt_obj))
1592                         GOTO(out, rc = PTR_ERR(dt_obj));
1593
1594                 if (dt->dd_record_fid_accessed) {
1595                         lfsck_pack_rfa(&tti->tti_lr,
1596                                        lu_object_fid(&dt_obj->do_lu),
1597                                        LE_FID_ACCESSED,
1598                                        LFSCK_TYPE_LAYOUT);
1599                         tgt_lfsck_in_notify(env, dt, &tti->tti_lr, NULL);
1600                 }
1601
1602                 tti->tti_u.update.tti_dt_object = dt_obj;
1603                 tti->tti_u.update.tti_update = update;
1604                 tti->tti_u.update.tti_update_reply_index = i;
1605
1606                 h = out_handler_find(update->ou_type);
1607                 if (unlikely(h == NULL)) {
1608                         CERROR("%s: unsupported opc: 0x%x\n",
1609                                tgt_name(tsi->tsi_tgt), update->ou_type);
1610                         GOTO(next, rc = -ENOTSUPP);
1611                 }
1612
1613                 /* Check resend case only for modifying RPC */
1614                 if (h->th_flags & MUTABOR) {
1615                         struct ptlrpc_request *req = tgt_ses_req(tsi);
1616
1617                         if (out_check_resent(env, dt, dt_obj, req,
1618                                              out_reconstruct, reply, i))
1619                                 GOTO(next, rc = 0);
1620                 }
1621
1622                 /* start transaction for modification RPC only */
1623                 if (h->th_flags & MUTABOR && current_batchid == -1) {
1624                         current_batchid = update->ou_batchid;
1625                         rc = out_tx_start(env, dt, ta, tsi->tsi_exp);
1626                         if (rc != 0)
1627                                 GOTO(next, rc);
1628
1629                         if (update->ou_flags & UPDATE_FL_SYNC)
1630                                 ta->ta_handle->th_sync = 1;
1631                 }
1632
1633                 /* Stop the current update transaction, if the update has
1634                  * different batchid, or read-only update */
1635                 if (((current_batchid != update->ou_batchid) ||
1636                      !(h->th_flags & MUTABOR)) && ta->ta_handle != NULL) {
1637                         rc = out_tx_end(env, ta, rc);
1638                         current_batchid = -1;
1639                         if (rc != 0)
1640                                 GOTO(next, rc);
1641
1642                         /* start a new transaction if needed */
1643                         if (h->th_flags & MUTABOR) {
1644                                 rc = out_tx_start(env, dt, ta, tsi->tsi_exp);
1645                                 if (rc != 0)
1646                                         GOTO(next, rc);
1647
1648                                 if (update->ou_flags & UPDATE_FL_SYNC)
1649                                         ta->ta_handle->th_sync = 1;
1650
1651                                 current_batchid = update->ou_batchid;
1652                         }
1653                 }
1654
1655                 rc = h->th_act(tsi);
1656 next:
1657                 lu_object_put(env, &dt_obj->do_lu);
1658                 if (rc < 0)
1659                         GOTO(out, rc);
1660         }
1661 out:
1662         if (current_batchid != -1) {
1663                 rc1 = out_tx_end(env, ta, rc);
1664                 if (rc == 0)
1665                         rc = rc1;
1666         }
1667
1668         RETURN(rc);
1669 }
1670
1671 struct tgt_handler tgt_out_handlers[] = {
1672 TGT_UPDATE_HDL(MUTABOR, OUT_UPDATE,     out_handle),
1673 };
1674 EXPORT_SYMBOL(tgt_out_handlers);
1675