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, 2015, 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                 lustre_set_wire_obdo(NULL, obdo, obdo);
 352                 bufs[buf_count] = obdo;
 353                 sizes[buf_count] = sizeof(*obdo);
 354                 buf_count++;
 355         }
 356
 357         if (hint != NULL && hint->dah_parent != NULL) {
 358                 parent_fid = lu_object_fid(&hint->dah_parent->do_lu);
 359                 fid_cpu_to_le(&tmp_fid, parent_fid);
 360                 bufs[buf_count] = &tmp_fid;
 361                 sizes[buf_count] = sizeof(tmp_fid);
 362                 buf_count++;
 363         }
 364
 365         rc = update_records_update_pack(env, fid, OUT_CREATE, ops, op_count,
 366                                         max_ops_size, params, param_count,
 367                                         max_param_size, buf_count, bufs, sizes);
 368         return rc;
 369 }
 370 EXPORT_SYMBOL(update_records_create_pack);
 371
 372 /**
 373  * Calculate attr set update size
 374  *
 375  * \param[in] env       execution environment
 376  * \param[in] ops       ur_ops in update records
 377  * \param[in] fid       FID of the object to set attr
 378  * \param[in] attr      attribute of attr set
 379  *
 380  * \retval              size of attr set update.
 381  */
 382 size_t update_records_attr_set_size(const struct lu_env *env,
 383                                     const struct lu_fid *fid,
 384                                     const struct lu_attr *attr)
 385 {
 386         size_t size = sizeof(struct obdo);
 387
 388         return update_records_update_size(1, &size);
 389 }
 390 EXPORT_SYMBOL(update_records_attr_set_size);
 391
 392 /**
 393  * Pack attr set update
 394  *
 395  * Pack attr_set update into update records.
 396  *
 397  * \param[in] env       execution environment
 398  * \param[in] ops       ur_ops in update records
 399  * \param[in|out] op_count pointer to the count of ops
 400  * \param[in|out] max_op_size maximum size of the update
 401  * \param[in] params    ur_params in update records
 402  * \param[in|out] param_count pointer to the count of params
 403  * \param[in|out] max_param_size maximum size of the parameter
 404  * \param[in] fid       FID of the object to set attr
 405  * \param[in] attr      attribute of attr set
 406  *
 407  * \retval              0 if packing succeeds.
 408  * \retval              negative errno if packing fails.
 409  */
 410 int update_records_attr_set_pack(const struct lu_env *env,
 411                                  struct update_ops *ops,
 412                                  unsigned int *op_count,
 413                                  size_t *max_ops_size,
 414                                  struct update_params *params,
 415                                  unsigned int *param_count,
 416                                  size_t *max_param_size,
 417                                  const struct lu_fid *fid,
 418                                  const struct lu_attr *attr)
 419 {
 420         struct obdo *obdo = &update_env_info(env)->uti_obdo;
 421         size_t size = sizeof(*obdo);
 422
 423         obdo->o_valid = 0;
 424         obdo_from_la(obdo, attr, attr->la_valid);
 425         lustre_set_wire_obdo(NULL, obdo, obdo);
 426         return update_records_update_pack(env, fid, OUT_ATTR_SET, ops, op_count,
 427                                           max_ops_size, params, param_count,
 428                                           max_param_size, 1,
 429                                           (const void **)&obdo, &size);
 430 }
 431 EXPORT_SYMBOL(update_records_attr_set_pack);
 432
 433 /**
 434  * Calculate ref add update size
 435  *
 436  * \param[in] env       execution environment
 437  * \param[in] fid       FID of the object to add reference
 438  *
 439  * \retval              size of ref_add udpate.
 440  */
 441 size_t update_records_ref_add_size(const struct lu_env *env,
 442                                    const struct lu_fid *fid)
 443 {
 444         return update_records_update_size(0, NULL);
 445 }
 446 EXPORT_SYMBOL(update_records_ref_add_size);
 447
 448 /**
 449  * Pack ref add update
 450  *
 451  * Pack ref add update into update records.
 452  *
 453  * \param[in] env       execution environment
 454  * \param[in] ops       ur_ops in update records
 455  * \param[in|out] op_count pointer to the count of ops
 456  * \param[in|out] max_op_size maximum size of the update
 457  * \param[in] params    ur_params in update records
 458  * \param[in|out] param_count pointer to the count of params
 459  * \param[in|out] max_param_size maximum size of the parameter
 460  * \param[in] fid       FID of the object to add reference
 461  *
 462  * \retval              0 if packing succeeds.
 463  * \retval              negative errno if packing fails.
 464  */
 465 int update_records_ref_add_pack(const struct lu_env *env,
 466                                 struct update_ops *ops,
 467                                 unsigned int *op_count,
 468                                 size_t *max_ops_size,
 469                                 struct update_params *params,
 470                                 unsigned int *param_count,
 471                                 size_t *max_param_size,
 472                                 const struct lu_fid *fid)
 473 {
 474         return update_records_update_pack(env, fid, OUT_REF_ADD, ops, op_count,
 475                                           max_ops_size, params, param_count,
 476                                           max_param_size, 0, NULL, NULL);
 477 }
 478 EXPORT_SYMBOL(update_records_ref_add_pack);
 479
 480 /**
 481  * Pack noop update
 482  *
 483  * Pack no op update into update records. Note: no op means
 484  * the update does not need do anything, which is only used
 485  * in test case to verify large size record.
 486  *
 487  * \param[in] env       execution environment
 488  * \param[in] ops       ur_ops in update records
 489  * \param[in|out] op_count pointer to the count of ops
 490  * \param[in|out] max_op_size maximum size of the update
 491  * \param[in] params    ur_params in update records
 492  * \param[in|out] param_count pointer to the count of params
 493  * \param[in|out] max_param_size maximum size of the parameter
 494  * \param[in] fid       FID of the object to add reference
 495  *
 496  * \retval              0 if packing succeeds.
 497  * \retval              negative errno if packing fails.
 498  */
 499 int update_records_noop_pack(const struct lu_env *env,
 500                              struct update_ops *ops,
 501                              unsigned int *op_count,
 502                              size_t *max_ops_size,
 503                              struct update_params *params,
 504                              unsigned int *param_count,
 505                              size_t *max_param_size,
 506                              const struct lu_fid *fid)
 507 {
 508         return update_records_update_pack(env, fid, OUT_NOOP, ops, op_count,
 509                                           max_ops_size, params, param_count,
 510                                           max_param_size, 0, NULL, NULL);
 511 }
 512 EXPORT_SYMBOL(update_records_noop_pack);
 513
 514 /**
 515  * Calculate ref del update size
 516  *
 517  * \param[in] env       execution environment
 518  * \param[in] fid       FID of the object to delete reference
 519  *
 520  * \retval              size of ref_del update.
 521  */
 522 size_t update_records_ref_del_size(const struct lu_env *env,
 523                                    const struct lu_fid *fid)
 524 {
 525         return update_records_update_size(0, NULL);
 526 }
 527 EXPORT_SYMBOL(update_records_ref_del_size);
 528
 529 /**
 530  * Pack ref del update
 531  *
 532  * Pack ref del update into update records.
 533  *
 534  * \param[in] env       execution environment
 535  * \param[in] ops       ur_ops in update records
 536  * \param[in|out] op_count      pointer to the count of ops
 537  * \param[in|out] max_op_size maximum size of the update
 538  * \param[in] params    ur_params in update records
 539  * \param[in] param_count       pointer to the count of params
 540  * \param[in|out] max_param_size maximum size of the parameter
 541  * \param[in] fid       FID of the object to delete reference
 542  *
 543  * \retval              0 if packing succeeds.
 544  * \retval              negative errno if packing fails.
 545  */
 546 int update_records_ref_del_pack(const struct lu_env *env,
 547                                 struct update_ops *ops,
 548                                 unsigned int *op_count,
 549                                 size_t *max_ops_size,
 550                                 struct update_params *params,
 551                                 unsigned int *param_count,
 552                                 size_t *max_param_size,
 553                                 const struct lu_fid *fid)
 554 {
 555         return update_records_update_pack(env, fid, OUT_REF_DEL, ops, op_count,
 556                                           max_ops_size, params, param_count,
 557                                           max_param_size, 0, NULL, NULL);
 558 }
 559 EXPORT_SYMBOL(update_records_ref_del_pack);
 560
 561 /**
 562  * Calculate object destroy update size
 563  *
 564  * \param[in] env       execution environment
 565  * \param[in] fid       FID of the object to delete reference
 566  *
 567  * \retval              size of object destroy update.
 568  */
 569 size_t update_records_object_destroy_size(const struct lu_env *env,
 570                                           const struct lu_fid *fid)
 571 {
 572         return update_records_update_size(0, NULL);
 573 }
 574 EXPORT_SYMBOL(update_records_object_destroy_size);
 575
 576 /**
 577  * Pack object destroy update
 578  *
 579  * Pack object destroy update into update records.
 580  *
 581  * \param[in] env       execution environment
 582  * \param[in] ops       ur_ops in update records
 583  * \param[in|out] op_count pointer to the count of ops
 584  * \param[in|out] max_op_size maximum size of the update
 585  * \param[in] params    ur_params in update records
 586  * \param[in|out] param_count   pointer to the count of params
 587  * \param[in|out] max_param_size maximum size of the parameter
 588  * \param[in] fid       FID of the object to delete reference
 589  *
 590  * \retval              0 if packing succeeds.
 591  * \retval              negative errno if packing fails.
 592  */
 593 int update_records_object_destroy_pack(const struct lu_env *env,
 594                                        struct update_ops *ops,
 595                                        unsigned int *op_count,
 596                                        size_t *max_ops_size,
 597                                        struct update_params *params,
 598                                        unsigned int *param_count,
 599                                        size_t *max_param_size,
 600                                        const struct lu_fid *fid)
 601 {
 602         return update_records_update_pack(env, fid, OUT_DESTROY, ops, op_count,
 603                                           max_ops_size, params, param_count,
 604                                           max_param_size, 0, NULL, NULL);
 605 }
 606 EXPORT_SYMBOL(update_records_object_destroy_pack);
 607
 608 /**
 609  * Calculate index insert update size
 610  *
 611  * \param[in] env       execution environment
 612  * \param[in] fid       FID of the object to insert index
 613  * \param[in] rec       record of insertion
 614  * \param[in] key       key of insertion
 615  *
 616  * \retval              the size of index insert update.
 617  */
 618 size_t update_records_index_insert_size(const struct lu_env *env,
 619                                         const struct lu_fid *fid,
 620                                         const struct dt_rec *rec,
 621                                         const struct dt_key *key)
 622 {
 623         size_t                     sizes[3] = { strlen((const char *)key) + 1,
 624                                                 sizeof(struct lu_fid),
 625                                                 sizeof(__u32) };
 626         return update_records_update_size(3, sizes);
 627 }
 628 EXPORT_SYMBOL(update_records_index_insert_size);
 629
 630 /**
 631  * Pack index insert update
 632  *
 633  * Pack index insert update into update records.
 634  *
 635  * \param[in] env       execution environment
 636  * \param[in] ops       ur_ops in update records
 637  * \param[in] op_count  pointer to the count of ops
 638  * \param[in|out] max_op_size maximum size of the update
 639  * \param[in] params    ur_params in update records
 640  * \param[in] param_count       pointer to the count of params
 641  * \param[in|out] max_param_size maximum size of the parameter
 642  * \param[in] fid       FID of the object to insert index
 643  * \param[in] rec       record of insertion
 644  * \param[in] key       key of insertion
 645  *
 646  * \retval              0 if packing succeeds.
 647  * \retval              negative errno if packing fails.
 648  */
 649 int update_records_index_insert_pack(const struct lu_env *env,
 650                                      struct update_ops *ops,
 651                                      unsigned int *op_count,
 652                                      size_t *max_ops_size,
 653                                      struct update_params *params,
 654                                      unsigned int *param_count,
 655                                      size_t *max_param_size,
 656                                      const struct lu_fid *fid,
 657                                      const struct dt_rec *rec,
 658                                      const struct dt_key *key)
 659 {
 660         struct dt_insert_rec       *rec1 = (struct dt_insert_rec *)rec;
 661         struct lu_fid              rec_fid;
 662         __u32                      type = cpu_to_le32(rec1->rec_type);
 663         size_t                     sizes[3] = { strlen((const char *)key) + 1,
 664                                                 sizeof(rec_fid),
 665                                                 sizeof(type) };
 666         const void                 *bufs[3] = { key,
 667                                                 &rec_fid,
 668                                                 &type };
 669
 670         fid_cpu_to_le(&rec_fid, rec1->rec_fid);
 671
 672         return update_records_update_pack(env, fid, OUT_INDEX_INSERT, ops,
 673                                           op_count, max_ops_size, params,
 674                                           param_count, max_param_size,
 675                                           3, bufs, sizes);
 676 }
 677 EXPORT_SYMBOL(update_records_index_insert_pack);
 678
 679 /**
 680  * Calculate index delete update size
 681  *
 682  * \param[in] env       execution environment
 683  * \param[in] fid       FID of the object to delete index
 684  * \param[in] key       key of deletion
 685  *
 686  * \retval              the size of index delete update
 687  */
 688 size_t update_records_index_delete_size(const struct lu_env *env,
 689                                         const struct lu_fid *fid,
 690                                         const struct dt_key *key)
 691 {
 692         size_t size = strlen((const char *)key) + 1;
 693
 694         return update_records_update_size(1, &size);
 695 }
 696 EXPORT_SYMBOL(update_records_index_delete_size);
 697
 698 /**
 699  * Pack index delete update
 700  *
 701  * Pack index delete update into update records.
 702  *
 703  * \param[in] env       execution environment
 704  * \param[in] ops       ur_ops in update records
 705  * \param[in|out] op_count      pointer to the count of ops
 706  * \param[in|out] max_op_size maximum size of the update
 707  * \param[in] params    ur_params in update records
 708  * \param[in|ount] param_count  pointer to the count of params
 709  * \param[in|out] max_param_size maximum size of the parameter
 710  * \param[in] fid       FID of the object to delete index
 711  * \param[in] key       key of deletion
 712  *
 713  * \retval              0 if packing succeeds.
 714  * \retval              negative errno if packing fails.
 715  */
 716 int update_records_index_delete_pack(const struct lu_env *env,
 717                                      struct update_ops *ops,
 718                                      unsigned int *op_count,
 719                                      size_t *max_ops_size,
 720                                      struct update_params *params,
 721                                      unsigned int *param_count,
 722                                      size_t *max_param_size,
 723                                      const struct lu_fid *fid,
 724                                      const struct dt_key *key)
 725 {
 726         size_t size = strlen((const char *)key) + 1;
 727
 728         return update_records_update_pack(env, fid, OUT_INDEX_DELETE, ops,
 729                                           op_count, max_ops_size, params,
 730                                           param_count, max_param_size,
 731                                           1, (const void **)&key, &size);
 732 }
 733 EXPORT_SYMBOL(update_records_index_delete_pack);
 734
 735 /**
 736  * Calculate xattr set size
 737  *
 738  * \param[in] env       execution environment
 739  * \param[in] fid       FID of the object to set xattr
 740  * \param[in] buf       xattr to be set
 741  * \param[in] name      name of the xattr
 742  * \param[in] flag      flag for setting xattr
 743  *
 744  * \retval              size of xattr set update.
 745  */
 746 size_t update_records_xattr_set_size(const struct lu_env *env,
 747                                      const struct lu_fid *fid,
 748                                      const struct lu_buf *buf,
 749                                      const char *name, __u32 flag)
 750 {
 751         size_t  sizes[3] = {strlen(name) + 1, buf->lb_len, sizeof(flag)};
 752
 753         return update_records_update_size(3, sizes);
 754 }
 755 EXPORT_SYMBOL(update_records_xattr_set_size);
 756
 757 /**
 758  * Pack xattr set update
 759  *
 760  * Pack xattr set update into update records.
 761  *
 762  * \param[in] env       execution environment
 763  * \param[in] ops       ur_ops in update records
 764  * \param[in|out] op_count      pointer to the count of ops
 765  * \param[in|out] max_op_size maximum size of the update
 766  * \param[in] params    ur_params in update records
 767  * \param[in|out] param_count   pointer to the count of params
 768  * \param[in|out] max_param_size maximum size of the parameter
 769  * \param[in] fid       FID of the object to set xattr
 770  * \param[in] buf       xattr to be set
 771  * \param[in] name      name of the xattr
 772  * \param[in] flag      flag for setting xattr
 773  *
 774  * \retval              0 if packing succeeds.
 775  * \retval              negative errno if packing fails.
 776  */
 777 int update_records_xattr_set_pack(const struct lu_env *env,
 778                                   struct update_ops *ops,
 779                                   unsigned int *op_count,
 780                                   size_t *max_ops_size,
 781                                   struct update_params *params,
 782                                   unsigned int *param_count,
 783                                   size_t *max_param_size,
 784                                   const struct lu_fid *fid,
 785                                   const struct lu_buf *buf, const char *name,
 786                                   __u32 flag)
 787 {
 788         size_t  sizes[3] = {strlen(name) + 1, buf->lb_len, sizeof(flag)};
 789         const void *bufs[3] = {name, buf->lb_buf, &flag};
 790
 791         flag = cpu_to_le32(flag);
 792
 793         return update_records_update_pack(env, fid, OUT_XATTR_SET, ops,
 794                                           op_count, max_ops_size, params,
 795                                           param_count, max_param_size,
 796                                           3, bufs, sizes);
 797 }
 798 EXPORT_SYMBOL(update_records_xattr_set_pack);
 799
 800 /**
 801  * Calculate xattr delete update size.
 802  *
 803  * \param[in] env       execution environment
 804  * \param[in] fid       FID of the object to delete xattr
 805  * \param[in] name      name of the xattr
 806  *
 807  * \retval              size of xattr delet updatee.
 808  */
 809 size_t update_records_xattr_del_size(const struct lu_env *env,
 810                                      const struct lu_fid *fid,
 811                                      const char *name)
 812 {
 813         size_t  size = strlen(name) + 1;
 814
 815         return update_records_update_size(1, &size);
 816 }
 817 EXPORT_SYMBOL(update_records_xattr_del_size);
 818
 819 /**
 820  * Pack xattr delete update
 821  *
 822  * Pack xattr delete update into update records.
 823  *
 824  * \param[in] env       execution environment
 825  * \param[in] ops       ur_ops in update records
 826  * \param[in|out] op_count      pointer to the count of ops
 827  * \param[in|out] max_op_size maximum size of the update
 828  * \param[in] params    ur_params in update records
 829  * \param[in|out] param_count   pointer to the count of params
 830  * \param[in|out] max_param_size maximum size of the parameter
 831  * \param[in] fid       FID of the object to delete xattr
 832  * \param[in] name      name of the xattr
 833  *
 834  * \retval              0 if packing succeeds.
 835  * \retval              negative errno if packing fails.
 836  */
 837 int update_records_xattr_del_pack(const struct lu_env *env,
 838                                   struct update_ops *ops,
 839                                   unsigned int *op_count,
 840                                   size_t *max_ops_size,
 841                                   struct update_params *params,
 842                                   unsigned int *param_count,
 843                                   size_t *max_param_size,
 844                                   const struct lu_fid *fid,
 845                                   const char *name)
 846 {
 847         size_t  size = strlen(name) + 1;
 848
 849         return update_records_update_pack(env, fid, OUT_XATTR_DEL, ops,
 850                                           op_count, max_ops_size, params,
 851                                           param_count, max_param_size,
 852                                           1, (const void **)&name, &size);
 853 }
 854 EXPORT_SYMBOL(update_records_xattr_del_pack);
 855
 856 /**
 857  * Calculate write update size
 858  *
 859  * \param[in] env       execution environment
 860  * \param[in] fid       FID of the object to write into
 861  * \param[in] buf       buffer to write which includes an embedded size field
 862  * \param[in] pos       offet in the object to start writing at
 863  *
 864  * \retval              size of write udpate.
 865  */
 866 size_t update_records_write_size(const struct lu_env *env,
 867                                  const struct lu_fid *fid,
 868                                  const struct lu_buf *buf,
 869                                  __u64 pos)
 870 {
 871         size_t  sizes[2] = {buf->lb_len, sizeof(pos)};
 872
 873         return update_records_update_size(2, sizes);
 874 }
 875 EXPORT_SYMBOL(update_records_write_size);
 876
 877 /**
 878  * Pack write update
 879  *
 880  * Pack write update into update records.
 881  *
 882  * \param[in] env       execution environment
 883  * \param[in] ops       ur_ops in update records
 884  * \param[in|out] op_count      pointer to the count of ops
 885  * \param[in|out] max_op_size maximum size of the update
 886  * \param[in] params    ur_params in update records
 887  * \param[in|out] param_count   pointer to the count of params
 888  * \param[in|out] max_param_size maximum size of the parameter
 889  * \param[in] fid       FID of the object to write into
 890  * \param[in] buf       buffer to write which includes an embedded size field
 891  * \param[in] pos       offet in the object to start writing at
 892  *
 893  * \retval              0 if packing succeeds.
 894  * \retval              negative errno if packing fails.
 895  */
 896 int update_records_write_pack(const struct lu_env *env,
 897                               struct update_ops *ops,
 898                               unsigned int *op_count,
 899                               size_t *max_ops_size,
 900                               struct update_params *params,
 901                               unsigned int *param_count,
 902                               size_t *max_param_size,
 903                               const struct lu_fid *fid,
 904                               const struct lu_buf *buf,
 905                               __u64 pos)
 906 {
 907         size_t          sizes[2] = {buf->lb_len, sizeof(pos)};
 908         const void      *bufs[2] = {buf->lb_buf, &pos};
 909
 910         pos = cpu_to_le64(pos);
 911
 912         return update_records_update_pack(env, fid, OUT_WRITE, ops,
 913                                           op_count, max_ops_size, params,
 914                                           param_count, max_param_size,
 915                                           2, bufs, sizes);
 916 }
 917 EXPORT_SYMBOL(update_records_write_pack);
 918
 919 /**
 920  * Calculate size of punch update.
 921  *
 922  * \param[in] env       execution environment
 923  * \param[in] fid       FID of the object to write into
 924  * \param[in] start     start offset of punch
 925  * \param[in] end       end offet of punch
 926  *
 927  * \retval              size of update punch.
 928  */
 929 size_t update_records_punch_size(const struct lu_env *env,
 930                                  const struct lu_fid *fid,
 931                                  __u64 start, __u64 end)
 932 {
 933         size_t  sizes[2] = {sizeof(start), sizeof(end)};
 934
 935         return update_records_update_size(2, sizes);
 936 }
 937 EXPORT_SYMBOL(update_records_punch_size);
 938
 939 /**
 940  * Pack punch
 941  *
 942  * Pack punch update into update records.
 943  *
 944  * \param[in] env       execution environment
 945  * \param[in] ops       ur_ops in update records
 946  * \param[in|out] op_count      pointer to the count of ops
 947  * \param[in|out] max_op_size maximum size of the update
 948  * \param[in] params    ur_params in update records
 949  * \param[in|out] param_count   pointer to the count of params
 950  * \param[in|out] max_param_size maximum size of the parameter
 951  * \param[in] fid       FID of the object to write into
 952  * \param[in] start     start offset of punch
 953  * \param[in] end       end offet of punch
 954  *
 955  * \retval              0 if packing succeeds.
 956  * \retval              negative errno if packing fails.
 957  */
 958 int update_records_punch_pack(const struct lu_env *env,
 959                               struct update_ops *ops,
 960                               unsigned int *op_count,
 961                               size_t *max_ops_size,
 962                               struct update_params *params,
 963                               unsigned int *param_count,
 964                               size_t *max_param_size,
 965                               const struct lu_fid *fid,
 966                               __u64 start, __u64 end)
 967 {
 968         size_t          sizes[2] = {sizeof(start), sizeof(end)};
 969         const void      *bufs[2] = {&start, &end};
 970
 971         start = cpu_to_le64(start);
 972         end = cpu_to_le64(end);
 973
 974         return update_records_update_pack(env, fid, OUT_PUNCH, ops, op_count,
 975                                           max_ops_size, params, param_count,
 976                                           max_param_size, 2, bufs, sizes);
 977 }
 978 EXPORT_SYMBOL(update_records_punch_pack);
 979
 980 /**
 981  * Create update records in thandle_update_records
 982  *
 983  * Allocate update_records for thandle_update_records, the initial size
 984  * will be 4KB.
 985  *
 986  * \param[in] tur       thandle_update_records where update_records will be
 987  *                      allocated
 988  * \retval              0 if allocation succeeds.
 989  * \retval              negative errno if allocation fails.
 990  */
 991 static int tur_update_records_create(struct thandle_update_records *tur)
 992 {
 993         if (tur->tur_update_records != NULL)
 994                 return 0;
 995
 996         OBD_ALLOC_LARGE(tur->tur_update_records,
 997                         UPDATE_RECORDS_BUFFER_SIZE);
 998
 999         if (tur->tur_update_records == NULL)
1000                 return -ENOMEM;
1001
1002         tur->tur_update_records_buf_size = UPDATE_RECORDS_BUFFER_SIZE;
1003
1004         return 0;
1005 }
1006
1007 /**
1008  * Extend update records
1009  *
1010  * Extend update_records to the new size in thandle_update_records.
1011  *
1012  * \param[in] tur       thandle_update_records where update_records will be
1013  *                      extended.
1014  * \retval              0 if extension succeeds.
1015  * \retval              negative errno if extension fails.
1016  */
1017 int tur_update_records_extend(struct thandle_update_records *tur,
1018                               size_t new_size)
1019 {
1020         struct llog_update_record       *record;
1021
1022         OBD_ALLOC_LARGE(record, new_size);
1023         if (record == NULL)
1024                 return -ENOMEM;
1025
1026         if (tur->tur_update_records != NULL) {
1027                 memcpy(record, tur->tur_update_records,
1028                        tur->tur_update_records_buf_size);
1029                 OBD_FREE_LARGE(tur->tur_update_records,
1030                                tur->tur_update_records_buf_size);
1031         }
1032
1033         tur->tur_update_records = record;
1034         tur->tur_update_records_buf_size = new_size;
1035
1036         return 0;
1037 }
1038 EXPORT_SYMBOL(tur_update_records_extend);
1039
1040 /**
1041  * Extend update records
1042  *
1043  * Extend update records in thandle to make sure it is able to hold
1044  * the update with certain update_op and params size.
1045  *
1046  * \param [in] tur      thandle_update_records to be extend
1047  * \param [in] new_op_size update_op size of the update record
1048  * \param [in] new_param_size params size of the update record
1049  *
1050  * \retval              0 if the update_records is being extended.
1051  * \retval              negative errno if the update_records is not being
1052  *                      extended.
1053  */
1054 int tur_update_extend(struct thandle_update_records *tur,
1055                       size_t new_op_size, size_t new_param_size)
1056 {
1057         size_t record_size;
1058         size_t params_size;
1059         size_t extend_size;
1060         int rc;
1061         ENTRY;
1062
1063         record_size = llog_update_record_size(tur->tur_update_records);
1064         /* extend update records buffer */
1065         if (new_op_size >= (tur->tur_update_records_buf_size - record_size)) {
1066                 extend_size = round_up(new_op_size, UPDATE_RECORDS_BUFFER_SIZE);
1067                 rc = tur_update_records_extend(tur,
1068                                 tur->tur_update_records_buf_size +
1069                                 extend_size);
1070                 if (rc != 0)
1071                         RETURN(rc);
1072         }
1073
1074         /* extend parameters buffer */
1075         params_size = update_params_size(tur->tur_update_params,
1076                                          tur->tur_update_param_count);
1077         if (new_param_size >= (tur->tur_update_params_buf_size -
1078                               params_size)) {
1079                 extend_size = round_up(new_param_size,
1080                                        UPDATE_PARAMS_BUFFER_SIZE);
1081                 rc = tur_update_params_extend(tur,
1082                                 tur->tur_update_params_buf_size +
1083                                 extend_size);
1084                 if (rc != 0)
1085                         RETURN(rc);
1086         }
1087
1088         RETURN(0);
1089 }
1090 EXPORT_SYMBOL(tur_update_extend);
1091
1092 /**
1093  * Create update params in thandle_update_records
1094  *
1095  * Allocate update_params for thandle_update_records, the initial size
1096  * will be 4KB.
1097  *
1098  * \param[in] tur       thandle_update_records where update_params will be
1099  *                      allocated
1100  * \retval              0 if allocation succeeds.
1101  * \retval              negative errno if allocation fails.
1102  */
1103 static int tur_update_params_create(struct thandle_update_records *tur)
1104 {
1105         if (tur->tur_update_params != NULL)
1106                 return 0;
1107
1108         OBD_ALLOC_LARGE(tur->tur_update_params, UPDATE_PARAMS_BUFFER_SIZE);
1109         if (tur->tur_update_params == NULL)
1110                 return -ENOMEM;
1111
1112         tur->tur_update_params_buf_size = UPDATE_PARAMS_BUFFER_SIZE;
1113         return 0;
1114 }
1115
1116 /**
1117  * Extend update params
1118  *
1119  * Extend update_params to the new size in thandle_update_records.
1120  *
1121  * \param[in] tur       thandle_update_records where update_params will be
1122  *                      extended.
1123  * \retval              0 if extension succeeds.
1124  * \retval              negative errno if extension fails.
1125  */
1126 int tur_update_params_extend(struct thandle_update_records *tur,
1127                              size_t new_size)
1128 {
1129         struct update_params    *params;
1130
1131         OBD_ALLOC_LARGE(params, new_size);
1132         if (params == NULL)
1133                 return -ENOMEM;
1134
1135         if (tur->tur_update_params != NULL) {
1136                 memcpy(params, tur->tur_update_params,
1137                        tur->tur_update_params_buf_size);
1138                 OBD_FREE_LARGE(tur->tur_update_params,
1139                                tur->tur_update_params_buf_size);
1140         }
1141
1142         tur->tur_update_params = params;
1143         tur->tur_update_params_buf_size = new_size;
1144
1145         return 0;
1146 }
1147 EXPORT_SYMBOL(tur_update_params_extend);
1148
1149 /**
1150  * Check and prepare whether it needs to record update.
1151  *
1152  * Checks if the transaction needs to record updates, and if it
1153  * does, then initialize the update record buffer in the transaction.
1154  *
1155  * \param[in] env       execution environment
1156  * \param[in] th        transaction handle
1157  *
1158  * \retval              0 if updates recording succeeds.
1159  * \retval              negative errno if updates recording fails.
1160  */
1161 int check_and_prepare_update_record(const struct lu_env *env,
1162                                     struct thandle_update_records *tur)
1163 {
1164         struct llog_update_record       *lur;
1165         int rc;
1166
1167         if (tur->tur_update_records == NULL) {
1168                 rc = tur_update_records_create(tur);
1169                 if (rc < 0)
1170                         RETURN(rc);
1171         }
1172
1173         if (tur->tur_update_params == NULL) {
1174                 rc = tur_update_params_create(tur);
1175                 if (rc < 0)
1176                         RETURN(rc);
1177         }
1178
1179         lur = tur->tur_update_records;
1180         lur->lur_update_rec.ur_update_count = 0;
1181         lur->lur_update_rec.ur_param_count = 0;
1182         lur->lur_update_rec.ur_master_transno = 0;
1183         lur->lur_update_rec.ur_batchid = 0;
1184         lur->lur_update_rec.ur_flags = 0;
1185         lur->lur_hdr.lrh_len = LLOG_MIN_CHUNK_SIZE;
1186
1187         tur->tur_update_param_count = 0;
1188
1189         RETURN(0);
1190 }
1191
1192 static void update_key_fini(const struct lu_context *ctx,
1193                             struct lu_context_key *key, void *data)
1194 {
1195         struct update_thread_info *info = data;
1196         struct thandle_exec_args  *args = &info->uti_tea;
1197         int                       i;
1198
1199         for (i = 0; i < args->ta_alloc_args; i++) {
1200                 if (args->ta_args[i] != NULL)
1201                         OBD_FREE_PTR(args->ta_args[i]);
1202         }
1203
1204         if (args->ta_args != NULL)
1205                 OBD_FREE(args->ta_args, sizeof(args->ta_args[0]) *
1206                          args->ta_alloc_args);
1207
1208         if (info->uti_tur.tur_update_records != NULL)
1209                 OBD_FREE_LARGE(info->uti_tur.tur_update_records,
1210                                info->uti_tur.tur_update_records_buf_size);
1211         if (info->uti_tur.tur_update_params != NULL)
1212                 OBD_FREE_LARGE(info->uti_tur.tur_update_params,
1213                                info->uti_tur.tur_update_params_buf_size);
1214
1215         OBD_FREE_PTR(info);
1216 }
1217
1218 /* context key constructor/destructor: update_key_init, update_key_fini */
1219 LU_KEY_INIT(update, struct update_thread_info);
1220 /* context key: update_thread_key */
1221 LU_CONTEXT_KEY_DEFINE(update, LCT_MD_THREAD | LCT_MG_THREAD |
1222                               LCT_DT_THREAD | LCT_TX_HANDLE | LCT_LOCAL);
1223 EXPORT_SYMBOL(update_thread_key);
1224 LU_KEY_INIT_GENERIC(update);
1225
1226 void update_info_init(void)
1227 {
1228         update_key_init_generic(&update_thread_key, NULL);
1229         lu_context_key_register(&update_thread_key);
1230 }
1231
1232 void update_info_fini(void)
1233 {
1234         lu_context_key_degister(&update_thread_key);
1235 }