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) 2014, 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_update.h>
46 #include <obd_class.h>
47 #include "tgt_internal.h"
49 #define UPDATE_RECORDS_BUFFER_SIZE 8192
50 #define UPDATE_PARAMS_BUFFER_SIZE 8192
54 * Dump all of updates in the update_records, mostly for debugging purpose.
56 * \param[in] records update records to be dumpped
57 * \param[in] mask debug level mask
58 * \param[in] dump_params if dump all of updates the updates.
61 void update_records_dump(const struct update_records *records,
62 unsigned int mask, bool dump_updates)
64 const struct update_ops *ops;
65 const struct update_op *op = NULL;
66 struct update_params *params;
69 ops = &records->ur_ops;
70 params = update_records_get_params(records);
72 CDEBUG(mask, "ops = %d params_count = %d\n", records->ur_update_count,
73 records->ur_param_count);
75 if (records->ur_update_count == 0)
81 op = &ops->uops_op[0];
82 for (i = 0; i < records->ur_update_count; i++) {
85 CDEBUG(mask, "update %dth "DFID" %s params_count = %hu\n", i,
86 PFID(&op->uop_fid), update_op_str(op->uop_type),
89 for (j = 0; j < op->uop_param_count; j++) {
90 struct object_update_param *param;
92 param = update_params_get_param(params,
93 (unsigned int)op->uop_params_off[j],
94 records->ur_param_count);
96 LASSERT(param != NULL);
97 CDEBUG(mask, "param = %p %dth off = %hu size = %hu\n",
98 param, j, op->uop_params_off[j], param->oup_len);
101 op = update_op_next_op(op);
106 * Pack parameters to update records
108 * Find and insert parameter to update records, if the parameter
109 * already exists in \a params, then just return the offset of this
110 * parameter, otherwise insert the parameter and return its offset
112 * \param[in] params update params in which to insert parameter
113 * \param[in] new_param parameters to be inserted.
114 * \param[in] new_param_size the size of \a new_param
116 * \retval index inside \a params if parameter insertion
118 * \retval negative errno if it fails.
120 static unsigned int update_records_param_pack(struct update_params *params,
121 const void *new_param,
122 size_t new_param_size,
123 unsigned int *param_count)
125 struct object_update_param *param;
128 for (i = 0; i < *param_count; i++) {
129 struct object_update_param *param;
131 param = update_params_get_param(params, i, *param_count);
132 if ((new_param == NULL && param->oup_len == new_param_size) ||
133 (param->oup_len == new_param_size &&
134 memcmp(param->oup_buf, new_param, new_param_size) == 0))
135 /* Found the parameter and return its index */
139 param = (struct object_update_param *)((char *)params +
140 update_params_size(params, *param_count));
142 param->oup_len = new_param_size;
143 if (new_param != NULL)
144 memcpy(param->oup_buf, new_param, new_param_size);
146 *param_count = *param_count + 1;
148 return *param_count - 1;
152 * Pack update to update records
154 * Pack the update and its parameters to the update records. First it will
155 * insert parameters, get the offset of these parameter, then fill the
156 * update with these offset. If insertion exceed the maximum size of
157 * current update records, it will return -E2BIG here, and the caller might
158 * extend the update_record size \see lod_updates_pack.
160 * \param[in] env execution environment
161 * \param[in] fid FID of the update.
162 * \param[in] op_type operation type of the update
163 * \param[in] ops ur_ops in update records
164 * \param[in|out] op_count pointer to the count of ops
165 * \param[in|out] max_op_size maximum size of the update
166 * \param[in] params ur_params in update records
167 * \param[in|out] param_count pointer to the count of params
168 * \param[in|out] max_param_size maximum size of the parameter
169 * \param[in] param_bufs buffers of parameters
170 * \param[in] params_buf_count the count of the parameter buffers
171 * \param[in] param_size sizes of parameters
173 * \retval 0 if packing succeeds
174 * \retval negative errno if packing fails
176 static int update_records_update_pack(const struct lu_env *env,
177 const struct lu_fid *fid,
178 enum update_type op_type,
179 struct update_ops *ops,
180 unsigned int *op_count,
182 struct update_params *params,
183 unsigned int *param_count,
184 size_t *max_param_size,
185 unsigned int param_bufs_count,
186 const void **param_bufs,
189 struct update_op *op;
190 size_t total_param_sizes = 0;
194 /* Check whether the packing exceeding the maximum update size */
195 if (unlikely(*max_op_size < update_op_size(param_bufs_count))) {
196 CDEBUG(D_INFO, "max_op_size = %zu update_op = %zu\n",
197 *max_op_size, update_op_size(param_bufs_count));
198 *max_op_size = update_op_size(param_bufs_count);
202 for (i = 0; i < param_bufs_count; i++)
204 cfs_size_round(sizeof(struct object_update_param) +
207 /* Check whether the packing exceeding the maximum parameter size */
208 if (unlikely(*max_param_size < total_param_sizes)) {
209 CDEBUG(D_INFO, "max_param_size = %zu params size = %zu\n",
210 *max_param_size, total_param_sizes);
212 *max_param_size = total_param_sizes;
216 op = update_ops_get_op(ops, *op_count, *op_count);
218 op->uop_type = op_type;
219 op->uop_param_count = param_bufs_count;
220 for (i = 0; i < param_bufs_count; i++) {
221 index = update_records_param_pack(params, param_bufs[i],
222 param_sizes[i], param_count);
226 CDEBUG(D_INFO, "%s %uth param offset = %d size = %zu\n",
227 update_op_str(op_type), i, index, param_sizes[i]);
229 op->uop_params_off[i] = index;
231 CDEBUG(D_INFO, "%huth "DFID" %s param_count = %u\n",
232 *op_count, PFID(fid), update_op_str(op_type), *param_count);
234 *op_count = *op_count + 1;
242 * Pack create update into update records.
244 * \param[in] env execution environment
245 * \param[in] ops ur_ops in update records
246 * \param[in|out] op_count pointer to the count of ops
247 * \param[in|out] max_op_size maximum size of the update
248 * \param[in] params ur_params in update records
249 * \param[in|out] param_count pointer to the count of params
250 * \param[in|out] max_param_size maximum size of the parameter
251 * \param[in] fid FID of the object to be created
252 * \param[in] attr attribute of the object to be created
253 * \param[in] hint creation hint
254 * \param[in] dof creation format information
256 * \retval 0 if packing succeeds.
257 * \retval negative errno if packing fails.
259 int update_records_create_pack(const struct lu_env *env,
260 struct update_ops *ops,
261 unsigned int *op_count,
262 size_t *max_ops_size,
263 struct update_params *params,
264 unsigned int *param_count,
265 size_t *max_param_size,
266 const struct lu_fid *fid,
267 const struct lu_attr *attr,
268 const struct dt_allocation_hint *hint,
269 struct dt_object_format *dof)
274 const struct lu_fid *parent_fid = NULL;
275 struct lu_fid tmp_fid;
280 obdo = &update_env_info(env)->uti_obdo;
282 obdo_from_la(obdo, attr, attr->la_valid);
283 lustre_set_wire_obdo(NULL, obdo, obdo);
284 bufs[buf_count] = obdo;
285 sizes[buf_count] = sizeof(*obdo);
289 if (hint != NULL && hint->dah_parent != NULL) {
290 parent_fid = lu_object_fid(&hint->dah_parent->do_lu);
291 fid_cpu_to_le(&tmp_fid, parent_fid);
292 bufs[buf_count] = &tmp_fid;
293 sizes[buf_count] = sizeof(tmp_fid);
297 rc = update_records_update_pack(env, fid, OUT_CREATE, ops, op_count,
298 max_ops_size, params, param_count,
299 max_param_size, buf_count, bufs, sizes);
302 EXPORT_SYMBOL(update_records_create_pack);
305 * Pack attr set update
307 * Pack attr_set update into update records.
309 * \param[in] env execution environment
310 * \param[in] ops ur_ops in update records
311 * \param[in|out] op_count pointer to the count of ops
312 * \param[in|out] max_op_size maximum size of the update
313 * \param[in] params ur_params in update records
314 * \param[in|out] param_count pointer to the count of params
315 * \param[in|out] max_param_size maximum size of the parameter
316 * \param[in] fid FID of the object to set attr
317 * \param[in] attr attribute of attr set
319 * \retval 0 if packing succeeds.
320 * \retval negative errno if packing fails.
322 int update_records_attr_set_pack(const struct lu_env *env,
323 struct update_ops *ops,
324 unsigned int *op_count,
325 size_t *max_ops_size,
326 struct update_params *params,
327 unsigned int *param_count,
328 size_t *max_param_size,
329 const struct lu_fid *fid,
330 const struct lu_attr *attr)
332 struct obdo *obdo = &update_env_info(env)->uti_obdo;
333 size_t size = sizeof(*obdo);
336 obdo_from_la(obdo, attr, attr->la_valid);
337 lustre_set_wire_obdo(NULL, obdo, obdo);
338 return update_records_update_pack(env, fid, OUT_ATTR_SET, ops, op_count,
339 max_ops_size, params, param_count,
341 (const void **)&obdo, &size);
343 EXPORT_SYMBOL(update_records_attr_set_pack);
346 * Pack ref add update
348 * Pack ref add update into update records.
350 * \param[in] env execution environment
351 * \param[in] ops ur_ops in update records
352 * \param[in|out] op_count pointer to the count of ops
353 * \param[in|out] max_op_size maximum size of the update
354 * \param[in] params ur_params in update records
355 * \param[in|out] param_count pointer to the count of params
356 * \param[in|out] max_param_size maximum size of the parameter
357 * \param[in] fid FID of the object to add reference
359 * \retval 0 if packing succeeds.
360 * \retval negative errno if packing fails.
362 int update_records_ref_add_pack(const struct lu_env *env,
363 struct update_ops *ops,
364 unsigned int *op_count,
365 size_t *max_ops_size,
366 struct update_params *params,
367 unsigned int *param_count,
368 size_t *max_param_size,
369 const struct lu_fid *fid)
371 return update_records_update_pack(env, fid, OUT_REF_ADD, ops, op_count,
372 max_ops_size, params, param_count,
373 max_param_size, 0, NULL, NULL);
375 EXPORT_SYMBOL(update_records_ref_add_pack);
378 * Pack ref del update
380 * Pack ref del update into update records.
382 * \param[in] env execution environment
383 * \param[in] ops ur_ops in update records
384 * \param[in|out] op_count pointer to the count of ops
385 * \param[in|out] max_op_size maximum size of the update
386 * \param[in] params ur_params in update records
387 * \param[in] param_count pointer to the count of params
388 * \param[in|out] max_param_size maximum size of the parameter
389 * \param[in] fid FID of the object to delete reference
391 * \retval 0 if packing succeeds.
392 * \retval negative errno if packing fails.
394 int update_records_ref_del_pack(const struct lu_env *env,
395 struct update_ops *ops,
396 unsigned int *op_count,
397 size_t *max_ops_size,
398 struct update_params *params,
399 unsigned int *param_count,
400 size_t *max_param_size,
401 const struct lu_fid *fid)
403 return update_records_update_pack(env, fid, OUT_REF_DEL, ops, op_count,
404 max_ops_size, params, param_count,
405 max_param_size, 0, NULL, NULL);
407 EXPORT_SYMBOL(update_records_ref_del_pack);
410 * Pack object destroy update
412 * Pack object destroy update into update records.
414 * \param[in] env execution environment
415 * \param[in] ops ur_ops in update records
416 * \param[in|out] op_count pointer to the count of ops
417 * \param[in|out] max_op_size maximum size of the update
418 * \param[in] params ur_params in update records
419 * \param[in|out] param_count pointer to the count of params
420 * \param[in|out] max_param_size maximum size of the parameter
421 * \param[in] fid FID of the object to delete reference
423 * \retval 0 if packing succeeds.
424 * \retval negative errno if packing fails.
426 int update_records_object_destroy_pack(const struct lu_env *env,
427 struct update_ops *ops,
428 unsigned int *op_count,
429 size_t *max_ops_size,
430 struct update_params *params,
431 unsigned int *param_count,
432 size_t *max_param_size,
433 const struct lu_fid *fid)
435 return update_records_update_pack(env, fid, OUT_DESTROY, ops, op_count,
436 max_ops_size, params, param_count,
437 max_param_size, 0, NULL, NULL);
439 EXPORT_SYMBOL(update_records_object_destroy_pack);
442 * Pack index insert update
444 * Pack index insert update into update records.
446 * \param[in] env execution environment
447 * \param[in] ops ur_ops in update records
448 * \param[in] op_count pointer to the count of ops
449 * \param[in|out] max_op_size maximum size of the update
450 * \param[in] params ur_params in update records
451 * \param[in] param_count pointer to the count of params
452 * \param[in|out] max_param_size maximum size of the parameter
453 * \param[in] fid FID of the object to insert index
454 * \param[in] rec record of insertion
455 * \param[in] key key of insertion
457 * \retval 0 if packing succeeds.
458 * \retval negative errno if packing fails.
460 int update_records_index_insert_pack(const struct lu_env *env,
461 struct update_ops *ops,
462 unsigned int *op_count,
463 size_t *max_ops_size,
464 struct update_params *params,
465 unsigned int *param_count,
466 size_t *max_param_size,
467 const struct lu_fid *fid,
468 const struct dt_rec *rec,
469 const struct dt_key *key)
471 struct dt_insert_rec *rec1 = (struct dt_insert_rec *)rec;
472 struct lu_fid rec_fid;
473 __u32 type = cpu_to_le32(rec1->rec_type);
474 size_t sizes[3] = { strlen((const char *)key) + 1,
477 const void *bufs[3] = { key,
481 fid_cpu_to_le(&rec_fid, rec1->rec_fid);
483 return update_records_update_pack(env, fid, OUT_INDEX_INSERT, ops,
484 op_count, max_ops_size, params,
485 param_count, max_param_size,
488 EXPORT_SYMBOL(update_records_index_insert_pack);
491 * Pack index delete update
493 * Pack index delete update into update records.
495 * \param[in] env execution environment
496 * \param[in] ops ur_ops in update records
497 * \param[in|out] op_count pointer to the count of ops
498 * \param[in|out] max_op_size maximum size of the update
499 * \param[in] params ur_params in update records
500 * \param[in|ount] param_count pointer to the count of params
501 * \param[in|out] max_param_size maximum size of the parameter
502 * \param[in] fid FID of the object to delete index
503 * \param[in] key key of deletion
505 * \retval 0 if packing succeeds.
506 * \retval negative errno if packing fails.
508 int update_records_index_delete_pack(const struct lu_env *env,
509 struct update_ops *ops,
510 unsigned int *op_count,
511 size_t *max_ops_size,
512 struct update_params *params,
513 unsigned int *param_count,
514 size_t *max_param_size,
515 const struct lu_fid *fid,
516 const struct dt_key *key)
518 size_t size = strlen((const char *)key) + 1;
520 return update_records_update_pack(env, fid, OUT_INDEX_DELETE, ops,
521 op_count, max_ops_size, params,
522 param_count, max_param_size,
523 1, (const void **)&key, &size);
525 EXPORT_SYMBOL(update_records_index_delete_pack);
528 * Pack xattr set update
530 * Pack xattr set 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|out] param_count pointer to the count of params
538 * \param[in|out] max_param_size maximum size of the parameter
539 * \param[in] fid FID of the object to set xattr
540 * \param[in] buf xattr to be set
541 * \param[in] name name of the xattr
542 * \param[in] flag flag for setting xattr
544 * \retval 0 if packing succeeds.
545 * \retval negative errno if packing fails.
547 int update_records_xattr_set_pack(const struct lu_env *env,
548 struct update_ops *ops,
549 unsigned int *op_count,
550 size_t *max_ops_size,
551 struct update_params *params,
552 unsigned int *param_count,
553 size_t *max_param_size,
554 const struct lu_fid *fid,
555 const struct lu_buf *buf, const char *name,
558 size_t sizes[3] = {strlen(name) + 1, buf->lb_len, sizeof(flag)};
559 const void *bufs[3] = {name, buf->lb_buf, &flag};
561 flag = cpu_to_le32(flag);
563 return update_records_update_pack(env, fid, OUT_XATTR_SET, ops,
564 op_count, max_ops_size, params,
565 param_count, max_param_size,
568 EXPORT_SYMBOL(update_records_xattr_set_pack);
571 * Pack xattr delete update
573 * Pack xattr delete update into update records.
575 * \param[in] env execution environment
576 * \param[in] ops ur_ops in update records
577 * \param[in|out] op_count pointer to the count of ops
578 * \param[in|out] max_op_size maximum size of the update
579 * \param[in] params ur_params in update records
580 * \param[in|out] param_count pointer to the count of params
581 * \param[in|out] max_param_size maximum size of the parameter
582 * \param[in] fid FID of the object to delete xattr
583 * \param[in] name name of the xattr
585 * \retval 0 if packing succeeds.
586 * \retval negative errno if packing fails.
588 int update_records_xattr_del_pack(const struct lu_env *env,
589 struct update_ops *ops,
590 unsigned int *op_count,
591 size_t *max_ops_size,
592 struct update_params *params,
593 unsigned int *param_count,
594 size_t *max_param_size,
595 const struct lu_fid *fid,
598 size_t size = strlen(name) + 1;
600 return update_records_update_pack(env, fid, OUT_XATTR_DEL, ops,
601 op_count, max_ops_size, params,
602 param_count, max_param_size,
603 1, (const void **)&name, &size);
605 EXPORT_SYMBOL(update_records_xattr_del_pack);
610 * Pack write update into update records.
612 * \param[in] env execution environment
613 * \param[in] ops ur_ops in update records
614 * \param[in|out] op_count pointer to the count of ops
615 * \param[in|out] max_op_size maximum size of the update
616 * \param[in] params ur_params in update records
617 * \param[in|out] param_count pointer to the count of params
618 * \param[in|out] max_param_size maximum size of the parameter
619 * \param[in] fid FID of the object to write into
620 * \param[in] buf buffer to write which includes an embedded size field
621 * \param[in] pos offet in the object to start writing at
623 * \retval 0 if packing succeeds.
624 * \retval negative errno if packing fails.
626 int update_records_write_pack(const struct lu_env *env,
627 struct update_ops *ops,
628 unsigned int *op_count,
629 size_t *max_ops_size,
630 struct update_params *params,
631 unsigned int *param_count,
632 size_t *max_param_size,
633 const struct lu_fid *fid,
634 const struct lu_buf *buf,
637 size_t sizes[2] = {buf->lb_len, sizeof(pos)};
638 const void *bufs[2] = {buf->lb_buf, &pos};
640 pos = cpu_to_le64(pos);
642 return update_records_update_pack(env, fid, OUT_WRITE, ops,
643 op_count, max_ops_size, params,
644 param_count, max_param_size,
647 EXPORT_SYMBOL(update_records_write_pack);
652 * Pack punch update into update records.
654 * \param[in] env execution environment
655 * \param[in] ops ur_ops in update records
656 * \param[in|out] op_count pointer to the count of ops
657 * \param[in|out] max_op_size maximum size of the update
658 * \param[in] params ur_params in update records
659 * \param[in|out] param_count pointer to the count of params
660 * \param[in|out] max_param_size maximum size of the parameter
661 * \param[in] fid FID of the object to write into
662 * \param[in] start start offset of punch
663 * \param[in] end end offet of punch
665 * \retval 0 if packing succeeds.
666 * \retval negative errno if packing fails.
668 int update_records_punch_pack(const struct lu_env *env,
669 struct update_ops *ops,
670 unsigned int *op_count,
671 size_t *max_ops_size,
672 struct update_params *params,
673 unsigned int *param_count,
674 size_t *max_param_size,
675 const struct lu_fid *fid,
676 __u64 start, __u64 end)
678 size_t sizes[2] = {sizeof(start), sizeof(end)};
679 const void *bufs[2] = {&start, &end};
681 start = cpu_to_le64(start);
682 end = cpu_to_le64(end);
684 return update_records_update_pack(env, fid, OUT_PUNCH, ops, op_count,
685 max_ops_size, params, param_count,
686 max_param_size, 2, bufs, sizes);
688 EXPORT_SYMBOL(update_records_punch_pack);
691 * Create update records in thandle_update_records
693 * Allocate update_records for thandle_update_records, the initial size
696 * \param[in] tur thandle_update_records where update_records will be
698 * \retval 0 if allocation succeeds.
699 * \retval negative errno if allocation fails.
701 static int tur_update_records_create(struct thandle_update_records *tur)
703 if (tur->tur_update_records != NULL)
706 OBD_ALLOC_LARGE(tur->tur_update_records,
707 UPDATE_RECORDS_BUFFER_SIZE);
709 if (tur->tur_update_records == NULL)
712 tur->tur_update_records_buf_size = UPDATE_RECORDS_BUFFER_SIZE;
718 * Extend update records
720 * Extend update_records to the new size in thandle_update_records.
722 * \param[in] tur thandle_update_records where update_records will be
724 * \retval 0 if extension succeeds.
725 * \retval negative errno if extension fails.
727 int tur_update_records_extend(struct thandle_update_records *tur,
730 struct llog_update_record *record;
732 OBD_ALLOC_LARGE(record, new_size);
736 if (tur->tur_update_records != NULL) {
737 memcpy(record, tur->tur_update_records,
738 tur->tur_update_records_buf_size);
739 OBD_FREE_LARGE(tur->tur_update_records,
740 tur->tur_update_records_buf_size);
743 tur->tur_update_records = record;
744 tur->tur_update_records_buf_size = new_size;
748 EXPORT_SYMBOL(tur_update_records_extend);
751 * Extend update records
753 * Extend update records in thandle to make sure it is able to hold
754 * the update with certain update_op and params size.
756 * \param [in] tur thandle_update_records to be extend
757 * \param [in] new_op_size update_op size of the update record
758 * \param [in] new_param_size params size of the update record
760 * \retval 0 if the update_records is being extended.
761 * \retval negative errno if the update_records is not being
764 int tur_update_extend(struct thandle_update_records *tur,
765 size_t new_op_size, size_t new_param_size)
773 record_size = llog_update_record_size(tur->tur_update_records);
774 /* extend update records buffer */
775 if (new_op_size > (tur->tur_update_records_buf_size - record_size -
776 sizeof(*tur->tur_update_records))) {
777 extend_size = round_up(new_op_size, UPDATE_RECORDS_BUFFER_SIZE);
778 rc = tur_update_records_extend(tur,
779 tur->tur_update_records_buf_size +
785 /* extend parameters buffer */
786 params_size = update_params_size(tur->tur_update_params,
787 tur->tur_update_param_count);
788 if (new_param_size > (tur->tur_update_params_buf_size -
790 extend_size = round_up(new_param_size,
791 UPDATE_PARAMS_BUFFER_SIZE);
792 rc = tur_update_params_extend(tur,
793 tur->tur_update_params_buf_size +
801 EXPORT_SYMBOL(tur_update_extend);
804 * Create update params in thandle_update_records
806 * Allocate update_params for thandle_update_records, the initial size
809 * \param[in] tur thandle_update_records where update_params will be
811 * \retval 0 if allocation succeeds.
812 * \retval negative errno if allocation fails.
814 static int tur_update_params_create(struct thandle_update_records *tur)
816 if (tur->tur_update_params != NULL)
819 OBD_ALLOC_LARGE(tur->tur_update_params, UPDATE_PARAMS_BUFFER_SIZE);
820 if (tur->tur_update_params == NULL)
823 tur->tur_update_params_buf_size = UPDATE_PARAMS_BUFFER_SIZE;
828 * Extend update params
830 * Extend update_params to the new size in thandle_update_records.
832 * \param[in] tur thandle_update_records where update_params will be
834 * \retval 0 if extension succeeds.
835 * \retval negative errno if extension fails.
837 int tur_update_params_extend(struct thandle_update_records *tur,
840 struct update_params *params;
842 OBD_ALLOC_LARGE(params, new_size);
846 if (tur->tur_update_params != NULL) {
847 memcpy(params, tur->tur_update_params,
848 tur->tur_update_params_buf_size);
849 OBD_FREE_LARGE(tur->tur_update_params,
850 tur->tur_update_params_buf_size);
853 tur->tur_update_params = params;
854 tur->tur_update_params_buf_size = new_size;
858 EXPORT_SYMBOL(tur_update_params_extend);
861 * Check and prepare whether it needs to record update.
863 * Checks if the transaction needs to record updates, and if it
864 * does, then initialize the update record buffer in the transaction.
866 * \param[in] env execution environment
867 * \param[in] th transaction handle
869 * \retval 0 if updates recording succeeds.
870 * \retval negative errno if updates recording fails.
872 int check_and_prepare_update_record(const struct lu_env *env,
873 struct thandle_update_records *tur)
875 struct llog_update_record *lur;
878 if (tur->tur_update_records == NULL) {
879 rc = tur_update_records_create(tur);
884 if (tur->tur_update_params == NULL) {
885 rc = tur_update_params_create(tur);
890 lur = tur->tur_update_records;
891 lur->lur_update_rec.ur_update_count = 0;
892 lur->lur_update_rec.ur_param_count = 0;
893 lur->lur_update_rec.ur_master_transno = 0;
894 lur->lur_update_rec.ur_batchid = 0;
895 lur->lur_update_rec.ur_flags = 0;
896 lur->lur_hdr.lrh_len = LLOG_CHUNK_SIZE;
898 tur->tur_update_param_count = 0;
903 static void update_key_fini(const struct lu_context *ctx,
904 struct lu_context_key *key, void *data)
906 struct update_thread_info *info = data;
908 if (info->uti_tur.tur_update_records != NULL)
909 OBD_FREE_LARGE(info->uti_tur.tur_update_records,
910 info->uti_tur.tur_update_records_buf_size);
911 if (info->uti_tur.tur_update_params != NULL)
912 OBD_FREE_LARGE(info->uti_tur.tur_update_params,
913 info->uti_tur.tur_update_params_buf_size);
918 /* context key constructor/destructor: update_key_init, update_key_fini */
919 LU_KEY_INIT(update, struct update_thread_info);
920 /* context key: update_thread_key */
921 LU_CONTEXT_KEY_DEFINE(update, LCT_MD_THREAD | LCT_MG_THREAD |
922 LCT_DT_THREAD | LCT_TX_HANDLE | LCT_LOCAL);
923 EXPORT_SYMBOL(update_thread_key);
924 LU_KEY_INIT_GENERIC(update);
926 void update_info_init(void)
928 update_key_init_generic(&update_thread_key, NULL);
929 lu_context_key_register(&update_thread_key);
932 void update_info_fini(void)
934 lu_context_key_degister(&update_thread_key);