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