4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
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.
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).
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
23 * Copyright (c) 2015, 2017, Intel Corporation.
27 * lustre/target/update_records.c
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
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.
37 * See comments above struct update_records for the format of update_records.
39 * Author: Di Wang <di.wang@intel.com>
41 #define DEBUG_SUBSYSTEM S_CLASS
43 #include <lu_target.h>
44 #include <lustre_obdo.h>
45 #include <lustre_update.h>
47 #include <obd_class.h>
49 #include "tgt_internal.h"
51 #define UPDATE_RECORDS_BUFFER_SIZE 8192
52 #define UPDATE_PARAMS_BUFFER_SIZE 8192
56 * Dump all of updates in the update_records, mostly for debugging purpose.
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.
63 void update_records_dump(const struct update_records *records,
64 unsigned int mask, bool dump_updates)
66 const struct update_ops *ops;
67 const struct update_op *op = NULL;
68 struct update_params *params = NULL;
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);
76 if (records->ur_update_count == 0)
82 ops = &records->ur_ops;
83 if (records->ur_param_count > 0)
84 params = update_records_get_params(records);
86 op = &ops->uops_op[0];
87 for (i = 0; i < records->ur_update_count; i++,
88 op = update_op_next_op(op)) {
91 CDEBUG(mask, "update %dth "DFID" %s params_count = %hu\n", i,
92 PFID(&op->uop_fid), update_op_str(op->uop_type),
98 for (j = 0; j < op->uop_param_count; j++) {
99 struct object_update_param *param;
101 param = update_params_get_param(params,
102 (unsigned int)op->uop_params_off[j],
103 records->ur_param_count);
107 CDEBUG(mask, "param = %p %dth off = %hu size = %hu\n",
108 param, j, op->uop_params_off[j], param->oup_len);
114 * Pack parameters to update records
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
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
124 * \retval index inside \a params if parameter insertion
126 * \retval negative errno if it fails.
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)
133 struct object_update_param *param;
136 for (i = 0; i < *param_count; i++) {
137 struct object_update_param *param;
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 */
147 param = (struct object_update_param *)((char *)params +
148 update_params_size(params, *param_count));
150 param->oup_len = new_param_size;
151 if (new_param != NULL)
152 memcpy(param->oup_buf, new_param, new_param_size);
154 *param_count = *param_count + 1;
156 return *param_count - 1;
160 * Pack update to update records
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.
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
181 * \retval 0 if packing succeeds
182 * \retval negative errno if packing fails
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,
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,
197 struct update_op *op;
198 size_t total_param_sizes = 0;
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);
210 for (i = 0; i < param_bufs_count; i++)
212 cfs_size_round(sizeof(struct object_update_param) +
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);
220 *max_param_size = total_param_sizes;
224 op = update_ops_get_op(ops, *op_count, *op_count);
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);
234 CDEBUG(D_INFO, "%s %uth param offset = %d size = %zu\n",
235 update_op_str(op_type), i, index, param_sizes[i]);
237 op->uop_params_off[i] = index;
239 CDEBUG(D_INFO, "%huth "DFID" %s param_count = %u\n",
240 *op_count, PFID(fid), update_op_str(op_type), *param_count);
242 *op_count = *op_count + 1;
248 * Calculate update_records size
250 * Calculate update_records size by param_count and param_sizes array.
252 * \param[in] param_count the count of parameters
253 * \param[in] sizes the size array of these parameters
255 * \retval the size of this update
257 static size_t update_records_update_size(__u32 param_count, size_t *sizes)
262 /* Check whether the packing exceeding the maximum update size */
263 size = update_op_size(param_count);
265 for (i = 0; i < param_count; i++)
266 size += cfs_size_round(sizeof(struct object_update_param) +
273 * Calculate create update size
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
282 * \retval size of create update.
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)
294 sizes[param_count] = sizeof(struct obdo);
298 if (hint != NULL && hint->dah_parent != NULL) {
299 sizes[param_count] = sizeof(*fid);
303 return update_records_update_size(param_count, sizes);
305 EXPORT_SYMBOL(update_records_create_size);
310 * Pack create update into update records.
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
324 * \retval 0 if packing succeeds.
325 * \retval negative errno if packing fails.
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)
342 const struct lu_fid *parent_fid = NULL;
343 struct lu_fid tmp_fid;
348 obdo = &update_env_info(env)->uti_obdo;
350 obdo_from_la(obdo, attr, attr->la_valid);
351 bufs[buf_count] = obdo;
352 sizes[buf_count] = sizeof(*obdo);
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);
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);
369 EXPORT_SYMBOL(update_records_create_pack);
372 * Calculate attr set update size
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
379 * \retval size of attr set update.
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)
385 size_t size = sizeof(struct obdo);
387 return update_records_update_size(1, &size);
389 EXPORT_SYMBOL(update_records_attr_set_size);
392 * Pack attr set update
394 * Pack attr_set update into update records.
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
406 * \retval 0 if packing succeeds.
407 * \retval negative errno if packing fails.
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)
419 struct obdo *obdo = &update_env_info(env)->uti_obdo;
420 size_t size = sizeof(*obdo);
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,
427 (const void **)&obdo, &size);
429 EXPORT_SYMBOL(update_records_attr_set_pack);
432 * Calculate ref add update size
434 * \param[in] env execution environment
435 * \param[in] fid FID of the object to add reference
437 * \retval size of ref_add udpate.
439 size_t update_records_ref_add_size(const struct lu_env *env,
440 const struct lu_fid *fid)
442 return update_records_update_size(0, NULL);
444 EXPORT_SYMBOL(update_records_ref_add_size);
447 * Pack ref add update
449 * Pack ref add update into update records.
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
460 * \retval 0 if packing succeeds.
461 * \retval negative errno if packing fails.
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)
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);
476 EXPORT_SYMBOL(update_records_ref_add_pack);
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.
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
494 * \retval 0 if packing succeeds.
495 * \retval negative errno if packing fails.
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)
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);
510 EXPORT_SYMBOL(update_records_noop_pack);
513 * Calculate ref del update size
515 * \param[in] env execution environment
516 * \param[in] fid FID of the object to delete reference
518 * \retval size of ref_del update.
520 size_t update_records_ref_del_size(const struct lu_env *env,
521 const struct lu_fid *fid)
523 return update_records_update_size(0, NULL);
525 EXPORT_SYMBOL(update_records_ref_del_size);
528 * Pack ref del update
530 * Pack ref del update into update records.
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
541 * \retval 0 if packing succeeds.
542 * \retval negative errno if packing fails.
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)
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);
557 EXPORT_SYMBOL(update_records_ref_del_pack);
560 * Calculate object destroy update size
562 * \param[in] env execution environment
563 * \param[in] fid FID of the object to delete reference
565 * \retval size of object destroy update.
567 size_t update_records_destroy_size(const struct lu_env *env,
568 const struct lu_fid *fid)
570 return update_records_update_size(0, NULL);
572 EXPORT_SYMBOL(update_records_destroy_size);
575 * Pack object destroy update
577 * Pack object destroy update into update records.
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
588 * \retval 0 if packing succeeds.
589 * \retval negative errno if packing fails.
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)
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);
604 EXPORT_SYMBOL(update_records_destroy_pack);
607 * Calculate index insert update size
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
614 * \retval the size of index insert update.
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)
621 size_t sizes[3] = { strlen((const char *)key) + 1,
622 sizeof(struct lu_fid),
624 return update_records_update_size(3, sizes);
626 EXPORT_SYMBOL(update_records_index_insert_size);
629 * Pack index insert update
631 * Pack index insert update into update records.
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
644 * \retval 0 if packing succeeds.
645 * \retval negative errno if packing fails.
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)
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,
664 const void *bufs[3] = { key,
668 fid_cpu_to_le(&rec_fid, rec1->rec_fid);
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,
675 EXPORT_SYMBOL(update_records_index_insert_pack);
678 * Calculate index delete update size
680 * \param[in] env execution environment
681 * \param[in] fid FID of the object to delete index
682 * \param[in] key key of deletion
684 * \retval the size of index delete update
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)
690 size_t size = strlen((const char *)key) + 1;
692 return update_records_update_size(1, &size);
694 EXPORT_SYMBOL(update_records_index_delete_size);
697 * Pack index delete update
699 * Pack index delete update into update records.
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
711 * \retval 0 if packing succeeds.
712 * \retval negative errno if packing fails.
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)
724 size_t size = strlen((const char *)key) + 1;
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);
731 EXPORT_SYMBOL(update_records_index_delete_pack);
734 * Calculate xattr set size
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
742 * \retval size of xattr set update.
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)
749 size_t sizes[3] = {strlen(name) + 1, buf->lb_len, sizeof(flag)};
751 return update_records_update_size(3, sizes);
753 EXPORT_SYMBOL(update_records_xattr_set_size);
756 * Pack xattr set update
758 * Pack xattr set update into update records.
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
772 * \retval 0 if packing succeeds.
773 * \retval negative errno if packing fails.
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,
786 size_t sizes[3] = {strlen(name) + 1, buf->lb_len, sizeof(flag)};
787 const void *bufs[3] = {name, buf->lb_buf, &flag};
789 flag = cpu_to_le32(flag);
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,
796 EXPORT_SYMBOL(update_records_xattr_set_pack);
799 * Calculate xattr delete update size.
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
805 * \retval size of xattr delet updatee.
807 size_t update_records_xattr_del_size(const struct lu_env *env,
808 const struct lu_fid *fid,
811 size_t size = strlen(name) + 1;
813 return update_records_update_size(1, &size);
815 EXPORT_SYMBOL(update_records_xattr_del_size);
818 * Pack xattr delete update
820 * Pack xattr delete update into update records.
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
832 * \retval 0 if packing succeeds.
833 * \retval negative errno if packing fails.
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,
845 size_t size = strlen(name) + 1;
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);
852 EXPORT_SYMBOL(update_records_xattr_del_pack);
855 * Calculate write update size
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
862 * \retval size of write udpate.
864 size_t update_records_write_size(const struct lu_env *env,
865 const struct lu_fid *fid,
866 const struct lu_buf *buf,
869 size_t sizes[2] = {buf->lb_len, sizeof(pos)};
871 return update_records_update_size(2, sizes);
873 EXPORT_SYMBOL(update_records_write_size);
878 * Pack write update into update records.
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
891 * \retval 0 if packing succeeds.
892 * \retval negative errno if packing fails.
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,
905 size_t sizes[2] = {buf->lb_len, sizeof(pos)};
906 const void *bufs[2] = {buf->lb_buf, &pos};
908 pos = cpu_to_le64(pos);
910 return update_records_update_pack(env, fid, OUT_WRITE, ops,
911 op_count, max_ops_size, params,
912 param_count, max_param_size,
915 EXPORT_SYMBOL(update_records_write_pack);
918 * Calculate size of punch update.
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
925 * \retval size of update punch.
927 size_t update_records_punch_size(const struct lu_env *env,
928 const struct lu_fid *fid,
929 __u64 start, __u64 end)
931 size_t sizes[2] = {sizeof(start), sizeof(end)};
933 return update_records_update_size(2, sizes);
935 EXPORT_SYMBOL(update_records_punch_size);
940 * Pack punch update into update records.
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
953 * \retval 0 if packing succeeds.
954 * \retval negative errno if packing fails.
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)
966 size_t sizes[2] = {sizeof(start), sizeof(end)};
967 const void *bufs[2] = {&start, &end};
969 start = cpu_to_le64(start);
970 end = cpu_to_le64(end);
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);
976 EXPORT_SYMBOL(update_records_punch_pack);
979 * Create update records in thandle_update_records
981 * Allocate update_records for thandle_update_records, the initial size
984 * \param[in] tur thandle_update_records where update_records will be
986 * \retval 0 if allocation succeeds.
987 * \retval negative errno if allocation fails.
989 static int tur_update_records_create(struct thandle_update_records *tur)
991 if (tur->tur_update_records != NULL)
994 OBD_ALLOC_LARGE(tur->tur_update_records,
995 UPDATE_RECORDS_BUFFER_SIZE);
997 if (tur->tur_update_records == NULL)
1000 tur->tur_update_records_buf_size = UPDATE_RECORDS_BUFFER_SIZE;
1006 * Extend update records
1008 * Extend update_records to the new size in thandle_update_records.
1010 * \param[in] tur thandle_update_records where update_records will be
1012 * \retval 0 if extension succeeds.
1013 * \retval negative errno if extension fails.
1015 int tur_update_records_extend(struct thandle_update_records *tur,
1018 struct llog_update_record *record;
1020 OBD_ALLOC_LARGE(record, new_size);
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);
1031 tur->tur_update_records = record;
1032 tur->tur_update_records_buf_size = new_size;
1036 EXPORT_SYMBOL(tur_update_records_extend);
1039 * Extend update records
1041 * Extend update records in thandle to make sure it is able to hold
1042 * the update with certain update_op and params size.
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
1048 * \retval 0 if the update_records is being extended.
1049 * \retval negative errno if the update_records is not being
1052 int tur_update_extend(struct thandle_update_records *tur,
1053 size_t new_op_size, size_t new_param_size)
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 +
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 -
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 +
1088 EXPORT_SYMBOL(tur_update_extend);
1091 * Create update params in thandle_update_records
1093 * Allocate update_params for thandle_update_records, the initial size
1096 * \param[in] tur thandle_update_records where update_params will be
1098 * \retval 0 if allocation succeeds.
1099 * \retval negative errno if allocation fails.
1101 static int tur_update_params_create(struct thandle_update_records *tur)
1103 if (tur->tur_update_params != NULL)
1106 OBD_ALLOC_LARGE(tur->tur_update_params, UPDATE_PARAMS_BUFFER_SIZE);
1107 if (tur->tur_update_params == NULL)
1110 tur->tur_update_params_buf_size = UPDATE_PARAMS_BUFFER_SIZE;
1115 * Extend update params
1117 * Extend update_params to the new size in thandle_update_records.
1119 * \param[in] tur thandle_update_records where update_params will be
1121 * \retval 0 if extension succeeds.
1122 * \retval negative errno if extension fails.
1124 int tur_update_params_extend(struct thandle_update_records *tur,
1127 struct update_params *params;
1129 OBD_ALLOC_LARGE(params, new_size);
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);
1140 tur->tur_update_params = params;
1141 tur->tur_update_params_buf_size = new_size;
1145 EXPORT_SYMBOL(tur_update_params_extend);
1148 * Check and prepare whether it needs to record update.
1150 * Checks if the transaction needs to record updates, and if it
1151 * does, then initialize the update record buffer in the transaction.
1153 * \param[in] env execution environment
1154 * \param[in] th transaction handle
1156 * \retval 0 if updates recording succeeds.
1157 * \retval negative errno if updates recording fails.
1159 int check_and_prepare_update_record(const struct lu_env *env,
1160 struct thandle_update_records *tur)
1162 struct llog_update_record *lur;
1165 if (tur->tur_update_records == NULL) {
1166 rc = tur_update_records_create(tur);
1171 if (tur->tur_update_params == NULL) {
1172 rc = tur_update_params_create(tur);
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;
1185 tur->tur_update_param_count = 0;
1190 static void update_key_fini(const struct lu_context *ctx,
1191 struct lu_context_key *key, void *data)
1193 struct update_thread_info *info = data;
1194 struct thandle_exec_args *args = &info->uti_tea;
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]);
1202 if (args->ta_args != NULL)
1203 OBD_FREE_PTR_ARRAY(args->ta_args, args->ta_alloc_args);
1205 if (info->uti_tur.tur_update_records != NULL)
1206 OBD_FREE_LARGE(info->uti_tur.tur_update_records,
1207 info->uti_tur.tur_update_records_buf_size);
1208 if (info->uti_tur.tur_update_params != NULL)
1209 OBD_FREE_LARGE(info->uti_tur.tur_update_params,
1210 info->uti_tur.tur_update_params_buf_size);
1215 /* context key constructor/destructor: update_key_init, update_key_fini */
1216 LU_KEY_INIT(update, struct update_thread_info);
1217 /* context key: update_thread_key */
1218 LU_CONTEXT_KEY_DEFINE(update, LCT_MD_THREAD | LCT_MG_THREAD |
1219 LCT_DT_THREAD | LCT_LOCAL);
1220 EXPORT_SYMBOL(update_thread_key);
1221 LU_KEY_INIT_GENERIC(update);
1223 void update_info_init(void)
1225 update_key_init_generic(&update_thread_key, NULL);
1226 lu_context_key_register(&update_thread_key);
1229 void update_info_fini(void)
1231 lu_context_key_degister(&update_thread_key);