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/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 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 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 || 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 || 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, NULL);
 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 || 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, NULL);
 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) {
 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, NULL);
 477         dt_read_unlock(env, obj);
 478         if (rc < 0) {
 479                 lbuf->lb_len = 0;
 480                 GOTO(out, rc);
 481         }
 482         if (rc == 0) {
 483                 lbuf->lb_len = 0;
 484                 GOTO(out, rc = -ENOENT);
 485         }
 486         lbuf->lb_len = rc;
 487         rc = 0;
 488         CDEBUG(D_INFO, "%s: "DFID" get xattr %s len %d\n",
 489                tgt_name(tsi->tsi_tgt), PFID(lu_object_fid(&obj->do_lu)),
 490                name, (int)lbuf->lb_len);
 491
 492         GOTO(out, rc);
 493
 494 out:
 495         object_update_result_insert(reply, lbuf->lb_buf, lbuf->lb_len, idx, rc);
 496         RETURN(rc);
 497 }
 498
 499 static int out_index_lookup(struct tgt_session_info *tsi)
 500 {
 501         const struct lu_env     *env = tsi->tsi_env;
 502         struct tgt_thread_info  *tti = tgt_th_info(env);
 503         struct object_update    *update = tti->tti_u.update.tti_update;
 504         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
 505         char                    *name;
 506         int                      rc;
 507
 508         ENTRY;
 509
 510         if (!lu_object_exists(&obj->do_lu))
 511                 RETURN(-ENOENT);
 512
 513         name = object_update_param_get(update, 0, NULL);
 514         if (name == NULL) {
 515                 CERROR("%s: empty name for lookup: rc = %d\n",
 516                        tgt_name(tsi->tsi_tgt), -EPROTO);
 517                 RETURN(err_serious(-EPROTO));
 518         }
 519
 520         dt_read_lock(env, obj, MOR_TGT_CHILD);
 521         if (!dt_try_as_dir(env, obj))
 522                 GOTO(out_unlock, rc = -ENOTDIR);
 523
 524         rc = dt_lookup(env, obj, (struct dt_rec *)&tti->tti_fid1,
 525                 (struct dt_key *)name, NULL);
 526
 527         if (rc < 0)
 528                 GOTO(out_unlock, rc);
 529
 530         if (rc == 0)
 531                 rc += 1;
 532
 533 out_unlock:
 534         dt_read_unlock(env, obj);
 535
 536         CDEBUG(D_INFO, "lookup "DFID" %s get "DFID" rc %d\n",
 537                PFID(lu_object_fid(&obj->do_lu)), name,
 538                PFID(&tti->tti_fid1), rc);
 539
 540         CDEBUG(D_INFO, "%s: insert lookup reply %p index %d: rc = %d\n",
 541                tgt_name(tsi->tsi_tgt), tti->tti_u.update.tti_update_reply,
 542                0, rc);
 543
 544         object_update_result_insert(tti->tti_u.update.tti_update_reply,
 545                             &tti->tti_fid1, sizeof(tti->tti_fid1),
 546                             tti->tti_u.update.tti_update_reply_index, rc);
 547         RETURN(rc);
 548 }
 549
 550 static int out_tx_xattr_set_exec(const struct lu_env *env,
 551                                  struct thandle *th,
 552                                  struct tx_arg *arg)
 553 {
 554         struct dt_object *dt_obj = arg->object;
 555         int rc;
 556
 557         CDEBUG(D_INFO, "%s: set xattr buf %p name %s flag %d\n",
 558                dt_obd_name(th->th_dev), arg->u.xattr_set.buf.lb_buf,
 559                arg->u.xattr_set.name, arg->u.xattr_set.flags);
 560
 561         if (!lu_object_exists(&dt_obj->do_lu))
 562                 GOTO(out, rc = -ENOENT);
 563
 564         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 565         rc = dt_xattr_set(env, dt_obj, &arg->u.xattr_set.buf,
 566                           arg->u.xattr_set.name, arg->u.xattr_set.flags,
 567                           th, NULL);
 568         /**
 569          * Ignore errors if this is LINK EA
 570          **/
 571         if (unlikely(rc != 0 &&
 572                      strcmp(arg->u.xattr_set.name, XATTR_NAME_LINK) == 0)) {
 573                 /* XXX: If the linkEA is overflow, then we need to notify the
 574                  *      namespace LFSCK to skip "nlink" attribute verification
 575                  *      on this object to avoid the "nlink" to be shrinked by
 576                  *      wrong. It may be not good an interaction with LFSCK
 577                  *      like this. We will consider to replace it with other
 578                  *      mechanism in future. LU-5802. */
 579                 if (rc == -ENOSPC) {
 580                         struct lfsck_request *lr = &tgt_th_info(env)->tti_lr;
 581
 582                         lfsck_pack_rfa(lr, lu_object_fid(&dt_obj->do_lu),
 583                                        LE_SKIP_NLINK, LFSCK_TYPE_NAMESPACE);
 584                         tgt_lfsck_in_notify(env,
 585                                 tgt_ses_info(env)->tsi_tgt->lut_bottom, lr, th);
 586                 }
 587
 588                 rc = 0;
 589         }
 590         dt_write_unlock(env, dt_obj);
 591
 592 out:
 593         CDEBUG(D_INFO, "%s: insert xattr set reply %p index %d: rc = %d\n",
 594                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
 595
 596         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
 597
 598         return rc;
 599 }
 600
 601 static int __out_tx_xattr_set(const struct lu_env *env,
 602                               struct dt_object *dt_obj,
 603                               const struct lu_buf *buf,
 604                               const char *name, int flags,
 605                               struct thandle_exec_args *ta,
 606                               struct object_update_reply *reply,
 607                               int index, const char *file, int line)
 608 {
 609         struct tx_arg   *arg;
 610         int             rc;
 611
 612         LASSERT(ta->ta_handle != NULL);
 613         rc = dt_declare_xattr_set(env, dt_obj, buf, name, flags, ta->ta_handle);
 614         if (rc != 0)
 615                 return rc;
 616
 617         if (strcmp(name, XATTR_NAME_LINK) == 0) {
 618                 struct lfsck_request *lr = &tgt_th_info(env)->tti_lr;
 619
 620                 /* XXX: If the linkEA is overflow, then we need to notify the
 621                  *      namespace LFSCK to skip "nlink" attribute verification
 622                  *      on this object to avoid the "nlink" to be shrinked by
 623                  *      wrong. It may be not good an interaction with LFSCK
 624                  *      like this. We will consider to replace it with other
 625                  *      mechanism in future. LU-5802. */
 626                 lfsck_pack_rfa(lr, lu_object_fid(&dt_obj->do_lu),
 627                                LE_SKIP_NLINK_DECLARE, LFSCK_TYPE_NAMESPACE);
 628                 rc = tgt_lfsck_in_notify(env,
 629                                          tgt_ses_info(env)->tsi_tgt->lut_bottom,
 630                                          lr, ta->ta_handle);
 631                 if (rc != 0)
 632                         return rc;
 633         }
 634
 635         arg = tx_add_exec(ta, out_tx_xattr_set_exec, NULL, file, line);
 636         if (IS_ERR(arg))
 637                 return PTR_ERR(arg);
 638
 639         lu_object_get(&dt_obj->do_lu);
 640         arg->object = dt_obj;
 641         arg->u.xattr_set.name = name;
 642         arg->u.xattr_set.flags = flags;
 643         arg->u.xattr_set.buf = *buf;
 644         arg->reply = reply;
 645         arg->index = index;
 646         arg->u.xattr_set.csum = 0;
 647         return 0;
 648 }
 649
 650 static int out_xattr_set(struct tgt_session_info *tsi)
 651 {
 652         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
 653         struct object_update    *update = tti->tti_u.update.tti_update;
 654         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
 655         struct lu_buf           *lbuf = &tti->tti_buf;
 656         char                    *name;
 657         char                    *buf;
 658         __u32                   *tmp;
 659         size_t                   buf_len = 0;
 660         int                      flag;
 661         size_t                   size = 0;
 662         int                      rc;
 663         ENTRY;
 664
 665         name = object_update_param_get(update, 0, NULL);
 666         if (name == NULL) {
 667                 CERROR("%s: empty name for xattr set: rc = %d\n",
 668                        tgt_name(tsi->tsi_tgt), -EPROTO);
 669                 RETURN(err_serious(-EPROTO));
 670         }
 671
 672         buf = object_update_param_get(update, 1, &buf_len);
 673         if (buf == NULL || buf_len == 0) {
 674                 CERROR("%s: empty buf for xattr set: rc = %d\n",
 675                        tgt_name(tsi->tsi_tgt), -EPROTO);
 676                 RETURN(err_serious(-EPROTO));
 677         }
 678
 679         lbuf->lb_buf = buf;
 680         lbuf->lb_len = buf_len;
 681
 682         tmp = object_update_param_get(update, 2, &size);
 683         if (tmp == NULL || size != sizeof(*tmp)) {
 684                 CERROR("%s: emptry or wrong size %zu flag: rc = %d\n",
 685                        tgt_name(tsi->tsi_tgt), size, -EPROTO);
 686                 RETURN(err_serious(-EPROTO));
 687         }
 688
 689         if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req))
 690                 __swab32s(tmp);
 691         flag = *tmp;
 692
 693         rc = out_tx_xattr_set(tsi->tsi_env, obj, lbuf, name, flag,
 694                               &tti->tti_tea,
 695                               tti->tti_u.update.tti_update_reply,
 696                               tti->tti_u.update.tti_update_reply_index);
 697         RETURN(rc);
 698 }
 699
 700 static int out_tx_xattr_del_exec(const struct lu_env *env, struct thandle *th,
 701                                  struct tx_arg *arg)
 702 {
 703         struct dt_object *dt_obj = arg->object;
 704         int rc;
 705
 706         CDEBUG(D_INFO, "%s: del xattr name '%s' on "DFID"\n",
 707                dt_obd_name(th->th_dev), arg->u.xattr_set.name,
 708                PFID(lu_object_fid(&dt_obj->do_lu)));
 709
 710         if (!lu_object_exists(&dt_obj->do_lu))
 711                 GOTO(out, rc = -ENOENT);
 712
 713         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 714         rc = dt_xattr_del(env, dt_obj, arg->u.xattr_set.name,
 715                           th, NULL);
 716         dt_write_unlock(env, dt_obj);
 717 out:
 718         CDEBUG(D_INFO, "%s: insert xattr del reply %p index %d: rc = %d\n",
 719                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
 720
 721         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
 722
 723         return rc;
 724 }
 725
 726 static int __out_tx_xattr_del(const struct lu_env *env,
 727                               struct dt_object *dt_obj, const char *name,
 728                               struct thandle_exec_args *ta,
 729                               struct object_update_reply *reply,
 730                               int index, const char *file, int line)
 731 {
 732         struct tx_arg   *arg;
 733         int             rc;
 734
 735         rc = dt_declare_xattr_del(env, dt_obj, name, ta->ta_handle);
 736         if (rc != 0)
 737                 return rc;
 738
 739         arg = tx_add_exec(ta, out_tx_xattr_del_exec, NULL, file, line);
 740         if (IS_ERR(arg))
 741                 return PTR_ERR(arg);
 742
 743         lu_object_get(&dt_obj->do_lu);
 744         arg->object = dt_obj;
 745         arg->u.xattr_set.name = name;
 746         arg->reply = reply;
 747         arg->index = index;
 748         return 0;
 749 }
 750
 751 static int out_xattr_del(struct tgt_session_info *tsi)
 752 {
 753         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
 754         struct object_update    *update = tti->tti_u.update.tti_update;
 755         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
 756         char                    *name;
 757         int                      rc;
 758         ENTRY;
 759
 760         name = object_update_param_get(update, 0, NULL);
 761         if (name == NULL) {
 762                 CERROR("%s: empty name for xattr set: rc = %d\n",
 763                        tgt_name(tsi->tsi_tgt), -EPROTO);
 764                 RETURN(err_serious(-EPROTO));
 765         }
 766
 767         rc = out_tx_xattr_del(tsi->tsi_env, obj, name, &tti->tti_tea,
 768                               tti->tti_u.update.tti_update_reply,
 769                               tti->tti_u.update.tti_update_reply_index);
 770         RETURN(rc);
 771 }
 772
 773 static int out_obj_ref_add(const struct lu_env *env,
 774                            struct dt_object *dt_obj,
 775                            struct thandle *th)
 776 {
 777         int rc;
 778
 779         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 780         rc = dt_ref_add(env, dt_obj, th);
 781         dt_write_unlock(env, dt_obj);
 782
 783         return rc;
 784 }
 785
 786 static int out_obj_ref_del(const struct lu_env *env,
 787                            struct dt_object *dt_obj,
 788                            struct thandle *th)
 789 {
 790         int rc;
 791
 792         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 793         rc = dt_ref_del(env, dt_obj, th);
 794         dt_write_unlock(env, dt_obj);
 795
 796         return rc;
 797 }
 798
 799 static int out_tx_ref_add_exec(const struct lu_env *env, struct thandle *th,
 800                                struct tx_arg *arg)
 801 {
 802         struct dt_object *dt_obj = arg->object;
 803         int rc;
 804
 805         rc = out_obj_ref_add(env, dt_obj, th);
 806
 807         CDEBUG(D_INFO, "%s: insert ref_add reply %p index %d: rc = %d\n",
 808                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
 809
 810         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
 811         return rc;
 812 }
 813
 814 static int out_tx_ref_add_undo(const struct lu_env *env, struct thandle *th,
 815                                struct tx_arg *arg)
 816 {
 817         return out_obj_ref_del(env, arg->object, th);
 818 }
 819
 820 static int __out_tx_ref_add(const struct lu_env *env,
 821                             struct dt_object *dt_obj,
 822                             struct thandle_exec_args *ta,
 823                             struct object_update_reply *reply,
 824                             int index, const char *file, int line)
 825 {
 826         struct tx_arg   *arg;
 827         int             rc;
 828
 829         LASSERT(ta->ta_handle != NULL);
 830         rc = dt_declare_ref_add(env, dt_obj, ta->ta_handle);
 831         if (rc != 0)
 832                 return rc;
 833
 834         arg = tx_add_exec(ta, out_tx_ref_add_exec, out_tx_ref_add_undo, file,
 835                           line);
 836         if (IS_ERR(arg))
 837                 return PTR_ERR(arg);
 838
 839         lu_object_get(&dt_obj->do_lu);
 840         arg->object = dt_obj;
 841         arg->reply = reply;
 842         arg->index = index;
 843         return 0;
 844 }
 845
 846 /**
 847  * increase ref of the object
 848  **/
 849 static int out_ref_add(struct tgt_session_info *tsi)
 850 {
 851         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
 852         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
 853         int                      rc;
 854
 855         ENTRY;
 856
 857         rc = out_tx_ref_add(tsi->tsi_env, obj, &tti->tti_tea,
 858                             tti->tti_u.update.tti_update_reply,
 859                             tti->tti_u.update.tti_update_reply_index);
 860         RETURN(rc);
 861 }
 862
 863 static int out_tx_ref_del_exec(const struct lu_env *env, struct thandle *th,
 864                                struct tx_arg *arg)
 865 {
 866         struct dt_object        *dt_obj = arg->object;
 867         int                      rc;
 868
 869         rc = out_obj_ref_del(env, dt_obj, th);
 870
 871         CDEBUG(D_INFO, "%s: insert ref_del reply %p index %d: rc = %d\n",
 872                dt_obd_name(th->th_dev), arg->reply, arg->index, 0);
 873
 874         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
 875
 876         return rc;
 877 }
 878
 879 static int out_tx_ref_del_undo(const struct lu_env *env, struct thandle *th,
 880                                struct tx_arg *arg)
 881 {
 882         return out_obj_ref_add(env, arg->object, th);
 883 }
 884
 885 static int __out_tx_ref_del(const struct lu_env *env,
 886                             struct dt_object *dt_obj,
 887                             struct thandle_exec_args *ta,
 888                             struct object_update_reply *reply,
 889                             int index, const char *file, int line)
 890 {
 891         struct tx_arg   *arg;
 892         int             rc;
 893
 894         LASSERT(ta->ta_handle != NULL);
 895         rc = dt_declare_ref_del(env, dt_obj, ta->ta_handle);
 896         if (rc != 0)
 897                 return rc;
 898
 899         arg = tx_add_exec(ta, out_tx_ref_del_exec, out_tx_ref_del_undo, file,
 900                           line);
 901         if (IS_ERR(arg))
 902                 return PTR_ERR(arg);
 903
 904         lu_object_get(&dt_obj->do_lu);
 905         arg->object = dt_obj;
 906         arg->reply = reply;
 907         arg->index = index;
 908         return 0;
 909 }
 910
 911 static int out_ref_del(struct tgt_session_info *tsi)
 912 {
 913         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
 914         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
 915         int                      rc;
 916
 917         ENTRY;
 918
 919         if (!lu_object_exists(&obj->do_lu))
 920                 RETURN(-ENOENT);
 921
 922         rc = out_tx_ref_del(tsi->tsi_env, obj, &tti->tti_tea,
 923                             tti->tti_u.update.tti_update_reply,
 924                             tti->tti_u.update.tti_update_reply_index);
 925         RETURN(rc);
 926 }
 927
 928 static int out_obj_index_insert(const struct lu_env *env,
 929                                 struct dt_object *dt_obj,
 930                                 const struct dt_rec *rec,
 931                                 const struct dt_key *key,
 932                                 struct thandle *th)
 933 {
 934         int rc;
 935
 936         CDEBUG(D_INFO, "%s: index insert "DFID" name: %s fid "DFID", type %u\n",
 937                dt_obd_name(th->th_dev), PFID(lu_object_fid(&dt_obj->do_lu)),
 938                (char *)key, PFID(((struct dt_insert_rec *)rec)->rec_fid),
 939                ((struct dt_insert_rec *)rec)->rec_type);
 940
 941         if (dt_try_as_dir(env, dt_obj) == 0)
 942                 return -ENOTDIR;
 943
 944         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 945         rc = dt_insert(env, dt_obj, rec, key, th, NULL, 0);
 946         dt_write_unlock(env, dt_obj);
 947
 948         return rc;
 949 }
 950
 951 static int out_obj_index_delete(const struct lu_env *env,
 952                                 struct dt_object *dt_obj,
 953                                 const struct dt_key *key,
 954                                 struct thandle *th)
 955 {
 956         int rc;
 957
 958         CDEBUG(D_INFO, "%s: index delete "DFID" name: %s\n",
 959                dt_obd_name(th->th_dev), PFID(lu_object_fid(&dt_obj->do_lu)),
 960                (char *)key);
 961
 962         if (dt_try_as_dir(env, dt_obj) == 0)
 963                 return -ENOTDIR;
 964
 965         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
 966         rc = dt_delete(env, dt_obj, key, th, NULL);
 967         dt_write_unlock(env, dt_obj);
 968
 969         return rc;
 970 }
 971
 972 static int out_tx_index_insert_exec(const struct lu_env *env,
 973                                     struct thandle *th, struct tx_arg *arg)
 974 {
 975         struct dt_object *dt_obj = arg->object;
 976         int rc;
 977
 978         if (unlikely(!dt_object_exists(dt_obj)))
 979                 RETURN(-ESTALE);
 980
 981         rc = out_obj_index_insert(env, dt_obj,
 982                                   (const struct dt_rec *)&arg->u.insert.rec,
 983                                   arg->u.insert.key, th);
 984
 985         CDEBUG(D_INFO, "%s: insert idx insert reply %p index %d: rc = %d\n",
 986                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
 987
 988         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
 989
 990         return rc;
 991 }
 992
 993 static int out_tx_index_insert_undo(const struct lu_env *env,
 994                                     struct thandle *th, struct tx_arg *arg)
 995 {
 996         return out_obj_index_delete(env, arg->object, arg->u.insert.key, th);
 997 }
 998
 999 static int __out_tx_index_insert(const struct lu_env *env,
1000                                  struct dt_object *dt_obj,
1001                                  const struct dt_rec *rec,
1002                                  const struct dt_key *key,
1003                                  struct thandle_exec_args *ta,
1004                                  struct object_update_reply *reply,
1005                                  int index, const char *file, int line)
1006 {
1007         struct tx_arg   *arg;
1008         int             rc;
1009
1010         LASSERT(ta->ta_handle != NULL);
1011         if (dt_try_as_dir(env, dt_obj) == 0) {
1012                 rc = -ENOTDIR;
1013                 return rc;
1014         }
1015
1016         rc = dt_declare_insert(env, dt_obj, rec, key, ta->ta_handle);
1017         if (rc != 0)
1018                 return rc;
1019
1020         arg = tx_add_exec(ta, out_tx_index_insert_exec,
1021                           out_tx_index_insert_undo, file, line);
1022         if (IS_ERR(arg))
1023                 return PTR_ERR(arg);
1024
1025         lu_object_get(&dt_obj->do_lu);
1026         arg->object = dt_obj;
1027         arg->reply = reply;
1028         arg->index = index;
1029         arg->u.insert.rec = *(const struct dt_insert_rec *)rec;
1030         arg->u.insert.key = key;
1031
1032         return 0;
1033 }
1034
1035 static int out_index_insert(struct tgt_session_info *tsi)
1036 {
1037         struct tgt_thread_info  *tti    = tgt_th_info(tsi->tsi_env);
1038         struct object_update    *update = tti->tti_u.update.tti_update;
1039         struct dt_object        *obj    = tti->tti_u.update.tti_dt_object;
1040         struct dt_insert_rec    *rec    = &tti->tti_rec;
1041         struct lu_fid           *fid;
1042         char                    *name;
1043         __u32                   *ptype;
1044         int                      rc     = 0;
1045         size_t                   size;
1046         ENTRY;
1047
1048         name = object_update_param_get(update, 0, NULL);
1049         if (name == NULL) {
1050                 CERROR("%s: empty name for index insert: rc = %d\n",
1051                        tgt_name(tsi->tsi_tgt), -EPROTO);
1052                 RETURN(err_serious(-EPROTO));
1053         }
1054
1055         fid = object_update_param_get(update, 1, &size);
1056         if (fid == NULL || size != sizeof(*fid)) {
1057                 CERROR("%s: invalid fid: rc = %d\n",
1058                        tgt_name(tsi->tsi_tgt), -EPROTO);
1059                        RETURN(err_serious(-EPROTO));
1060         }
1061
1062         if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req))
1063                 lustre_swab_lu_fid(fid);
1064
1065         if (!fid_is_sane(fid)) {
1066                 CERROR("%s: invalid FID "DFID": rc = %d\n",
1067                        tgt_name(tsi->tsi_tgt), PFID(fid), -EPROTO);
1068                 RETURN(err_serious(-EPROTO));
1069         }
1070
1071         ptype = object_update_param_get(update, 2, &size);
1072         if (ptype == NULL || size != sizeof(*ptype)) {
1073                 CERROR("%s: invalid type for index insert: rc = %d\n",
1074                        tgt_name(tsi->tsi_tgt), -EPROTO);
1075                 RETURN(err_serious(-EPROTO));
1076         }
1077
1078         if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req))
1079                 __swab32s(ptype);
1080
1081         rec->rec_fid = fid;
1082         rec->rec_type = *ptype;
1083
1084         rc = out_tx_index_insert(tsi->tsi_env, obj, (const struct dt_rec *)rec,
1085                                  (const struct dt_key *)name, &tti->tti_tea,
1086                                  tti->tti_u.update.tti_update_reply,
1087                                  tti->tti_u.update.tti_update_reply_index);
1088         RETURN(rc);
1089 }
1090
1091 static int out_tx_index_delete_exec(const struct lu_env *env,
1092                                     struct thandle *th,
1093                                     struct tx_arg *arg)
1094 {
1095         int rc;
1096
1097         rc = out_obj_index_delete(env, arg->object, arg->u.insert.key, th);
1098
1099         CDEBUG(D_INFO, "%s: delete idx insert reply %p index %d: rc = %d\n",
1100                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
1101
1102         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
1103
1104         return rc;
1105 }
1106
1107 static int out_tx_index_delete_undo(const struct lu_env *env,
1108                                     struct thandle *th,
1109                                     struct tx_arg *arg)
1110 {
1111         CERROR("%s: Oops, can not rollback index_delete yet: rc = %d\n",
1112                dt_obd_name(th->th_dev), -ENOTSUPP);
1113         return -ENOTSUPP;
1114 }
1115
1116 static int __out_tx_index_delete(const struct lu_env *env,
1117                                  struct dt_object *dt_obj,
1118                                  const struct dt_key *key,
1119                                  struct thandle_exec_args *ta,
1120                                  struct object_update_reply *reply,
1121                                  int index, const char *file, int line)
1122 {
1123         struct tx_arg   *arg;
1124         int             rc;
1125
1126         if (dt_try_as_dir(env, dt_obj) == 0) {
1127                 rc = -ENOTDIR;
1128                 return rc;
1129         }
1130
1131         LASSERT(ta->ta_handle != NULL);
1132         rc = dt_declare_delete(env, dt_obj, key, ta->ta_handle);
1133         if (rc != 0)
1134                 return rc;
1135
1136         arg = tx_add_exec(ta, out_tx_index_delete_exec,
1137                           out_tx_index_delete_undo, file, line);
1138         if (IS_ERR(arg))
1139                 return PTR_ERR(arg);
1140
1141         lu_object_get(&dt_obj->do_lu);
1142         arg->object = dt_obj;
1143         arg->reply = reply;
1144         arg->index = index;
1145         arg->u.insert.key = key;
1146         return 0;
1147 }
1148
1149 static int out_index_delete(struct tgt_session_info *tsi)
1150 {
1151         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
1152         struct object_update    *update = tti->tti_u.update.tti_update;
1153         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
1154         char                    *name;
1155         int                      rc = 0;
1156
1157         if (!lu_object_exists(&obj->do_lu))
1158                 RETURN(-ENOENT);
1159
1160         name = object_update_param_get(update, 0, NULL);
1161         if (name == NULL) {
1162                 CERROR("%s: empty name for index delete: rc = %d\n",
1163                        tgt_name(tsi->tsi_tgt), -EPROTO);
1164                 RETURN(err_serious(-EPROTO));
1165         }
1166
1167         rc = out_tx_index_delete(tsi->tsi_env, obj, (const struct dt_key *)name,
1168                                  &tti->tti_tea,
1169                                  tti->tti_u.update.tti_update_reply,
1170                                  tti->tti_u.update.tti_update_reply_index);
1171         RETURN(rc);
1172 }
1173
1174 static int out_tx_destroy_exec(const struct lu_env *env, struct thandle *th,
1175                                struct tx_arg *arg)
1176 {
1177         struct dt_object *dt_obj = arg->object;
1178         int rc;
1179
1180         rc = out_obj_destroy(env, dt_obj, th);
1181
1182         CDEBUG(D_INFO, "%s: insert destroy reply %p index %d: rc = %d\n",
1183                dt_obd_name(th->th_dev), arg->reply, arg->index, rc);
1184
1185         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
1186
1187         RETURN(rc);
1188 }
1189
1190 static int out_tx_destroy_undo(const struct lu_env *env, struct thandle *th,
1191                                struct tx_arg *arg)
1192 {
1193         CERROR("%s: not support destroy undo yet!: rc = %d\n",
1194                dt_obd_name(th->th_dev), -ENOTSUPP);
1195         return -ENOTSUPP;
1196 }
1197
1198 static int __out_tx_destroy(const struct lu_env *env, struct dt_object *dt_obj,
1199                              struct thandle_exec_args *ta,
1200                              struct object_update_reply *reply,
1201                              int index, const char *file, int line)
1202 {
1203         struct tx_arg   *arg;
1204         int             rc;
1205
1206         LASSERT(ta->ta_handle != NULL);
1207         rc = dt_declare_destroy(env, dt_obj, ta->ta_handle);
1208         if (rc != 0)
1209                 return rc;
1210
1211         arg = tx_add_exec(ta, out_tx_destroy_exec, out_tx_destroy_undo,
1212                           file, line);
1213         if (IS_ERR(arg))
1214                 return PTR_ERR(arg);
1215
1216         lu_object_get(&dt_obj->do_lu);
1217         arg->object = dt_obj;
1218         arg->reply = reply;
1219         arg->index = index;
1220         return 0;
1221 }
1222
1223 static int out_destroy(struct tgt_session_info *tsi)
1224 {
1225         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
1226         struct object_update    *update = tti->tti_u.update.tti_update;
1227         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
1228         struct lu_fid           *fid;
1229         int                      rc;
1230         ENTRY;
1231
1232         fid = &update->ou_fid;
1233         if (!fid_is_sane(fid)) {
1234                 CERROR("%s: invalid FID "DFID": rc = %d\n",
1235                        tgt_name(tsi->tsi_tgt), PFID(fid), -EPROTO);
1236                 RETURN(err_serious(-EPROTO));
1237         }
1238
1239         if (!lu_object_exists(&obj->do_lu))
1240                 RETURN(-ENOENT);
1241
1242         rc = out_tx_destroy(tsi->tsi_env, obj, &tti->tti_tea,
1243                             tti->tti_u.update.tti_update_reply,
1244                             tti->tti_u.update.tti_update_reply_index);
1245
1246         RETURN(rc);
1247 }
1248
1249 static int out_tx_write_exec(const struct lu_env *env, struct thandle *th,
1250                              struct tx_arg *arg)
1251 {
1252         struct dt_object *dt_obj = arg->object;
1253         int rc;
1254
1255         dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
1256         rc = dt_record_write(env, dt_obj, &arg->u.write.buf,
1257                              &arg->u.write.pos, th);
1258         dt_write_unlock(env, dt_obj);
1259
1260         if (rc == 0)
1261                 rc = arg->u.write.buf.lb_len;
1262
1263         object_update_result_insert(arg->reply, NULL, 0, arg->index, rc);
1264
1265         return rc > 0 ? 0 : rc;
1266 }
1267
1268 static int __out_tx_write(const struct lu_env *env,
1269                           struct dt_object *dt_obj,
1270                           const struct lu_buf *buf,
1271                           loff_t pos, struct thandle_exec_args *ta,
1272                           struct object_update_reply *reply,
1273                           int index, const char *file, int line)
1274 {
1275         struct tx_arg   *arg;
1276         int             rc;
1277
1278         LASSERT(ta->ta_handle != NULL);
1279         rc = dt_declare_record_write(env, dt_obj, buf, pos, ta->ta_handle);
1280         if (rc != 0)
1281                 return rc;
1282
1283         arg = tx_add_exec(ta, out_tx_write_exec, NULL, file, line);
1284         if (IS_ERR(arg))
1285                 return PTR_ERR(arg);
1286
1287         lu_object_get(&dt_obj->do_lu);
1288         arg->object = dt_obj;
1289         arg->u.write.buf = *buf;
1290         arg->u.write.pos = pos;
1291         arg->reply = reply;
1292         arg->index = index;
1293         return 0;
1294 }
1295
1296 static int out_write(struct tgt_session_info *tsi)
1297 {
1298         struct tgt_thread_info  *tti = tgt_th_info(tsi->tsi_env);
1299         struct object_update    *update = tti->tti_u.update.tti_update;
1300         struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
1301         struct lu_buf           *lbuf = &tti->tti_buf;
1302         char                    *buf;
1303         __u64                   *tmp;
1304         size_t                  size = 0;
1305         size_t                  buf_len = 0;
1306         loff_t                  pos;
1307         int                      rc;
1308         ENTRY;
1309
1310         buf = object_update_param_get(update, 0, &buf_len);
1311         if (buf == NULL || buf_len == 0) {
1312                 CERROR("%s: empty buf for xattr set: rc = %d\n",
1313                        tgt_name(tsi->tsi_tgt), -EPROTO);
1314                 RETURN(err_serious(-EPROTO));
1315         }
1316         lbuf->lb_buf = buf;
1317         lbuf->lb_len = buf_len;
1318
1319         tmp = object_update_param_get(update, 1, &size);
1320         if (tmp == NULL || size != sizeof(*tmp)) {
1321                 CERROR("%s: empty or wrong size %zu pos: rc = %d\n",
1322                        tgt_name(tsi->tsi_tgt), size, -EPROTO);
1323                 RETURN(err_serious(-EPROTO));
1324         }
1325
1326         if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req))
1327                 __swab64s(tmp);
1328         pos = *tmp;
1329
1330         rc = out_tx_write(tsi->tsi_env, obj, lbuf, pos,
1331                           &tti->tti_tea,
1332                           tti->tti_u.update.tti_update_reply,
1333                           tti->tti_u.update.tti_update_reply_index);
1334         RETURN(rc);
1335 }
1336
1337 #define DEF_OUT_HNDL(opc, name, flags, fn)     \
1338 [opc - OUT_CREATE] = {                                  \
1339         .th_name    = name,                             \
1340         .th_fail_id = 0,                                \
1341         .th_opc     = opc,                              \
1342         .th_flags   = flags,                            \
1343         .th_act     = fn,                               \
1344         .th_fmt     = NULL,                             \
1345         .th_version = 0,                                \
1346 }
1347
1348 static struct tgt_handler out_update_ops[] = {
1349         DEF_OUT_HNDL(OUT_CREATE, "out_create", MUTABOR | HABEO_REFERO,
1350                      out_create),
1351         DEF_OUT_HNDL(OUT_DESTROY, "out_create", MUTABOR | HABEO_REFERO,
1352                      out_destroy),
1353         DEF_OUT_HNDL(OUT_REF_ADD, "out_ref_add", MUTABOR | HABEO_REFERO,
1354                      out_ref_add),
1355         DEF_OUT_HNDL(OUT_REF_DEL, "out_ref_del", MUTABOR | HABEO_REFERO,
1356                      out_ref_del),
1357         DEF_OUT_HNDL(OUT_ATTR_SET, "out_attr_set",  MUTABOR | HABEO_REFERO,
1358                      out_attr_set),
1359         DEF_OUT_HNDL(OUT_ATTR_GET, "out_attr_get",  HABEO_REFERO,
1360                      out_attr_get),
1361         DEF_OUT_HNDL(OUT_XATTR_SET, "out_xattr_set", MUTABOR | HABEO_REFERO,
1362                      out_xattr_set),
1363         DEF_OUT_HNDL(OUT_XATTR_DEL, "out_xattr_del", MUTABOR | HABEO_REFERO,
1364                      out_xattr_del),
1365         DEF_OUT_HNDL(OUT_XATTR_GET, "out_xattr_get", HABEO_REFERO,
1366                      out_xattr_get),
1367         DEF_OUT_HNDL(OUT_INDEX_LOOKUP, "out_index_lookup", HABEO_REFERO,
1368                      out_index_lookup),
1369         DEF_OUT_HNDL(OUT_INDEX_INSERT, "out_index_insert",
1370                      MUTABOR | HABEO_REFERO, out_index_insert),
1371         DEF_OUT_HNDL(OUT_INDEX_DELETE, "out_index_delete",
1372                      MUTABOR | HABEO_REFERO, out_index_delete),
1373         DEF_OUT_HNDL(OUT_WRITE, "out_write", MUTABOR | HABEO_REFERO, out_write),
1374 };
1375
1376 struct tgt_handler *out_handler_find(__u32 opc)
1377 {
1378         struct tgt_handler *h;
1379
1380         h = NULL;
1381         if (OUT_CREATE <= opc && opc < OUT_LAST) {
1382                 h = &out_update_ops[opc - OUT_CREATE];
1383                 LASSERTF(h->th_opc == opc, "opcode mismatch %d != %d\n",
1384                          h->th_opc, opc);
1385         } else {
1386                 h = NULL; /* unsupported opc */
1387         }
1388         return h;
1389 }
1390
1391 static int out_tx_start(const struct lu_env *env, struct dt_device *dt,
1392                         struct thandle_exec_args *ta, struct obd_export *exp)
1393 {
1394         ta->ta_argno = 0;
1395         ta->ta_handle = dt_trans_create(env, dt);
1396         if (IS_ERR(ta->ta_handle)) {
1397                 int rc;
1398
1399                 rc = PTR_ERR(ta->ta_handle);
1400                 ta->ta_handle = NULL;
1401                 CERROR("%s: start handle error: rc = %d\n", dt_obd_name(dt),
1402                        rc);
1403                 return rc;
1404         }
1405         if (exp->exp_need_sync)
1406                 ta->ta_handle->th_sync = 1;
1407
1408         return 0;
1409 }
1410
1411 static int out_trans_start(const struct lu_env *env,
1412                            struct thandle_exec_args *ta)
1413 {
1414         return dt_trans_start(env, ta->ta_handle->th_dev, ta->ta_handle);
1415 }
1416
1417 static int out_trans_stop(const struct lu_env *env,
1418                           struct thandle_exec_args *ta, int err)
1419 {
1420         int i;
1421         int rc;
1422
1423         ta->ta_handle->th_result = err;
1424         rc = dt_trans_stop(env, ta->ta_handle->th_dev, ta->ta_handle);
1425         for (i = 0; i < ta->ta_argno; i++) {
1426                 if (ta->ta_args[i]->object != NULL) {
1427                         struct dt_object *obj = ta->ta_args[i]->object;
1428
1429                         /* If the object is being created during this
1430                          * transaction, we need to remove them from the
1431                          * cache immediately, because a few layers are
1432                          * missing in OUT handler, i.e. the object might
1433                          * not be initialized in all layers */
1434                         if (ta->ta_args[i]->exec_fn == out_tx_create_exec)
1435                                 set_bit(LU_OBJECT_HEARD_BANSHEE,
1436                                         &obj->do_lu.lo_header->loh_flags);
1437                         lu_object_put(env, &ta->ta_args[i]->object->do_lu);
1438                         ta->ta_args[i]->object = NULL;
1439                 }
1440         }
1441         ta->ta_handle = NULL;
1442         ta->ta_argno = 0;
1443
1444         return rc;
1445 }
1446
1447 int out_tx_end(const struct lu_env *env, struct thandle_exec_args *ta,
1448                int declare_ret)
1449 {
1450         struct tgt_session_info *tsi = tgt_ses_info(env);
1451         int                     i;
1452         int                     rc;
1453         int                     rc1;
1454         ENTRY;
1455
1456         if (ta->ta_handle == NULL)
1457                 RETURN(0);
1458
1459         if (declare_ret != 0 || ta->ta_argno == 0)
1460                 GOTO(stop, rc = declare_ret);
1461
1462         LASSERT(ta->ta_handle->th_dev != NULL);
1463         rc = out_trans_start(env, ta);
1464         if (unlikely(rc != 0))
1465                 GOTO(stop, rc);
1466
1467         for (i = 0; i < ta->ta_argno; i++) {
1468                 rc = ta->ta_args[i]->exec_fn(env, ta->ta_handle,
1469                                              ta->ta_args[i]);
1470                 if (unlikely(rc != 0)) {
1471                         CDEBUG(D_INFO, "error during execution of #%u from"
1472                                " %s:%d: rc = %d\n", i, ta->ta_args[i]->file,
1473                                ta->ta_args[i]->line, rc);
1474                         while (--i >= 0) {
1475                                 if (ta->ta_args[i]->undo_fn != NULL)
1476                                         ta->ta_args[i]->undo_fn(env,
1477                                                                ta->ta_handle,
1478                                                                ta->ta_args[i]);
1479                                 else
1480                                         CERROR("%s: undo for %s:%d: rc = %d\n",
1481                                              dt_obd_name(ta->ta_handle->th_dev),
1482                                                ta->ta_args[i]->file,
1483                                                ta->ta_args[i]->line, -ENOTSUPP);
1484                         }
1485                         break;
1486                 }
1487                 CDEBUG(D_INFO, "%s: executed %u/%u: rc = %d\n",
1488                        dt_obd_name(ta->ta_handle->th_dev), i, ta->ta_argno, rc);
1489         }
1490
1491         /* Only fail for real update */
1492         tsi->tsi_reply_fail_id = OBD_FAIL_OUT_UPDATE_NET_REP;
1493 stop:
1494         rc1 = out_trans_stop(env, ta, rc);
1495         if (rc == 0)
1496                 rc = rc1;
1497
1498         ta->ta_handle = NULL;
1499         ta->ta_argno = 0;
1500
1501         RETURN(rc);
1502 }
1503
1504 /**
1505  * Object updates between Targets. Because all the updates has been
1506  * dis-assemblied into object updates at sender side, so OUT will
1507  * call OSD API directly to execute these updates.
1508  *
1509  * In DNE phase I all of the updates in the request need to be executed
1510  * in one transaction, and the transaction has to be synchronously.
1511  *
1512  * Please refer to lustre/include/lustre/lustre_idl.h for req/reply
1513  * format.
1514  */
1515 int out_handle(struct tgt_session_info *tsi)
1516 {
1517         const struct lu_env             *env = tsi->tsi_env;
1518         struct tgt_thread_info          *tti = tgt_th_info(env);
1519         struct thandle_exec_args        *ta = &tti->tti_tea;
1520         struct req_capsule              *pill = tsi->tsi_pill;
1521         struct dt_device                *dt = tsi->tsi_tgt->lut_bottom;
1522         struct object_update_request    *ureq;
1523         struct object_update            *update;
1524         struct object_update_reply      *reply;
1525         int                              bufsize;
1526         int                              count;
1527         int                              current_batchid = -1;
1528         int                              i;
1529         int                              rc = 0;
1530         int                              rc1 = 0;
1531
1532         ENTRY;
1533
1534         req_capsule_set(pill, &RQF_OUT_UPDATE);
1535         ureq = req_capsule_client_get(pill, &RMF_OUT_UPDATE);
1536         if (ureq == NULL) {
1537                 CERROR("%s: No buf!: rc = %d\n", tgt_name(tsi->tsi_tgt),
1538                        -EPROTO);
1539                 RETURN(err_serious(-EPROTO));
1540         }
1541
1542         bufsize = req_capsule_get_size(pill, &RMF_OUT_UPDATE, RCL_CLIENT);
1543         if (bufsize != object_update_request_size(ureq)) {
1544                 CERROR("%s: invalid bufsize %d: rc = %d\n",
1545                        tgt_name(tsi->tsi_tgt), bufsize, -EPROTO);
1546                 RETURN(err_serious(-EPROTO));
1547         }
1548
1549         if (ureq->ourq_magic != UPDATE_REQUEST_MAGIC) {
1550                 CERROR("%s: invalid update buffer magic %x expect %x: "
1551                        "rc = %d\n", tgt_name(tsi->tsi_tgt), ureq->ourq_magic,
1552                        UPDATE_REQUEST_MAGIC, -EPROTO);
1553                 RETURN(err_serious(-EPROTO));
1554         }
1555
1556         count = ureq->ourq_count;
1557         if (count <= 0) {
1558                 CERROR("%s: empty update: rc = %d\n", tgt_name(tsi->tsi_tgt),
1559                        -EPROTO);
1560                 RETURN(err_serious(-EPROTO));
1561         }
1562
1563         req_capsule_set_size(pill, &RMF_OUT_UPDATE_REPLY, RCL_SERVER,
1564                              OUT_UPDATE_REPLY_SIZE);
1565         rc = req_capsule_server_pack(pill);
1566         if (rc != 0) {
1567                 CERROR("%s: Can't pack response: rc = %d\n",
1568                        tgt_name(tsi->tsi_tgt), rc);
1569                 RETURN(rc);
1570         }
1571
1572         /* Prepare the update reply buffer */
1573         reply = req_capsule_server_get(pill, &RMF_OUT_UPDATE_REPLY);
1574         if (reply == NULL)
1575                 RETURN(err_serious(-EPROTO));
1576         object_update_reply_init(reply, count);
1577         tti->tti_u.update.tti_update_reply = reply;
1578         tti->tti_mult_trans = !req_is_replay(tgt_ses_req(tsi));
1579
1580         /* Walk through updates in the request to execute them synchronously */
1581         for (i = 0; i < count; i++) {
1582                 struct tgt_handler      *h;
1583                 struct dt_object        *dt_obj;
1584
1585                 update = object_update_request_get(ureq, i, NULL);
1586                 if (update == NULL)
1587                         GOTO(out, rc = -EPROTO);
1588
1589                 if (ptlrpc_req_need_swab(pill->rc_req))
1590                         lustre_swab_object_update(update);
1591
1592                 if (!fid_is_sane(&update->ou_fid)) {
1593                         CERROR("%s: invalid FID "DFID": rc = %d\n",
1594                                tgt_name(tsi->tsi_tgt), PFID(&update->ou_fid),
1595                                -EPROTO);
1596                         GOTO(out, rc = err_serious(-EPROTO));
1597                 }
1598
1599                 dt_obj = dt_locate(env, dt, &update->ou_fid);
1600                 if (IS_ERR(dt_obj))
1601                         GOTO(out, rc = PTR_ERR(dt_obj));
1602
1603                 if (dt->dd_record_fid_accessed) {
1604                         lfsck_pack_rfa(&tti->tti_lr,
1605                                        lu_object_fid(&dt_obj->do_lu),
1606                                        LE_FID_ACCESSED,
1607                                        LFSCK_TYPE_LAYOUT);
1608                         tgt_lfsck_in_notify(env, dt, &tti->tti_lr, NULL);
1609                 }
1610
1611                 tti->tti_u.update.tti_dt_object = dt_obj;
1612                 tti->tti_u.update.tti_update = update;
1613                 tti->tti_u.update.tti_update_reply_index = i;
1614
1615                 h = out_handler_find(update->ou_type);
1616                 if (unlikely(h == NULL)) {
1617                         CERROR("%s: unsupported opc: 0x%x\n",
1618                                tgt_name(tsi->tsi_tgt), update->ou_type);
1619                         GOTO(next, rc = -ENOTSUPP);
1620                 }
1621
1622                 /* Check resend case only for modifying RPC */
1623                 if (h->th_flags & MUTABOR) {
1624                         struct ptlrpc_request *req = tgt_ses_req(tsi);
1625
1626                         if (out_check_resent(env, dt, dt_obj, req,
1627                                              out_reconstruct, reply, i))
1628                                 GOTO(next, rc = 0);
1629                 }
1630
1631                 /* start transaction for modification RPC only */
1632                 if (h->th_flags & MUTABOR && current_batchid == -1) {
1633                         current_batchid = update->ou_batchid;
1634                         rc = out_tx_start(env, dt, ta, tsi->tsi_exp);
1635                         if (rc != 0)
1636                                 GOTO(next, rc);
1637                 }
1638
1639                 /* Stop the current update transaction, if the update has
1640                  * different batchid, or read-only update */
1641                 if (((current_batchid != update->ou_batchid) ||
1642                      !(h->th_flags & MUTABOR)) && ta->ta_handle != NULL) {
1643                         rc = out_tx_end(env, ta, rc);
1644                         current_batchid = -1;
1645                         if (rc != 0)
1646                                 GOTO(next, rc);
1647
1648                         /* start a new transaction if needed */
1649                         if (h->th_flags & MUTABOR) {
1650                                 rc = out_tx_start(env, dt, ta, tsi->tsi_exp);
1651                                 if (rc != 0)
1652                                         GOTO(next, rc);
1653
1654                                 current_batchid = update->ou_batchid;
1655                         }
1656                 }
1657
1658                 rc = h->th_act(tsi);
1659 next:
1660                 lu_object_put(env, &dt_obj->do_lu);
1661                 if (rc < 0)
1662                         GOTO(out, rc);
1663         }
1664 out:
1665         if (current_batchid != -1) {
1666                 rc1 = out_tx_end(env, ta, rc);
1667                 if (rc == 0)
1668                         rc = rc1;
1669         }
1670
1671         RETURN(rc);
1672 }
1673
1674 struct tgt_handler tgt_out_handlers[] = {
1675 TGT_UPDATE_HDL(MUTABOR, OUT_UPDATE,     out_handle),
1676 };
1677 EXPORT_SYMBOL(tgt_out_handlers);
1678