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