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) 2015, 2017, 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_obdo.h>
  45 #include <lustre_update.h>
  46 #include <obd.h>
  47 #include <obd_class.h>
  48
  49 #include "tgt_internal.h"
  50
  51 #define UPDATE_RECORDS_BUFFER_SIZE      8192
  52 #define UPDATE_PARAMS_BUFFER_SIZE       8192
  53 /**
  54  * Dump update record.
  55  *
  56  * Dump all of updates in the update_records, mostly for debugging purpose.
  57  *
  58  * \param[in] records   update records to be dumpped
  59  * \param[in] mask      debug level mask
  60  * \param[in] dump_params if dump all of updates the updates.
  61  *
  62  */
  63 void update_records_dump(const struct update_records *records,
  64                          unsigned int mask, bool dump_updates)
  65 {
  66         const struct update_ops *ops;
  67         const struct update_op  *op = NULL;
  68         struct update_params    *params = NULL;
  69         unsigned int            i;
  70
  71         CDEBUG(mask, "master transno = %llu batchid = %llu flags = %x"
  72                " ops = %d params = %d\n", records->ur_master_transno,
  73                records->ur_batchid, records->ur_flags, records->ur_update_count,
  74                records->ur_param_count);
  75
  76         if (records->ur_update_count == 0)
  77                 return;
  78
  79         if (!dump_updates)
  80                 return;
  81
  82         ops = &records->ur_ops;
  83         if (records->ur_param_count > 0)
  84                 params = update_records_get_params(records);
  85
  86         op = &ops->uops_op[0];
  87         for (i = 0; i < records->ur_update_count; i++,
  88                                   op = update_op_next_op(op)) {
  89                 unsigned int j;
  90
  91                 CDEBUG(mask, "update %dth "DFID" %s params_count = %hu\n", i,
  92                        PFID(&op->uop_fid), update_op_str(op->uop_type),
  93                        op->uop_param_count);
  94
  95                 if (params == NULL)
  96                         continue;
  97
  98                 for (j = 0;  j < op->uop_param_count; j++) {
  99                         struct object_update_param *param;
 100
 101                         param = update_params_get_param(params,
 102                                 (unsigned int)op->uop_params_off[j],
 103                                         records->ur_param_count);
 104
 105                         if (param == NULL)
 106                                 continue;
 107                         CDEBUG(mask, "param = %p %dth off = %hu size = %hu\n",
 108                                param, j, op->uop_params_off[j], param->oup_len);
 109                 }
 110         }
 111 }
 112
 113 /**
 114  * Pack parameters to update records
 115  *
 116  * Find and insert parameter to update records, if the parameter
 117  * already exists in \a params, then just return the offset of this
 118  * parameter, otherwise insert the parameter and return its offset
 119  *
 120  * \param[in] params    update params in which to insert parameter
 121  * \param[in] new_param parameters to be inserted.
 122  * \param[in] new_param_size    the size of \a new_param
 123  *
 124  * \retval              index inside \a params if parameter insertion
 125  *                      succeeds.
 126  * \retval              negative errno if it fails.
 127  */
 128 static unsigned int update_records_param_pack(struct update_params *params,
 129                                               const void *new_param,
 130                                               size_t new_param_size,
 131                                               unsigned int *param_count)
 132 {
 133         struct object_update_param      *param;
 134         unsigned int                    i;
 135
 136         for (i = 0; i < *param_count; i++) {
 137                 struct object_update_param *param;
 138
 139                 param = update_params_get_param(params, i, *param_count);
 140                 if ((new_param == NULL && param->oup_len == new_param_size) ||
 141                     (param->oup_len == new_param_size &&
 142                      memcmp(param->oup_buf, new_param, new_param_size) == 0))
 143                         /* Found the parameter and return its index */
 144                         return i;
 145         }
 146
 147         param = (struct object_update_param *)((char *)params +
 148                                 update_params_size(params, *param_count));
 149
 150         param->oup_len = new_param_size;
 151         if (new_param != NULL)
 152                 memcpy(param->oup_buf, new_param, new_param_size);
 153
 154         *param_count = *param_count + 1;
 155
 156         return *param_count - 1;
 157 }
 158
 159 /**
 160  * Pack update to update records
 161  *
 162  * Pack the update and its parameters to the update records. First it will
 163  * insert parameters, get the offset of these parameter, then fill the
 164  * update with these offset. If insertion exceed the maximum size of
 165  * current update records, it will return -E2BIG here, and the caller might
 166  * extend the update_record size \see lod_updates_pack.
 167  *
 168  * \param[in] env       execution environment
 169  * \param[in] fid       FID of the update.
 170  * \param[in] op_type   operation type of the update
 171  * \param[in] ops       ur_ops in update records
 172  * \param[in|out] op_count      pointer to the count of ops
 173  * \param[in|out] max_op_size maximum size of the update
 174  * \param[in] params    ur_params in update records
 175  * \param[in|out] param_count   pointer to the count of params
 176  * \param[in|out] max_param_size maximum size of the parameter
 177  * \param[in] param_bufs        buffers of parameters
 178  * \param[in] params_buf_count  the count of the parameter buffers
 179  * \param[in] param_size        sizes of parameters
 180  *
 181  * \retval              0 if packing succeeds
 182  * \retval              negative errno if packing fails
 183  */
 184 static int update_records_update_pack(const struct lu_env *env,
 185                                       const struct lu_fid *fid,
 186                                       enum update_type op_type,
 187                                       struct update_ops *ops,
 188                                       unsigned int *op_count,
 189                                       size_t *max_op_size,
 190                                       struct update_params *params,
 191                                       unsigned int *param_count,
 192                                       size_t *max_param_size,
 193                                       unsigned int param_bufs_count,
 194                                       const void **param_bufs,
 195                                       size_t *param_sizes)
 196 {
 197         struct update_op        *op;
 198         size_t                  total_param_sizes = 0;
 199         int                     index;
 200         unsigned int            i;
 201
 202         /* Check whether the packing exceeding the maximum update size */
 203         if (unlikely(*max_op_size < update_op_size(param_bufs_count))) {
 204                 CDEBUG(D_INFO, "max_op_size = %zu update_op = %zu\n",
 205                        *max_op_size, update_op_size(param_bufs_count));
 206                 *max_op_size = update_op_size(param_bufs_count);
 207                 return -E2BIG;
 208         }
 209
 210         for (i = 0; i < param_bufs_count; i++)
 211                 total_param_sizes +=
 212                         cfs_size_round(sizeof(struct object_update_param) +
 213                                        param_sizes[i]);
 214
 215         /* Check whether the packing exceeding the maximum parameter size */
 216         if (unlikely(*max_param_size < total_param_sizes)) {
 217                 CDEBUG(D_INFO, "max_param_size = %zu params size = %zu\n",
 218                        *max_param_size, total_param_sizes);
 219
 220                 *max_param_size = total_param_sizes;
 221                 return -E2BIG;
 222         }
 223
 224         op = update_ops_get_op(ops, *op_count, *op_count);
 225         op->uop_fid = *fid;
 226         op->uop_type = op_type;
 227         op->uop_param_count = param_bufs_count;
 228         for (i = 0; i < param_bufs_count; i++) {
 229                 index = update_records_param_pack(params, param_bufs[i],
 230                                                   param_sizes[i], param_count);
 231                 if (index < 0)
 232                         return index;
 233
 234                 CDEBUG(D_INFO, "%s %uth param offset = %d size = %zu\n",
 235                        update_op_str(op_type), i, index, param_sizes[i]);
 236
 237                 op->uop_params_off[i] = index;
 238         }
 239         CDEBUG(D_INFO, "%huth "DFID" %s param_count = %u\n",
 240                *op_count, PFID(fid), update_op_str(op_type), *param_count);
 241
 242         *op_count = *op_count + 1;
 243
 244         return 0;
 245 }
 246
 247 /**
 248  * Calculate update_records size
 249  *
 250  * Calculate update_records size by param_count and param_sizes array.
 251  *
 252  * \param[in] param_count       the count of parameters
 253  * \param[in] sizes             the size array of these parameters
 254  *
 255  * \retval                      the size of this update
 256  */
 257 static size_t update_records_update_size(__u32 param_count, size_t *sizes)
 258 {
 259         int i;
 260         size_t size;
 261
 262         /* Check whether the packing exceeding the maximum update size */
 263         size = update_op_size(param_count);
 264
 265         for (i = 0; i < param_count; i++)
 266                 size += cfs_size_round(sizeof(struct object_update_param) +
 267                                        sizes[i]);
 268
 269         return size;
 270 }
 271
 272 /**
 273  * Calculate create update size
 274  *
 275  * \param[in] env       execution environment
 276  * \param[in] ops       ur_ops in update records
 277  * \param[in] fid       FID of the object to be created
 278  * \param[in] attr      attribute of the object to be created
 279  * \param[in] hint      creation hint
 280  * \param[in] dof       creation format information
 281  *
 282  * \retval              size of create update.
 283  */
 284 size_t update_records_create_size(const struct lu_env *env,
 285                                   const struct lu_fid *fid,
 286                                   const struct lu_attr *attr,
 287                                   const struct dt_allocation_hint *hint,
 288                                   struct dt_object_format *dof)
 289 {
 290         size_t  sizes[2];
 291         int     param_count = 0;
 292
 293         if (attr != NULL) {
 294                 sizes[param_count] = sizeof(struct obdo);
 295                 param_count++;
 296         }
 297
 298         if (hint != NULL && hint->dah_parent != NULL) {
 299                 sizes[param_count] = sizeof(*fid);
 300                 param_count++;
 301         }
 302
 303         return update_records_update_size(param_count, sizes);
 304 }
 305 EXPORT_SYMBOL(update_records_create_size);
 306
 307 /**
 308  * Pack create update
 309  *
 310  * Pack create update into update records.
 311  *
 312  * \param[in] env       execution environment
 313  * \param[in] ops       ur_ops in update records
 314  * \param[in|out] op_count      pointer to the count of ops
 315  * \param[in|out] max_op_size maximum size of the update
 316  * \param[in] params    ur_params in update records
 317  * \param[in|out] param_count   pointer to the count of params
 318  * \param[in|out] max_param_size maximum size of the parameter
 319  * \param[in] fid       FID of the object to be created
 320  * \param[in] attr      attribute of the object to be created
 321  * \param[in] hint      creation hint
 322  * \param[in] dof       creation format information
 323  *
 324  * \retval              0 if packing succeeds.
 325  * \retval              negative errno if packing fails.
 326  */
 327 int update_records_create_pack(const struct lu_env *env,
 328                                struct update_ops *ops,
 329                                unsigned int *op_count,
 330                                size_t *max_ops_size,
 331                                struct update_params *params,
 332                                unsigned int *param_count,
 333                                size_t *max_param_size,
 334                                const struct lu_fid *fid,
 335                                const struct lu_attr *attr,
 336                                const struct dt_allocation_hint *hint,
 337                                struct dt_object_format *dof)
 338 {
 339         size_t                  sizes[2];
 340         const void              *bufs[2];
 341         int                     buf_count = 0;
 342         const struct lu_fid     *parent_fid = NULL;
 343         struct lu_fid           tmp_fid;
 344         int                     rc;
 345         struct obdo             *obdo;
 346
 347         if (attr != NULL) {
 348                 obdo = &update_env_info(env)->uti_obdo;
 349                 obdo->o_valid = 0;
 350                 obdo_from_la(obdo, attr, attr->la_valid);
 351                 bufs[buf_count] = obdo;
 352                 sizes[buf_count] = sizeof(*obdo);
 353                 buf_count++;
 354         }
 355
 356         if (hint != NULL && hint->dah_parent != NULL) {
 357                 parent_fid = lu_object_fid(&hint->dah_parent->do_lu);
 358                 fid_cpu_to_le(&tmp_fid, parent_fid);
 359                 bufs[buf_count] = &tmp_fid;
 360                 sizes[buf_count] = sizeof(tmp_fid);
 361                 buf_count++;
 362         }
 363
 364         rc = update_records_update_pack(env, fid, OUT_CREATE, ops, op_count,
 365                                         max_ops_size, params, param_count,
 366                                         max_param_size, buf_count, bufs, sizes);
 367         return rc;
 368 }
 369 EXPORT_SYMBOL(update_records_create_pack);
 370
 371 /**
 372  * Calculate attr set update size
 373  *
 374  * \param[in] env       execution environment
 375  * \param[in] ops       ur_ops in update records
 376  * \param[in] fid       FID of the object to set attr
 377  * \param[in] attr      attribute of attr set
 378  *
 379  * \retval              size of attr set update.
 380  */
 381 size_t update_records_attr_set_size(const struct lu_env *env,
 382                                     const struct lu_fid *fid,
 383                                     const struct lu_attr *attr)
 384 {
 385         size_t size = sizeof(struct obdo);
 386
 387         return update_records_update_size(1, &size);
 388 }
 389 EXPORT_SYMBOL(update_records_attr_set_size);
 390
 391 /**
 392  * Pack attr set update
 393  *
 394  * Pack attr_set update into update records.
 395  *
 396  * \param[in] env       execution environment
 397  * \param[in] ops       ur_ops in update records
 398  * \param[in|out] op_count pointer to the count of ops
 399  * \param[in|out] max_op_size maximum size of the update
 400  * \param[in] params    ur_params in update records
 401  * \param[in|out] param_count pointer to the count of params
 402  * \param[in|out] max_param_size maximum size of the parameter
 403  * \param[in] fid       FID of the object to set attr
 404  * \param[in] attr      attribute of attr set
 405  *
 406  * \retval              0 if packing succeeds.
 407  * \retval              negative errno if packing fails.
 408  */
 409 int update_records_attr_set_pack(const struct lu_env *env,
 410                                  struct update_ops *ops,
 411                                  unsigned int *op_count,
 412                                  size_t *max_ops_size,
 413                                  struct update_params *params,
 414                                  unsigned int *param_count,
 415                                  size_t *max_param_size,
 416                                  const struct lu_fid *fid,
 417                                  const struct lu_attr *attr)
 418 {
 419         struct obdo *obdo = &update_env_info(env)->uti_obdo;
 420         size_t size = sizeof(*obdo);
 421
 422         obdo->o_valid = 0;
 423         obdo_from_la(obdo, attr, attr->la_valid);
 424         return update_records_update_pack(env, fid, OUT_ATTR_SET, ops, op_count,
 425                                           max_ops_size, params, param_count,
 426                                           max_param_size, 1,
 427                                           (const void **)&obdo, &size);
 428 }
 429 EXPORT_SYMBOL(update_records_attr_set_pack);
 430
 431 /**
 432  * Calculate ref add update size
 433  *
 434  * \param[in] env       execution environment
 435  * \param[in] fid       FID of the object to add reference
 436  *
 437  * \retval              size of ref_add udpate.
 438  */
 439 size_t update_records_ref_add_size(const struct lu_env *env,
 440                                    const struct lu_fid *fid)
 441 {
 442         return update_records_update_size(0, NULL);
 443 }
 444 EXPORT_SYMBOL(update_records_ref_add_size);
 445
 446 /**
 447  * Pack ref add update
 448  *
 449  * Pack ref add update into update records.
 450  *
 451  * \param[in] env       execution environment
 452  * \param[in] ops       ur_ops in update records
 453  * \param[in|out] op_count pointer to the count of ops
 454  * \param[in|out] max_op_size maximum size of the update
 455  * \param[in] params    ur_params in update records
 456  * \param[in|out] param_count pointer to the count of params
 457  * \param[in|out] max_param_size maximum size of the parameter
 458  * \param[in] fid       FID of the object to add reference
 459  *
 460  * \retval              0 if packing succeeds.
 461  * \retval              negative errno if packing fails.
 462  */
 463 int update_records_ref_add_pack(const struct lu_env *env,
 464                                 struct update_ops *ops,
 465                                 unsigned int *op_count,
 466                                 size_t *max_ops_size,
 467                                 struct update_params *params,
 468                                 unsigned int *param_count,
 469                                 size_t *max_param_size,
 470                                 const struct lu_fid *fid)
 471 {
 472         return update_records_update_pack(env, fid, OUT_REF_ADD, ops, op_count,
 473                                           max_ops_size, params, param_count,
 474                                           max_param_size, 0, NULL, NULL);
 475 }
 476 EXPORT_SYMBOL(update_records_ref_add_pack);
 477
 478 /**
 479  * Pack noop update
 480  *
 481  * Pack no op update into update records. Note: no op means
 482  * the update does not need do anything, which is only used
 483  * in test case to verify large size record.
 484  *
 485  * \param[in] env       execution environment
 486  * \param[in] ops       ur_ops in update records
 487  * \param[in|out] op_count pointer to the count of ops
 488  * \param[in|out] max_op_size maximum size of the update
 489  * \param[in] params    ur_params in update records
 490  * \param[in|out] param_count pointer to the count of params
 491  * \param[in|out] max_param_size maximum size of the parameter
 492  * \param[in] fid       FID of the object to add reference
 493  *
 494  * \retval              0 if packing succeeds.
 495  * \retval              negative errno if packing fails.
 496  */
 497 int update_records_noop_pack(const struct lu_env *env,
 498                              struct update_ops *ops,
 499                              unsigned int *op_count,
 500                              size_t *max_ops_size,
 501                              struct update_params *params,
 502                              unsigned int *param_count,
 503                              size_t *max_param_size,
 504                              const struct lu_fid *fid)
 505 {
 506         return update_records_update_pack(env, fid, OUT_NOOP, ops, op_count,
 507                                           max_ops_size, params, param_count,
 508                                           max_param_size, 0, NULL, NULL);
 509 }
 510 EXPORT_SYMBOL(update_records_noop_pack);
 511
 512 /**
 513  * Calculate ref del update size
 514  *
 515  * \param[in] env       execution environment
 516  * \param[in] fid       FID of the object to delete reference
 517  *
 518  * \retval              size of ref_del update.
 519  */
 520 size_t update_records_ref_del_size(const struct lu_env *env,
 521                                    const struct lu_fid *fid)
 522 {
 523         return update_records_update_size(0, NULL);
 524 }
 525 EXPORT_SYMBOL(update_records_ref_del_size);
 526
 527 /**
 528  * Pack ref del update
 529  *
 530  * Pack ref del 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] 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 delete reference
 540  *
 541  * \retval              0 if packing succeeds.
 542  * \retval              negative errno if packing fails.
 543  */
 544 int update_records_ref_del_pack(const struct lu_env *env,
 545                                 struct update_ops *ops,
 546                                 unsigned int *op_count,
 547                                 size_t *max_ops_size,
 548                                 struct update_params *params,
 549                                 unsigned int *param_count,
 550                                 size_t *max_param_size,
 551                                 const struct lu_fid *fid)
 552 {
 553         return update_records_update_pack(env, fid, OUT_REF_DEL, ops, op_count,
 554                                           max_ops_size, params, param_count,
 555                                           max_param_size, 0, NULL, NULL);
 556 }
 557 EXPORT_SYMBOL(update_records_ref_del_pack);
 558
 559 /**
 560  * Calculate object destroy update size
 561  *
 562  * \param[in] env       execution environment
 563  * \param[in] fid       FID of the object to delete reference
 564  *
 565  * \retval              size of object destroy update.
 566  */
 567 size_t update_records_destroy_size(const struct lu_env *env,
 568                                           const struct lu_fid *fid)
 569 {
 570         return update_records_update_size(0, NULL);
 571 }
 572 EXPORT_SYMBOL(update_records_destroy_size);
 573
 574 /**
 575  * Pack object destroy update
 576  *
 577  * Pack object destroy update into update records.
 578  *
 579  * \param[in] env       execution environment
 580  * \param[in] ops       ur_ops in update records
 581  * \param[in|out] op_count pointer to the count of ops
 582  * \param[in|out] max_op_size maximum size of the update
 583  * \param[in] params    ur_params in update records
 584  * \param[in|out] param_count   pointer to the count of params
 585  * \param[in|out] max_param_size maximum size of the parameter
 586  * \param[in] fid       FID of the object to delete reference
 587  *
 588  * \retval              0 if packing succeeds.
 589  * \retval              negative errno if packing fails.
 590  */
 591 int update_records_destroy_pack(const struct lu_env *env,
 592                                        struct update_ops *ops,
 593                                        unsigned int *op_count,
 594                                        size_t *max_ops_size,
 595                                        struct update_params *params,
 596                                        unsigned int *param_count,
 597                                        size_t *max_param_size,
 598                                        const struct lu_fid *fid)
 599 {
 600         return update_records_update_pack(env, fid, OUT_DESTROY, ops, op_count,
 601                                           max_ops_size, params, param_count,
 602                                           max_param_size, 0, NULL, NULL);
 603 }
 604 EXPORT_SYMBOL(update_records_destroy_pack);
 605
 606 /**
 607  * Calculate index insert update size
 608  *
 609  * \param[in] env       execution environment
 610  * \param[in] fid       FID of the object to insert index
 611  * \param[in] rec       record of insertion
 612  * \param[in] key       key of insertion
 613  *
 614  * \retval              the size of index insert update.
 615  */
 616 size_t update_records_index_insert_size(const struct lu_env *env,
 617                                         const struct lu_fid *fid,
 618                                         const struct dt_rec *rec,
 619                                         const struct dt_key *key)
 620 {
 621         size_t                     sizes[3] = { strlen((const char *)key) + 1,
 622                                                 sizeof(struct lu_fid),
 623                                                 sizeof(__u32) };
 624         return update_records_update_size(3, sizes);
 625 }
 626 EXPORT_SYMBOL(update_records_index_insert_size);
 627
 628 /**
 629  * Pack index insert update
 630  *
 631  * Pack index insert update into update records.
 632  *
 633  * \param[in] env       execution environment
 634  * \param[in] ops       ur_ops in update records
 635  * \param[in] op_count  pointer to the count of ops
 636  * \param[in|out] max_op_size maximum size of the update
 637  * \param[in] params    ur_params in update records
 638  * \param[in] param_count       pointer to the count of params
 639  * \param[in|out] max_param_size maximum size of the parameter
 640  * \param[in] fid       FID of the object to insert index
 641  * \param[in] rec       record of insertion
 642  * \param[in] key       key of insertion
 643  *
 644  * \retval              0 if packing succeeds.
 645  * \retval              negative errno if packing fails.
 646  */
 647 int update_records_index_insert_pack(const struct lu_env *env,
 648                                      struct update_ops *ops,
 649                                      unsigned int *op_count,
 650                                      size_t *max_ops_size,
 651                                      struct update_params *params,
 652                                      unsigned int *param_count,
 653                                      size_t *max_param_size,
 654                                      const struct lu_fid *fid,
 655                                      const struct dt_rec *rec,
 656                                      const struct dt_key *key)
 657 {
 658         struct dt_insert_rec       *rec1 = (struct dt_insert_rec *)rec;
 659         struct lu_fid              rec_fid;
 660         __u32                      type = cpu_to_le32(rec1->rec_type);
 661         size_t                     sizes[3] = { strlen((const char *)key) + 1,
 662                                                 sizeof(rec_fid),
 663                                                 sizeof(type) };
 664         const void                 *bufs[3] = { key,
 665                                                 &rec_fid,
 666                                                 &type };
 667
 668         fid_cpu_to_le(&rec_fid, rec1->rec_fid);
 669
 670         return update_records_update_pack(env, fid, OUT_INDEX_INSERT, ops,
 671                                           op_count, max_ops_size, params,
 672                                           param_count, max_param_size,
 673                                           3, bufs, sizes);
 674 }
 675 EXPORT_SYMBOL(update_records_index_insert_pack);
 676
 677 /**
 678  * Calculate index delete update size
 679  *
 680  * \param[in] env       execution environment
 681  * \param[in] fid       FID of the object to delete index
 682  * \param[in] key       key of deletion
 683  *
 684  * \retval              the size of index delete update
 685  */
 686 size_t update_records_index_delete_size(const struct lu_env *env,
 687                                         const struct lu_fid *fid,
 688                                         const struct dt_key *key)
 689 {
 690         size_t size = strlen((const char *)key) + 1;
 691
 692         return update_records_update_size(1, &size);
 693 }
 694 EXPORT_SYMBOL(update_records_index_delete_size);
 695
 696 /**
 697  * Pack index delete update
 698  *
 699  * Pack index delete update into update records.
 700  *
 701  * \param[in] env       execution environment
 702  * \param[in] ops       ur_ops in update records
 703  * \param[in|out] op_count      pointer to the count of ops
 704  * \param[in|out] max_op_size maximum size of the update
 705  * \param[in] params    ur_params in update records
 706  * \param[in|ount] param_count  pointer to the count of params
 707  * \param[in|out] max_param_size maximum size of the parameter
 708  * \param[in] fid       FID of the object to delete index
 709  * \param[in] key       key of deletion
 710  *
 711  * \retval              0 if packing succeeds.
 712  * \retval              negative errno if packing fails.
 713  */
 714 int update_records_index_delete_pack(const struct lu_env *env,
 715                                      struct update_ops *ops,
 716                                      unsigned int *op_count,
 717                                      size_t *max_ops_size,
 718                                      struct update_params *params,
 719                                      unsigned int *param_count,
 720                                      size_t *max_param_size,
 721                                      const struct lu_fid *fid,
 722                                      const struct dt_key *key)
 723 {
 724         size_t size = strlen((const char *)key) + 1;
 725
 726         return update_records_update_pack(env, fid, OUT_INDEX_DELETE, ops,
 727                                           op_count, max_ops_size, params,
 728                                           param_count, max_param_size,
 729                                           1, (const void **)&key, &size);
 730 }
 731 EXPORT_SYMBOL(update_records_index_delete_pack);
 732
 733 /**
 734  * Calculate xattr set size
 735  *
 736  * \param[in] env       execution environment
 737  * \param[in] fid       FID of the object to set xattr
 738  * \param[in] buf       xattr to be set
 739  * \param[in] name      name of the xattr
 740  * \param[in] flag      flag for setting xattr
 741  *
 742  * \retval              size of xattr set update.
 743  */
 744 size_t update_records_xattr_set_size(const struct lu_env *env,
 745                                      const struct lu_fid *fid,
 746                                      const struct lu_buf *buf,
 747                                      const char *name, __u32 flag)
 748 {
 749         size_t  sizes[3] = {strlen(name) + 1, buf->lb_len, sizeof(flag)};
 750
 751         return update_records_update_size(3, sizes);
 752 }
 753 EXPORT_SYMBOL(update_records_xattr_set_size);
 754
 755 /**
 756  * Pack xattr set update
 757  *
 758  * Pack xattr set update into update records.
 759  *
 760  * \param[in] env       execution environment
 761  * \param[in] ops       ur_ops in update records
 762  * \param[in|out] op_count      pointer to the count of ops
 763  * \param[in|out] max_op_size maximum size of the update
 764  * \param[in] params    ur_params in update records
 765  * \param[in|out] param_count   pointer to the count of params
 766  * \param[in|out] max_param_size maximum size of the parameter
 767  * \param[in] fid       FID of the object to set xattr
 768  * \param[in] buf       xattr to be set
 769  * \param[in] name      name of the xattr
 770  * \param[in] flag      flag for setting xattr
 771  *
 772  * \retval              0 if packing succeeds.
 773  * \retval              negative errno if packing fails.
 774  */
 775 int update_records_xattr_set_pack(const struct lu_env *env,
 776                                   struct update_ops *ops,
 777                                   unsigned int *op_count,
 778                                   size_t *max_ops_size,
 779                                   struct update_params *params,
 780                                   unsigned int *param_count,
 781                                   size_t *max_param_size,
 782                                   const struct lu_fid *fid,
 783                                   const struct lu_buf *buf, const char *name,
 784                                   __u32 flag)
 785 {
 786         size_t  sizes[3] = {strlen(name) + 1, buf->lb_len, sizeof(flag)};
 787         const void *bufs[3] = {name, buf->lb_buf, &flag};
 788
 789         flag = cpu_to_le32(flag);
 790
 791         return update_records_update_pack(env, fid, OUT_XATTR_SET, ops,
 792                                           op_count, max_ops_size, params,
 793                                           param_count, max_param_size,
 794                                           3, bufs, sizes);
 795 }
 796 EXPORT_SYMBOL(update_records_xattr_set_pack);
 797
 798 /**
 799  * Calculate xattr delete update size.
 800  *
 801  * \param[in] env       execution environment
 802  * \param[in] fid       FID of the object to delete xattr
 803  * \param[in] name      name of the xattr
 804  *
 805  * \retval              size of xattr delet updatee.
 806  */
 807 size_t update_records_xattr_del_size(const struct lu_env *env,
 808                                      const struct lu_fid *fid,
 809                                      const char *name)
 810 {
 811         size_t  size = strlen(name) + 1;
 812
 813         return update_records_update_size(1, &size);
 814 }
 815 EXPORT_SYMBOL(update_records_xattr_del_size);
 816
 817 /**
 818  * Pack xattr delete update
 819  *
 820  * Pack xattr delete update into update records.
 821  *
 822  * \param[in] env       execution environment
 823  * \param[in] ops       ur_ops in update records
 824  * \param[in|out] op_count      pointer to the count of ops
 825  * \param[in|out] max_op_size maximum size of the update
 826  * \param[in] params    ur_params in update records
 827  * \param[in|out] param_count   pointer to the count of params
 828  * \param[in|out] max_param_size maximum size of the parameter
 829  * \param[in] fid       FID of the object to delete xattr
 830  * \param[in] name      name of the xattr
 831  *
 832  * \retval              0 if packing succeeds.
 833  * \retval              negative errno if packing fails.
 834  */
 835 int update_records_xattr_del_pack(const struct lu_env *env,
 836                                   struct update_ops *ops,
 837                                   unsigned int *op_count,
 838                                   size_t *max_ops_size,
 839                                   struct update_params *params,
 840                                   unsigned int *param_count,
 841                                   size_t *max_param_size,
 842                                   const struct lu_fid *fid,
 843                                   const char *name)
 844 {
 845         size_t  size = strlen(name) + 1;
 846
 847         return update_records_update_pack(env, fid, OUT_XATTR_DEL, ops,
 848                                           op_count, max_ops_size, params,
 849                                           param_count, max_param_size,
 850                                           1, (const void **)&name, &size);
 851 }
 852 EXPORT_SYMBOL(update_records_xattr_del_pack);
 853
 854 /**
 855  * Calculate write update size
 856  *
 857  * \param[in] env       execution environment
 858  * \param[in] fid       FID of the object to write into
 859  * \param[in] buf       buffer to write which includes an embedded size field
 860  * \param[in] pos       offet in the object to start writing at
 861  *
 862  * \retval              size of write udpate.
 863  */
 864 size_t update_records_write_size(const struct lu_env *env,
 865                                  const struct lu_fid *fid,
 866                                  const struct lu_buf *buf,
 867                                  __u64 pos)
 868 {
 869         size_t  sizes[2] = {buf->lb_len, sizeof(pos)};
 870
 871         return update_records_update_size(2, sizes);
 872 }
 873 EXPORT_SYMBOL(update_records_write_size);
 874
 875 /**
 876  * Pack write update
 877  *
 878  * Pack write update into update records.
 879  *
 880  * \param[in] env       execution environment
 881  * \param[in] ops       ur_ops in update records
 882  * \param[in|out] op_count      pointer to the count of ops
 883  * \param[in|out] max_op_size maximum size of the update
 884  * \param[in] params    ur_params in update records
 885  * \param[in|out] param_count   pointer to the count of params
 886  * \param[in|out] max_param_size maximum size of the parameter
 887  * \param[in] fid       FID of the object to write into
 888  * \param[in] buf       buffer to write which includes an embedded size field
 889  * \param[in] pos       offet in the object to start writing at
 890  *
 891  * \retval              0 if packing succeeds.
 892  * \retval              negative errno if packing fails.
 893  */
 894 int update_records_write_pack(const struct lu_env *env,
 895                               struct update_ops *ops,
 896                               unsigned int *op_count,
 897                               size_t *max_ops_size,
 898                               struct update_params *params,
 899                               unsigned int *param_count,
 900                               size_t *max_param_size,
 901                               const struct lu_fid *fid,
 902                               const struct lu_buf *buf,
 903                               __u64 pos)
 904 {
 905         size_t          sizes[2] = {buf->lb_len, sizeof(pos)};
 906         const void      *bufs[2] = {buf->lb_buf, &pos};
 907
 908         pos = cpu_to_le64(pos);
 909
 910         return update_records_update_pack(env, fid, OUT_WRITE, ops,
 911                                           op_count, max_ops_size, params,
 912                                           param_count, max_param_size,
 913                                           2, bufs, sizes);
 914 }
 915 EXPORT_SYMBOL(update_records_write_pack);
 916
 917 /**
 918  * Calculate size of punch update.
 919  *
 920  * \param[in] env       execution environment
 921  * \param[in] fid       FID of the object to write into
 922  * \param[in] start     start offset of punch
 923  * \param[in] end       end offet of punch
 924  *
 925  * \retval              size of update punch.
 926  */
 927 size_t update_records_punch_size(const struct lu_env *env,
 928                                  const struct lu_fid *fid,
 929                                  __u64 start, __u64 end)
 930 {
 931         size_t  sizes[2] = {sizeof(start), sizeof(end)};
 932
 933         return update_records_update_size(2, sizes);
 934 }
 935 EXPORT_SYMBOL(update_records_punch_size);
 936
 937 /**
 938  * Pack punch
 939  *
 940  * Pack punch update into update records.
 941  *
 942  * \param[in] env       execution environment
 943  * \param[in] ops       ur_ops in update records
 944  * \param[in|out] op_count      pointer to the count of ops
 945  * \param[in|out] max_op_size maximum size of the update
 946  * \param[in] params    ur_params in update records
 947  * \param[in|out] param_count   pointer to the count of params
 948  * \param[in|out] max_param_size maximum size of the parameter
 949  * \param[in] fid       FID of the object to write into
 950  * \param[in] start     start offset of punch
 951  * \param[in] end       end offet of punch
 952  *
 953  * \retval              0 if packing succeeds.
 954  * \retval              negative errno if packing fails.
 955  */
 956 int update_records_punch_pack(const struct lu_env *env,
 957                               struct update_ops *ops,
 958                               unsigned int *op_count,
 959                               size_t *max_ops_size,
 960                               struct update_params *params,
 961                               unsigned int *param_count,
 962                               size_t *max_param_size,
 963                               const struct lu_fid *fid,
 964                               __u64 start, __u64 end)
 965 {
 966         size_t          sizes[2] = {sizeof(start), sizeof(end)};
 967         const void      *bufs[2] = {&start, &end};
 968
 969         start = cpu_to_le64(start);
 970         end = cpu_to_le64(end);
 971
 972         return update_records_update_pack(env, fid, OUT_PUNCH, ops, op_count,
 973                                           max_ops_size, params, param_count,
 974                                           max_param_size, 2, bufs, sizes);
 975 }
 976 EXPORT_SYMBOL(update_records_punch_pack);
 977
 978 /**
 979  * Create update records in thandle_update_records
 980  *
 981  * Allocate update_records for thandle_update_records, the initial size
 982  * will be 4KB.
 983  *
 984  * \param[in] tur       thandle_update_records where update_records will be
 985  *                      allocated
 986  * \retval              0 if allocation succeeds.
 987  * \retval              negative errno if allocation fails.
 988  */
 989 static int tur_update_records_create(struct thandle_update_records *tur)
 990 {
 991         if (tur->tur_update_records != NULL)
 992                 return 0;
 993
 994         OBD_ALLOC_LARGE(tur->tur_update_records,
 995                         UPDATE_RECORDS_BUFFER_SIZE);
 996
 997         if (tur->tur_update_records == NULL)
 998                 return -ENOMEM;
 999
1000         tur->tur_update_records_buf_size = UPDATE_RECORDS_BUFFER_SIZE;
1001
1002         return 0;
1003 }
1004
1005 /**
1006  * Extend update records
1007  *
1008  * Extend update_records to the new size in thandle_update_records.
1009  *
1010  * \param[in] tur       thandle_update_records where update_records will be
1011  *                      extended.
1012  * \retval              0 if extension succeeds.
1013  * \retval              negative errno if extension fails.
1014  */
1015 int tur_update_records_extend(struct thandle_update_records *tur,
1016                               size_t new_size)
1017 {
1018         struct llog_update_record       *record;
1019
1020         OBD_ALLOC_LARGE(record, new_size);
1021         if (record == NULL)
1022                 return -ENOMEM;
1023
1024         if (tur->tur_update_records != NULL) {
1025                 memcpy(record, tur->tur_update_records,
1026                        tur->tur_update_records_buf_size);
1027                 OBD_FREE_LARGE(tur->tur_update_records,
1028                                tur->tur_update_records_buf_size);
1029         }
1030
1031         tur->tur_update_records = record;
1032         tur->tur_update_records_buf_size = new_size;
1033
1034         return 0;
1035 }
1036 EXPORT_SYMBOL(tur_update_records_extend);
1037
1038 /**
1039  * Extend update records
1040  *
1041  * Extend update records in thandle to make sure it is able to hold
1042  * the update with certain update_op and params size.
1043  *
1044  * \param [in] tur      thandle_update_records to be extend
1045  * \param [in] new_op_size update_op size of the update record
1046  * \param [in] new_param_size params size of the update record
1047  *
1048  * \retval              0 if the update_records is being extended.
1049  * \retval              negative errno if the update_records is not being
1050  *                      extended.
1051  */
1052 int tur_update_extend(struct thandle_update_records *tur,
1053                       size_t new_op_size, size_t new_param_size)
1054 {
1055         size_t record_size;
1056         size_t params_size;
1057         size_t extend_size;
1058         int rc;
1059         ENTRY;
1060
1061         record_size = llog_update_record_size(tur->tur_update_records);
1062         /* extend update records buffer */
1063         if (new_op_size >= (tur->tur_update_records_buf_size - record_size)) {
1064                 extend_size = round_up(new_op_size, UPDATE_RECORDS_BUFFER_SIZE);
1065                 rc = tur_update_records_extend(tur,
1066                                 tur->tur_update_records_buf_size +
1067                                 extend_size);
1068                 if (rc != 0)
1069                         RETURN(rc);
1070         }
1071
1072         /* extend parameters buffer */
1073         params_size = update_params_size(tur->tur_update_params,
1074                                          tur->tur_update_param_count);
1075         if (new_param_size >= (tur->tur_update_params_buf_size -
1076                               params_size)) {
1077                 extend_size = round_up(new_param_size,
1078                                        UPDATE_PARAMS_BUFFER_SIZE);
1079                 rc = tur_update_params_extend(tur,
1080                                 tur->tur_update_params_buf_size +
1081                                 extend_size);
1082                 if (rc != 0)
1083                         RETURN(rc);
1084         }
1085
1086         RETURN(0);
1087 }
1088 EXPORT_SYMBOL(tur_update_extend);
1089
1090 /**
1091  * Create update params in thandle_update_records
1092  *
1093  * Allocate update_params for thandle_update_records, the initial size
1094  * will be 4KB.
1095  *
1096  * \param[in] tur       thandle_update_records where update_params will be
1097  *                      allocated
1098  * \retval              0 if allocation succeeds.
1099  * \retval              negative errno if allocation fails.
1100  */
1101 static int tur_update_params_create(struct thandle_update_records *tur)
1102 {
1103         if (tur->tur_update_params != NULL)
1104                 return 0;
1105
1106         OBD_ALLOC_LARGE(tur->tur_update_params, UPDATE_PARAMS_BUFFER_SIZE);
1107         if (tur->tur_update_params == NULL)
1108                 return -ENOMEM;
1109
1110         tur->tur_update_params_buf_size = UPDATE_PARAMS_BUFFER_SIZE;
1111         return 0;
1112 }
1113
1114 /**
1115  * Extend update params
1116  *
1117  * Extend update_params to the new size in thandle_update_records.
1118  *
1119  * \param[in] tur       thandle_update_records where update_params will be
1120  *                      extended.
1121  * \retval              0 if extension succeeds.
1122  * \retval              negative errno if extension fails.
1123  */
1124 int tur_update_params_extend(struct thandle_update_records *tur,
1125                              size_t new_size)
1126 {
1127         struct update_params    *params;
1128
1129         OBD_ALLOC_LARGE(params, new_size);
1130         if (params == NULL)
1131                 return -ENOMEM;
1132
1133         if (tur->tur_update_params != NULL) {
1134                 memcpy(params, tur->tur_update_params,
1135                        tur->tur_update_params_buf_size);
1136                 OBD_FREE_LARGE(tur->tur_update_params,
1137                                tur->tur_update_params_buf_size);
1138         }
1139
1140         tur->tur_update_params = params;
1141         tur->tur_update_params_buf_size = new_size;
1142
1143         return 0;
1144 }
1145 EXPORT_SYMBOL(tur_update_params_extend);
1146
1147 /**
1148  * Check and prepare whether it needs to record update.
1149  *
1150  * Checks if the transaction needs to record updates, and if it
1151  * does, then initialize the update record buffer in the transaction.
1152  *
1153  * \param[in] env       execution environment
1154  * \param[in] th        transaction handle
1155  *
1156  * \retval              0 if updates recording succeeds.
1157  * \retval              negative errno if updates recording fails.
1158  */
1159 int check_and_prepare_update_record(const struct lu_env *env,
1160                                     struct thandle_update_records *tur)
1161 {
1162         struct llog_update_record       *lur;
1163         int rc;
1164
1165         if (tur->tur_update_records == NULL) {
1166                 rc = tur_update_records_create(tur);
1167                 if (rc < 0)
1168                         RETURN(rc);
1169         }
1170
1171         if (tur->tur_update_params == NULL) {
1172                 rc = tur_update_params_create(tur);
1173                 if (rc < 0)
1174                         RETURN(rc);
1175         }
1176
1177         lur = tur->tur_update_records;
1178         lur->lur_update_rec.ur_update_count = 0;
1179         lur->lur_update_rec.ur_param_count = 0;
1180         lur->lur_update_rec.ur_master_transno = 0;
1181         lur->lur_update_rec.ur_batchid = 0;
1182         lur->lur_update_rec.ur_flags = 0;
1183         lur->lur_hdr.lrh_len = LLOG_MIN_CHUNK_SIZE;
1184
1185         tur->tur_update_param_count = 0;
1186
1187         RETURN(0);
1188 }
1189
1190 static void update_key_fini(const struct lu_context *ctx,
1191                             struct lu_context_key *key, void *data)
1192 {
1193         struct update_thread_info *info = data;
1194         struct thandle_exec_args  *args = &info->uti_tea;
1195         int                       i;
1196
1197         for (i = 0; i < args->ta_alloc_args; i++) {
1198                 if (args->ta_args[i] != NULL)
1199                         OBD_FREE_PTR(args->ta_args[i]);
1200         }
1201
1202         if (args->ta_args != NULL)
1203                 OBD_FREE(args->ta_args, sizeof(args->ta_args[0]) *
1204                          args->ta_alloc_args);
1205
1206         if (info->uti_tur.tur_update_records != NULL)
1207                 OBD_FREE_LARGE(info->uti_tur.tur_update_records,
1208                                info->uti_tur.tur_update_records_buf_size);
1209         if (info->uti_tur.tur_update_params != NULL)
1210                 OBD_FREE_LARGE(info->uti_tur.tur_update_params,
1211                                info->uti_tur.tur_update_params_buf_size);
1212
1213         OBD_FREE_PTR(info);
1214 }
1215
1216 /* context key constructor/destructor: update_key_init, update_key_fini */
1217 LU_KEY_INIT(update, struct update_thread_info);
1218 /* context key: update_thread_key */
1219 LU_CONTEXT_KEY_DEFINE(update, LCT_MD_THREAD | LCT_MG_THREAD |
1220                               LCT_DT_THREAD | LCT_LOCAL);
1221 EXPORT_SYMBOL(update_thread_key);
1222 LU_KEY_INIT_GENERIC(update);
1223
1224 void update_info_init(void)
1225 {
1226         update_key_init_generic(&update_thread_key, NULL);
1227         lu_context_key_register(&update_thread_key);
1228 }
1229
1230 void update_info_fini(void)
1231 {
1232         lu_context_key_degister(&update_thread_key);
1233 }