lustre/target/update_records.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) 2014, Intel Corporation.
  24  */
  25
  26 /*
  27  * lustre/target/update_records.c
  28  *
  29  * This file implement the methods to pack updates as update records, which
  30  * will be written to the disk as llog record, and might be used during
  31  * recovery.
  32  *
  33  * For cross-MDT operation, all of updates of the operation needs to be
  34  * recorded in the disk, then during recovery phase, the recovery thread
  35  * will retrieve and redo these updates if it needed.
  36  *
  37  * See comments above struct update_records for the format of update_records.
  38  *
  39  * Author: Di Wang <di.wang@intel.com>
  40  */
  41 #define DEBUG_SUBSYSTEM S_CLASS
  42
  43 #include <lu_target.h>
  44 #include <lustre_update.h>
  45 #include <obd.h>
  46 #include <obd_class.h>
  47 #include "tgt_internal.h"
  48
  49 #define UPDATE_RECORDS_BUFFER_SIZE      8192
  50 #define UPDATE_PARAMS_BUFFER_SIZE       8192
  51 /**
  52  * Dump update record.
  53  *
  54  * Dump all of updates in the update_records, mostly for debugging purpose.
  55  *
  56  * \param[in] records   update records to be dumpped
  57  * \param[in] mask      debug level mask
  58  * \param[in] dump_params if dump all of updates the updates.
  59  *
  60  */
  61 void update_records_dump(const struct update_records *records,
  62                          unsigned int mask, bool dump_updates)
  63 {
  64         const struct update_ops *ops;
  65         const struct update_op  *op = NULL;
  66         struct update_params    *params;
  67         unsigned int            i;
  68
  69         ops = &records->ur_ops;
  70         params = update_records_get_params(records);
  71
  72         CDEBUG(mask, "master transno = "LPU64" batchid = "LPU64" flags = %x"
  73                " ops = %d params = %d\n", records->ur_master_transno,
  74                records->ur_batchid, records->ur_flags, records->ur_update_count,
  75                records->ur_param_count);
  76
  77         if (records->ur_update_count == 0)
  78                 return;
  79
  80         if (!dump_updates)
  81                 return;
  82
  83         op = &ops->uops_op[0];
  84         for (i = 0; i < records->ur_update_count; i++) {
  85                 unsigned int j;
  86
  87                 CDEBUG(mask, "update %dth "DFID" %s params_count = %hu\n", i,
  88                        PFID(&op->uop_fid), update_op_str(op->uop_type),
  89                        op->uop_param_count);
  90
  91                 for (j = 0;  j < op->uop_param_count; j++) {
  92                         struct object_update_param *param;
  93
  94                         param = update_params_get_param(params,
  95                                         (unsigned int)op->uop_params_off[j],
  96                                         records->ur_param_count);
  97
  98                         LASSERT(param != NULL);
  99                         CDEBUG(mask, "param = %p %dth off = %hu size = %hu\n",
 100                                param, j, op->uop_params_off[j], param->oup_len);
 101                 }
 102
 103                 op = update_op_next_op(op);
 104         }
 105 }
 106
 107 /**
 108  * Pack parameters to update records
 109  *
 110  * Find and insert parameter to update records, if the parameter
 111  * already exists in \a params, then just return the offset of this
 112  * parameter, otherwise insert the parameter and return its offset
 113  *
 114  * \param[in] params    update params in which to insert parameter
 115  * \param[in] new_param parameters to be inserted.
 116  * \param[in] new_param_size    the size of \a new_param
 117  *
 118  * \retval              index inside \a params if parameter insertion
 119  *                      succeeds.
 120  * \retval              negative errno if it fails.
 121  */
 122 static unsigned int update_records_param_pack(struct update_params *params,
 123                                               const void *new_param,
 124                                               size_t new_param_size,
 125                                               unsigned int *param_count)
 126 {
 127         struct object_update_param      *param;
 128         unsigned int                    i;
 129
 130         for (i = 0; i < *param_count; i++) {
 131                 struct object_update_param *param;
 132
 133                 param = update_params_get_param(params, i, *param_count);
 134                 if ((new_param == NULL && param->oup_len == new_param_size) ||
 135                     (param->oup_len == new_param_size &&
 136                      memcmp(param->oup_buf, new_param, new_param_size) == 0))
 137                         /* Found the parameter and return its index */
 138                         return i;
 139         }
 140
 141         param = (struct object_update_param *)((char *)params +
 142                                 update_params_size(params, *param_count));
 143
 144         param->oup_len = new_param_size;
 145         if (new_param != NULL)
 146                 memcpy(param->oup_buf, new_param, new_param_size);
 147
 148         *param_count = *param_count + 1;
 149
 150         return *param_count - 1;
 151 }
 152
 153 /**
 154  * Pack update to update records
 155  *
 156  * Pack the update and its parameters to the update records. First it will
 157  * insert parameters, get the offset of these parameter, then fill the
 158  * update with these offset. If insertion exceed the maximum size of
 159  * current update records, it will return -E2BIG here, and the caller might
 160  * extend the update_record size \see lod_updates_pack.
 161  *
 162  * \param[in] env       execution environment
 163  * \param[in] fid       FID of the update.
 164  * \param[in] op_type   operation type of the update
 165  * \param[in] ops       ur_ops in update records
 166  * \param[in|out] op_count      pointer to the count of ops
 167  * \param[in|out] max_op_size maximum size of the update
 168  * \param[in] params    ur_params in update records
 169  * \param[in|out] param_count   pointer to the count of params
 170  * \param[in|out] max_param_size maximum size of the parameter
 171  * \param[in] param_bufs        buffers of parameters
 172  * \param[in] params_buf_count  the count of the parameter buffers
 173  * \param[in] param_size        sizes of parameters
 174  *
 175  * \retval              0 if packing succeeds
 176  * \retval              negative errno if packing fails
 177  */
 178 static int update_records_update_pack(const struct lu_env *env,
 179                                       const struct lu_fid *fid,
 180                                       enum update_type op_type,
 181                                       struct update_ops *ops,
 182                                       unsigned int *op_count,
 183                                       size_t *max_op_size,
 184                                       struct update_params *params,
 185                                       unsigned int *param_count,
 186                                       size_t *max_param_size,
 187                                       unsigned int param_bufs_count,
 188                                       const void **param_bufs,
 189                                       size_t *param_sizes)
 190 {
 191         struct update_op        *op;
 192         size_t                  total_param_sizes = 0;
 193         int                     index;
 194         unsigned int            i;
 195
 196         /* Check whether the packing exceeding the maximum update size */
 197         if (unlikely(*max_op_size < update_op_size(param_bufs_count))) {
 198                 CDEBUG(D_INFO, "max_op_size = %zu update_op = %zu\n",
 199                        *max_op_size, update_op_size(param_bufs_count));
 200                 *max_op_size = update_op_size(param_bufs_count);
 201                 return -E2BIG;
 202         }
 203
 204         for (i = 0; i < param_bufs_count; i++)
 205                 total_param_sizes +=
 206                         cfs_size_round(sizeof(struct object_update_param) +
 207                                        param_sizes[i]);
 208
 209         /* Check whether the packing exceeding the maximum parameter size */
 210         if (unlikely(*max_param_size < total_param_sizes)) {
 211                 CDEBUG(D_INFO, "max_param_size = %zu params size = %zu\n",
 212                        *max_param_size, total_param_sizes);
 213
 214                 *max_param_size = total_param_sizes;
 215                 return -E2BIG;
 216         }
 217
 218         op = update_ops_get_op(ops, *op_count, *op_count);
 219         op->uop_fid = *fid;
 220         op->uop_type = op_type;
 221         op->uop_param_count = param_bufs_count;
 222         for (i = 0; i < param_bufs_count; i++) {
 223                 index = update_records_param_pack(params, param_bufs[i],
 224                                                   param_sizes[i], param_count);
 225                 if (index < 0)
 226                         return index;
 227
 228                 CDEBUG(D_INFO, "%s %uth param offset = %d size = %zu\n",
 229                        update_op_str(op_type), i, index, param_sizes[i]);
 230
 231                 op->uop_params_off[i] = index;
 232         }
 233         CDEBUG(D_INFO, "%huth "DFID" %s param_count = %u\n",
 234                *op_count, PFID(fid), update_op_str(op_type), *param_count);
 235
 236         *op_count = *op_count + 1;
 237
 238         return 0;
 239 }
 240
 241 /**
 242  * Pack create update
 243  *
 244  * Pack create update into update records.
 245  *
 246  * \param[in] env       execution environment
 247  * \param[in] ops       ur_ops in update records
 248  * \param[in|out] op_count      pointer to the count of ops
 249  * \param[in|out] max_op_size maximum size of the update
 250  * \param[in] params    ur_params in update records
 251  * \param[in|out] param_count   pointer to the count of params
 252  * \param[in|out] max_param_size maximum size of the parameter
 253  * \param[in] fid       FID of the object to be created
 254  * \param[in] attr      attribute of the object to be created
 255  * \param[in] hint      creation hint
 256  * \param[in] dof       creation format information
 257  *
 258  * \retval              0 if packing succeeds.
 259  * \retval              negative errno if packing fails.
 260  */
 261 int update_records_create_pack(const struct lu_env *env,
 262                                struct update_ops *ops,
 263                                unsigned int *op_count,
 264                                size_t *max_ops_size,
 265                                struct update_params *params,
 266                                unsigned int *param_count,
 267                                size_t *max_param_size,
 268                                const struct lu_fid *fid,
 269                                const struct lu_attr *attr,
 270                                const struct dt_allocation_hint *hint,
 271                                struct dt_object_format *dof)
 272 {
 273         size_t                  sizes[2];
 274         const void              *bufs[2];
 275         int                     buf_count = 0;
 276         const struct lu_fid     *parent_fid = NULL;
 277         struct lu_fid           tmp_fid;
 278         int                     rc;
 279         struct obdo             *obdo;
 280
 281         if (attr != NULL) {
 282                 obdo = &update_env_info(env)->uti_obdo;
 283                 obdo->o_valid = 0;
 284                 obdo_from_la(obdo, attr, attr->la_valid);
 285                 lustre_set_wire_obdo(NULL, obdo, obdo);
 286                 bufs[buf_count] = obdo;
 287                 sizes[buf_count] = sizeof(*obdo);
 288                 buf_count++;
 289         }
 290
 291         if (hint != NULL && hint->dah_parent != NULL) {
 292                 parent_fid = lu_object_fid(&hint->dah_parent->do_lu);
 293                 fid_cpu_to_le(&tmp_fid, parent_fid);
 294                 bufs[buf_count] = &tmp_fid;
 295                 sizes[buf_count] = sizeof(tmp_fid);
 296                 buf_count++;
 297         }
 298
 299         rc = update_records_update_pack(env, fid, OUT_CREATE, ops, op_count,
 300                                         max_ops_size, params, param_count,
 301                                         max_param_size, buf_count, bufs, sizes);
 302         return rc;
 303 }
 304 EXPORT_SYMBOL(update_records_create_pack);
 305
 306 /**
 307  * Pack attr set update
 308  *
 309  * Pack attr_set update into update records.
 310  *
 311  * \param[in] env       execution environment
 312  * \param[in] ops       ur_ops in update records
 313  * \param[in|out] op_count pointer to the count of ops
 314  * \param[in|out] max_op_size maximum size of the update
 315  * \param[in] params    ur_params in update records
 316  * \param[in|out] param_count pointer to the count of params
 317  * \param[in|out] max_param_size maximum size of the parameter
 318  * \param[in] fid       FID of the object to set attr
 319  * \param[in] attr      attribute of attr set
 320  *
 321  * \retval              0 if packing succeeds.
 322  * \retval              negative errno if packing fails.
 323  */
 324 int update_records_attr_set_pack(const struct lu_env *env,
 325                                  struct update_ops *ops,
 326                                  unsigned int *op_count,
 327                                  size_t *max_ops_size,
 328                                  struct update_params *params,
 329                                  unsigned int *param_count,
 330                                  size_t *max_param_size,
 331                                  const struct lu_fid *fid,
 332                                  const struct lu_attr *attr)
 333 {
 334         struct obdo *obdo = &update_env_info(env)->uti_obdo;
 335         size_t size = sizeof(*obdo);
 336
 337         obdo->o_valid = 0;
 338         obdo_from_la(obdo, attr, attr->la_valid);
 339         lustre_set_wire_obdo(NULL, obdo, obdo);
 340         return update_records_update_pack(env, fid, OUT_ATTR_SET, ops, op_count,
 341                                           max_ops_size, params, param_count,
 342                                           max_param_size, 1,
 343                                           (const void **)&obdo, &size);
 344 }
 345 EXPORT_SYMBOL(update_records_attr_set_pack);
 346
 347 /**
 348  * Pack ref add update
 349  *
 350  * Pack ref add update into update records.
 351  *
 352  * \param[in] env       execution environment
 353  * \param[in] ops       ur_ops in update records
 354  * \param[in|out] op_count pointer to the count of ops
 355  * \param[in|out] max_op_size maximum size of the update
 356  * \param[in] params    ur_params in update records
 357  * \param[in|out] param_count pointer to the count of params
 358  * \param[in|out] max_param_size maximum size of the parameter
 359  * \param[in] fid       FID of the object to add reference
 360  *
 361  * \retval              0 if packing succeeds.
 362  * \retval              negative errno if packing fails.
 363  */
 364 int update_records_ref_add_pack(const struct lu_env *env,
 365                                 struct update_ops *ops,
 366                                 unsigned int *op_count,
 367                                 size_t *max_ops_size,
 368                                 struct update_params *params,
 369                                 unsigned int *param_count,
 370                                 size_t *max_param_size,
 371                                 const struct lu_fid *fid)
 372 {
 373         return update_records_update_pack(env, fid, OUT_REF_ADD, ops, op_count,
 374                                           max_ops_size, params, param_count,
 375                                           max_param_size, 0, NULL, NULL);
 376 }
 377 EXPORT_SYMBOL(update_records_ref_add_pack);
 378
 379 /**
 380  * Pack ref del update
 381  *
 382  * Pack ref del update into update records.
 383  *
 384  * \param[in] env       execution environment
 385  * \param[in] ops       ur_ops in update records
 386  * \param[in|out] op_count      pointer to the count of ops
 387  * \param[in|out] max_op_size maximum size of the update
 388  * \param[in] params    ur_params in update records
 389  * \param[in] param_count       pointer to the count of params
 390  * \param[in|out] max_param_size maximum size of the parameter
 391  * \param[in] fid       FID of the object to delete reference
 392  *
 393  * \retval              0 if packing succeeds.
 394  * \retval              negative errno if packing fails.
 395  */
 396 int update_records_ref_del_pack(const struct lu_env *env,
 397                                 struct update_ops *ops,
 398                                 unsigned int *op_count,
 399                                 size_t *max_ops_size,
 400                                 struct update_params *params,
 401                                 unsigned int *param_count,
 402                                 size_t *max_param_size,
 403                                 const struct lu_fid *fid)
 404 {
 405         return update_records_update_pack(env, fid, OUT_REF_DEL, ops, op_count,
 406                                           max_ops_size, params, param_count,
 407                                           max_param_size, 0, NULL, NULL);
 408 }
 409 EXPORT_SYMBOL(update_records_ref_del_pack);
 410
 411 /**
 412  * Pack object destroy update
 413  *
 414  * Pack object destroy update into update records.
 415  *
 416  * \param[in] env       execution environment
 417  * \param[in] ops       ur_ops in update records
 418  * \param[in|out] op_count pointer to the count of ops
 419  * \param[in|out] max_op_size maximum size of the update
 420  * \param[in] params    ur_params in update records
 421  * \param[in|out] param_count   pointer to the count of params
 422  * \param[in|out] max_param_size maximum size of the parameter
 423  * \param[in] fid       FID of the object to delete reference
 424  *
 425  * \retval              0 if packing succeeds.
 426  * \retval              negative errno if packing fails.
 427  */
 428 int update_records_object_destroy_pack(const struct lu_env *env,
 429                                        struct update_ops *ops,
 430                                        unsigned int *op_count,
 431                                        size_t *max_ops_size,
 432                                        struct update_params *params,
 433                                        unsigned int *param_count,
 434                                        size_t *max_param_size,
 435                                        const struct lu_fid *fid)
 436 {
 437         return update_records_update_pack(env, fid, OUT_DESTROY, ops, op_count,
 438                                           max_ops_size, params, param_count,
 439                                           max_param_size, 0, NULL, NULL);
 440 }
 441 EXPORT_SYMBOL(update_records_object_destroy_pack);
 442
 443 /**
 444  * Pack index insert update
 445  *
 446  * Pack index insert update into update records.
 447  *
 448  * \param[in] env       execution environment
 449  * \param[in] ops       ur_ops in update records
 450  * \param[in] op_count  pointer to the count of ops
 451  * \param[in|out] max_op_size maximum size of the update
 452  * \param[in] params    ur_params in update records
 453  * \param[in] param_count       pointer to the count of params
 454  * \param[in|out] max_param_size maximum size of the parameter
 455  * \param[in] fid       FID of the object to insert index
 456  * \param[in] rec       record of insertion
 457  * \param[in] key       key of insertion
 458  *
 459  * \retval              0 if packing succeeds.
 460  * \retval              negative errno if packing fails.
 461  */
 462 int update_records_index_insert_pack(const struct lu_env *env,
 463                                      struct update_ops *ops,
 464                                      unsigned int *op_count,
 465                                      size_t *max_ops_size,
 466                                      struct update_params *params,
 467                                      unsigned int *param_count,
 468                                      size_t *max_param_size,
 469                                      const struct lu_fid *fid,
 470                                      const struct dt_rec *rec,
 471                                      const struct dt_key *key)
 472 {
 473         struct dt_insert_rec       *rec1 = (struct dt_insert_rec *)rec;
 474         struct lu_fid              rec_fid;
 475         __u32                      type = cpu_to_le32(rec1->rec_type);
 476         size_t                     sizes[3] = { strlen((const char *)key) + 1,
 477                                                 sizeof(rec_fid),
 478                                                 sizeof(type) };
 479         const void                 *bufs[3] = { key,
 480                                                 &rec_fid,
 481                                                 &type };
 482
 483         fid_cpu_to_le(&rec_fid, rec1->rec_fid);
 484
 485         return update_records_update_pack(env, fid, OUT_INDEX_INSERT, ops,
 486                                           op_count, max_ops_size, params,
 487                                           param_count, max_param_size,
 488                                           3, bufs, sizes);
 489 }
 490 EXPORT_SYMBOL(update_records_index_insert_pack);
 491
 492 /**
 493  * Pack index delete update
 494  *
 495  * Pack index delete update into update records.
 496  *
 497  * \param[in] env       execution environment
 498  * \param[in] ops       ur_ops in update records
 499  * \param[in|out] op_count      pointer to the count of ops
 500  * \param[in|out] max_op_size maximum size of the update
 501  * \param[in] params    ur_params in update records
 502  * \param[in|ount] param_count  pointer to the count of params
 503  * \param[in|out] max_param_size maximum size of the parameter
 504  * \param[in] fid       FID of the object to delete index
 505  * \param[in] key       key of deletion
 506  *
 507  * \retval              0 if packing succeeds.
 508  * \retval              negative errno if packing fails.
 509  */
 510 int update_records_index_delete_pack(const struct lu_env *env,
 511                                      struct update_ops *ops,
 512                                      unsigned int *op_count,
 513                                      size_t *max_ops_size,
 514                                      struct update_params *params,
 515                                      unsigned int *param_count,
 516                                      size_t *max_param_size,
 517                                      const struct lu_fid *fid,
 518                                      const struct dt_key *key)
 519 {
 520         size_t size = strlen((const char *)key) + 1;
 521
 522         return update_records_update_pack(env, fid, OUT_INDEX_DELETE, ops,
 523                                           op_count, max_ops_size, params,
 524                                           param_count, max_param_size,
 525                                           1, (const void **)&key, &size);
 526 }
 527 EXPORT_SYMBOL(update_records_index_delete_pack);
 528
 529 /**
 530  * Pack xattr set update
 531  *
 532  * Pack xattr set update into update records.
 533  *
 534  * \param[in] env       execution environment
 535  * \param[in] ops       ur_ops in update records
 536  * \param[in|out] op_count      pointer to the count of ops
 537  * \param[in|out] max_op_size maximum size of the update
 538  * \param[in] params    ur_params in update records
 539  * \param[in|out] param_count   pointer to the count of params
 540  * \param[in|out] max_param_size maximum size of the parameter
 541  * \param[in] fid       FID of the object to set xattr
 542  * \param[in] buf       xattr to be set
 543  * \param[in] name      name of the xattr
 544  * \param[in] flag      flag for setting xattr
 545  *
 546  * \retval              0 if packing succeeds.
 547  * \retval              negative errno if packing fails.
 548  */
 549 int update_records_xattr_set_pack(const struct lu_env *env,
 550                                   struct update_ops *ops,
 551                                   unsigned int *op_count,
 552                                   size_t *max_ops_size,
 553                                   struct update_params *params,
 554                                   unsigned int *param_count,
 555                                   size_t *max_param_size,
 556                                   const struct lu_fid *fid,
 557                                   const struct lu_buf *buf, const char *name,
 558                                   __u32 flag)
 559 {
 560         size_t  sizes[3] = {strlen(name) + 1, buf->lb_len, sizeof(flag)};
 561         const void *bufs[3] = {name, buf->lb_buf, &flag};
 562
 563         flag = cpu_to_le32(flag);
 564
 565         return update_records_update_pack(env, fid, OUT_XATTR_SET, ops,
 566                                           op_count, max_ops_size, params,
 567                                           param_count, max_param_size,
 568                                           3, bufs, sizes);
 569 }
 570 EXPORT_SYMBOL(update_records_xattr_set_pack);
 571
 572 /**
 573  * Pack xattr delete update
 574  *
 575  * Pack xattr delete update into update records.
 576  *
 577  * \param[in] env       execution environment
 578  * \param[in] ops       ur_ops in update records
 579  * \param[in|out] op_count      pointer to the count of ops
 580  * \param[in|out] max_op_size maximum size of the update
 581  * \param[in] params    ur_params in update records
 582  * \param[in|out] param_count   pointer to the count of params
 583  * \param[in|out] max_param_size maximum size of the parameter
 584  * \param[in] fid       FID of the object to delete xattr
 585  * \param[in] name      name of the xattr
 586  *
 587  * \retval              0 if packing succeeds.
 588  * \retval              negative errno if packing fails.
 589  */
 590 int update_records_xattr_del_pack(const struct lu_env *env,
 591                                   struct update_ops *ops,
 592                                   unsigned int *op_count,
 593                                   size_t *max_ops_size,
 594                                   struct update_params *params,
 595                                   unsigned int *param_count,
 596                                   size_t *max_param_size,
 597                                   const struct lu_fid *fid,
 598                                   const char *name)
 599 {
 600         size_t  size = strlen(name) + 1;
 601
 602         return update_records_update_pack(env, fid, OUT_XATTR_DEL, ops,
 603                                           op_count, max_ops_size, params,
 604                                           param_count, max_param_size,
 605                                           1, (const void **)&name, &size);
 606 }
 607 EXPORT_SYMBOL(update_records_xattr_del_pack);
 608
 609 /**
 610  * Pack write update
 611  *
 612  * Pack write update into update records.
 613  *
 614  * \param[in] env       execution environment
 615  * \param[in] ops       ur_ops in update records
 616  * \param[in|out] op_count      pointer to the count of ops
 617  * \param[in|out] max_op_size maximum size of the update
 618  * \param[in] params    ur_params in update records
 619  * \param[in|out] param_count   pointer to the count of params
 620  * \param[in|out] max_param_size maximum size of the parameter
 621  * \param[in] fid       FID of the object to write into
 622  * \param[in] buf       buffer to write which includes an embedded size field
 623  * \param[in] pos       offet in the object to start writing at
 624  *
 625  * \retval              0 if packing succeeds.
 626  * \retval              negative errno if packing fails.
 627  */
 628 int update_records_write_pack(const struct lu_env *env,
 629                               struct update_ops *ops,
 630                               unsigned int *op_count,
 631                               size_t *max_ops_size,
 632                               struct update_params *params,
 633                               unsigned int *param_count,
 634                               size_t *max_param_size,
 635                               const struct lu_fid *fid,
 636                               const struct lu_buf *buf,
 637                               __u64 pos)
 638 {
 639         size_t          sizes[2] = {buf->lb_len, sizeof(pos)};
 640         const void      *bufs[2] = {buf->lb_buf, &pos};
 641
 642         pos = cpu_to_le64(pos);
 643
 644         return update_records_update_pack(env, fid, OUT_WRITE, ops,
 645                                           op_count, max_ops_size, params,
 646                                           param_count, max_param_size,
 647                                           2, bufs, sizes);
 648 }
 649 EXPORT_SYMBOL(update_records_write_pack);
 650
 651 /**
 652  * Pack punch
 653  *
 654  * Pack punch update into update records.
 655  *
 656  * \param[in] env       execution environment
 657  * \param[in] ops       ur_ops in update records
 658  * \param[in|out] op_count      pointer to the count of ops
 659  * \param[in|out] max_op_size maximum size of the update
 660  * \param[in] params    ur_params in update records
 661  * \param[in|out] param_count   pointer to the count of params
 662  * \param[in|out] max_param_size maximum size of the parameter
 663  * \param[in] fid       FID of the object to write into
 664  * \param[in] start     start offset of punch
 665  * \param[in] end       end offet of punch
 666  *
 667  * \retval              0 if packing succeeds.
 668  * \retval              negative errno if packing fails.
 669  */
 670 int update_records_punch_pack(const struct lu_env *env,
 671                               struct update_ops *ops,
 672                               unsigned int *op_count,
 673                               size_t *max_ops_size,
 674                               struct update_params *params,
 675                               unsigned int *param_count,
 676                               size_t *max_param_size,
 677                               const struct lu_fid *fid,
 678                               __u64 start, __u64 end)
 679 {
 680         size_t          sizes[2] = {sizeof(start), sizeof(end)};
 681         const void      *bufs[2] = {&start, &end};
 682
 683         start = cpu_to_le64(start);
 684         end = cpu_to_le64(end);
 685
 686         return update_records_update_pack(env, fid, OUT_PUNCH, ops, op_count,
 687                                           max_ops_size, params, param_count,
 688                                           max_param_size, 2, bufs, sizes);
 689 }
 690 EXPORT_SYMBOL(update_records_punch_pack);
 691
 692 /**
 693  * Create update records in thandle_update_records
 694  *
 695  * Allocate update_records for thandle_update_records, the initial size
 696  * will be 4KB.
 697  *
 698  * \param[in] tur       thandle_update_records where update_records will be
 699  *                      allocated
 700  * \retval              0 if allocation succeeds.
 701  * \retval              negative errno if allocation fails.
 702  */
 703 static int tur_update_records_create(struct thandle_update_records *tur)
 704 {
 705         if (tur->tur_update_records != NULL)
 706                 return 0;
 707
 708         OBD_ALLOC_LARGE(tur->tur_update_records,
 709                         UPDATE_RECORDS_BUFFER_SIZE);
 710
 711         if (tur->tur_update_records == NULL)
 712                 return -ENOMEM;
 713
 714         tur->tur_update_records_buf_size = UPDATE_RECORDS_BUFFER_SIZE;
 715
 716         return 0;
 717 }
 718
 719 /**
 720  * Extend update records
 721  *
 722  * Extend update_records to the new size in thandle_update_records.
 723  *
 724  * \param[in] tur       thandle_update_records where update_records will be
 725  *                      extended.
 726  * \retval              0 if extension succeeds.
 727  * \retval              negative errno if extension fails.
 728  */
 729 int tur_update_records_extend(struct thandle_update_records *tur,
 730                               size_t new_size)
 731 {
 732         struct llog_update_record       *record;
 733
 734         OBD_ALLOC_LARGE(record, new_size);
 735         if (record == NULL)
 736                 return -ENOMEM;
 737
 738         if (tur->tur_update_records != NULL) {
 739                 memcpy(record, tur->tur_update_records,
 740                        tur->tur_update_records_buf_size);
 741                 OBD_FREE_LARGE(tur->tur_update_records,
 742                                tur->tur_update_records_buf_size);
 743         }
 744
 745         tur->tur_update_records = record;
 746         tur->tur_update_records_buf_size = new_size;
 747
 748         return 0;
 749 }
 750 EXPORT_SYMBOL(tur_update_records_extend);
 751
 752 /**
 753  * Extend update records
 754  *
 755  * Extend update records in thandle to make sure it is able to hold
 756  * the update with certain update_op and params size.
 757  *
 758  * \param [in] tur      thandle_update_records to be extend
 759  * \param [in] new_op_size update_op size of the update record
 760  * \param [in] new_param_size params size of the update record
 761  *
 762  * \retval              0 if the update_records is being extended.
 763  * \retval              negative errno if the update_records is not being
 764  *                      extended.
 765  */
 766 int tur_update_extend(struct thandle_update_records *tur,
 767                       size_t new_op_size, size_t new_param_size)
 768 {
 769         size_t record_size;
 770         size_t params_size;
 771         size_t extend_size;
 772         int rc;
 773         ENTRY;
 774
 775         record_size = llog_update_record_size(tur->tur_update_records);
 776         /* extend update records buffer */
 777         if (new_op_size > (tur->tur_update_records_buf_size - record_size -
 778                            sizeof(*tur->tur_update_records))) {
 779                 extend_size = round_up(new_op_size, UPDATE_RECORDS_BUFFER_SIZE);
 780                 rc = tur_update_records_extend(tur,
 781                                 tur->tur_update_records_buf_size +
 782                                 extend_size);
 783                 if (rc != 0)
 784                         RETURN(rc);
 785         }
 786
 787         /* extend parameters buffer */
 788         params_size = update_params_size(tur->tur_update_params,
 789                                          tur->tur_update_param_count);
 790         if (new_param_size > (tur->tur_update_params_buf_size -
 791                               params_size)) {
 792                 extend_size = round_up(new_param_size,
 793                                        UPDATE_PARAMS_BUFFER_SIZE);
 794                 rc = tur_update_params_extend(tur,
 795                                 tur->tur_update_params_buf_size +
 796                                 extend_size);
 797                 if (rc != 0)
 798                         RETURN(rc);
 799         }
 800
 801         RETURN(0);
 802 }
 803 EXPORT_SYMBOL(tur_update_extend);
 804
 805 /**
 806  * Create update params in thandle_update_records
 807  *
 808  * Allocate update_params for thandle_update_records, the initial size
 809  * will be 4KB.
 810  *
 811  * \param[in] tur       thandle_update_records where update_params will be
 812  *                      allocated
 813  * \retval              0 if allocation succeeds.
 814  * \retval              negative errno if allocation fails.
 815  */
 816 static int tur_update_params_create(struct thandle_update_records *tur)
 817 {
 818         if (tur->tur_update_params != NULL)
 819                 return 0;
 820
 821         OBD_ALLOC_LARGE(tur->tur_update_params, UPDATE_PARAMS_BUFFER_SIZE);
 822         if (tur->tur_update_params == NULL)
 823                 return -ENOMEM;
 824
 825         tur->tur_update_params_buf_size = UPDATE_PARAMS_BUFFER_SIZE;
 826         return 0;
 827 }
 828
 829 /**
 830  * Extend update params
 831  *
 832  * Extend update_params to the new size in thandle_update_records.
 833  *
 834  * \param[in] tur       thandle_update_records where update_params will be
 835  *                      extended.
 836  * \retval              0 if extension succeeds.
 837  * \retval              negative errno if extension fails.
 838  */
 839 int tur_update_params_extend(struct thandle_update_records *tur,
 840                              size_t new_size)
 841 {
 842         struct update_params    *params;
 843
 844         OBD_ALLOC_LARGE(params, new_size);
 845         if (params == NULL)
 846                 return -ENOMEM;
 847
 848         if (tur->tur_update_params != NULL) {
 849                 memcpy(params, tur->tur_update_params,
 850                        tur->tur_update_params_buf_size);
 851                 OBD_FREE_LARGE(tur->tur_update_params,
 852                                tur->tur_update_params_buf_size);
 853         }
 854
 855         tur->tur_update_params = params;
 856         tur->tur_update_params_buf_size = new_size;
 857
 858         return 0;
 859 }
 860 EXPORT_SYMBOL(tur_update_params_extend);
 861
 862 /**
 863  * Check and prepare whether it needs to record update.
 864  *
 865  * Checks if the transaction needs to record updates, and if it
 866  * does, then initialize the update record buffer in the transaction.
 867  *
 868  * \param[in] env       execution environment
 869  * \param[in] th        transaction handle
 870  *
 871  * \retval              0 if updates recording succeeds.
 872  * \retval              negative errno if updates recording fails.
 873  */
 874 int check_and_prepare_update_record(const struct lu_env *env,
 875                                     struct thandle_update_records *tur)
 876 {
 877         struct llog_update_record       *lur;
 878         int rc;
 879
 880         if (tur->tur_update_records == NULL) {
 881                 rc = tur_update_records_create(tur);
 882                 if (rc < 0)
 883                         RETURN(rc);
 884         }
 885
 886         if (tur->tur_update_params == NULL) {
 887                 rc = tur_update_params_create(tur);
 888                 if (rc < 0)
 889                         RETURN(rc);
 890         }
 891
 892         lur = tur->tur_update_records;
 893         lur->lur_update_rec.ur_update_count = 0;
 894         lur->lur_update_rec.ur_param_count = 0;
 895         lur->lur_update_rec.ur_master_transno = 0;
 896         lur->lur_update_rec.ur_batchid = 0;
 897         lur->lur_update_rec.ur_flags = 0;
 898         lur->lur_hdr.lrh_len = LLOG_CHUNK_SIZE;
 899
 900         tur->tur_update_param_count = 0;
 901
 902         RETURN(0);
 903 }
 904
 905 static void update_key_fini(const struct lu_context *ctx,
 906                             struct lu_context_key *key, void *data)
 907 {
 908         struct update_thread_info *info = data;
 909         struct thandle_exec_args  *args = &info->uti_tea;
 910         int                       i;
 911
 912         for (i = 0; i < args->ta_alloc_args; i++) {
 913                 if (args->ta_args[i] != NULL)
 914                         OBD_FREE_PTR(args->ta_args[i]);
 915         }
 916
 917         if (args->ta_args != NULL)
 918                 OBD_FREE(args->ta_args, sizeof(args->ta_args[0]) *
 919                          args->ta_alloc_args);
 920
 921         if (info->uti_tur.tur_update_records != NULL)
 922                 OBD_FREE_LARGE(info->uti_tur.tur_update_records,
 923                                info->uti_tur.tur_update_records_buf_size);
 924         if (info->uti_tur.tur_update_params != NULL)
 925                 OBD_FREE_LARGE(info->uti_tur.tur_update_params,
 926                                info->uti_tur.tur_update_params_buf_size);
 927
 928         OBD_FREE_PTR(info);
 929 }
 930
 931 /* context key constructor/destructor: update_key_init, update_key_fini */
 932 LU_KEY_INIT(update, struct update_thread_info);
 933 /* context key: update_thread_key */
 934 LU_CONTEXT_KEY_DEFINE(update, LCT_MD_THREAD | LCT_MG_THREAD |
 935                               LCT_DT_THREAD | LCT_TX_HANDLE | LCT_LOCAL);
 936 EXPORT_SYMBOL(update_thread_key);
 937 LU_KEY_INIT_GENERIC(update);
 938
 939 void update_info_init(void)
 940 {
 941         update_key_init_generic(&update_thread_key, NULL);
 942         lu_context_key_register(&update_thread_key);
 943 }
 944
 945 void update_info_fini(void)
 946 {
 947         lu_context_key_degister(&update_thread_key);
 948 }