Whamcloud - gitweb
LU-3536 lod: write updates to update log
[fs/lustre-release.git] / lustre / target / update_records.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2014, Intel Corporation.
24  */
25
26 /*
27  * lustre/target/update_records.c
28  *
29  * This file implement the methods to pack updates as update records, which
30  * will be written to the disk as llog record, and might be used during
31  * recovery.
32  *
33  * For cross-MDT operation, all of updates of the operation needs to be
34  * recorded in the disk, then during recovery phase, the recovery thread
35  * will retrieve and redo these updates if it needed.
36  *
37  * See comments above struct update_records for the format of update_records.
38  *
39  * Author: Di Wang <di.wang@intel.com>
40  */
41 #define DEBUG_SUBSYSTEM S_CLASS
42
43 #include <lu_target.h>
44 #include <lustre_update.h>
45 #include <obd.h>
46 #include <obd_class.h>
47 #include "tgt_internal.h"
48
49 #define UPDATE_RECORDS_BUFFER_SIZE      8192
50 #define UPDATE_PARAMS_BUFFER_SIZE       8192
51 /**
52  * Dump update record.
53  *
54  * Dump all of updates in the update_records, mostly for debugging purpose.
55  *
56  * \param[in] records   update records to be dumpped
57  * \param[in] mask      debug level mask
58  * \param[in] dump_params if dump all of updates the updates.
59  *
60  */
61 void update_records_dump(const struct update_records *records,
62                          unsigned int mask, bool dump_updates)
63 {
64         const struct update_ops *ops;
65         const struct update_op  *op = NULL;
66         struct update_params    *params;
67         unsigned int            i;
68
69         ops = &records->ur_ops;
70         params = update_records_get_params(records);
71
72         CDEBUG(mask, "ops = %d params_count = %d\n", records->ur_update_count,
73                records->ur_param_count);
74
75         if (records->ur_update_count == 0)
76                 return;
77
78         if (!dump_updates)
79                 return;
80
81         op = &ops->uops_op[0];
82         for (i = 0; i < records->ur_update_count; i++) {
83                 unsigned int j;
84
85                 CDEBUG(mask, "update %dth "DFID" %s params_count = %hu\n", i,
86                        PFID(&op->uop_fid), update_op_str(op->uop_type),
87                        op->uop_param_count);
88
89                 for (j = 0;  j < op->uop_param_count; j++) {
90                         struct object_update_param *param;
91
92                         param = update_params_get_param(params,
93                                         (unsigned int)op->uop_params_off[j],
94                                         records->ur_param_count);
95
96                         LASSERT(param != NULL);
97                         CDEBUG(mask, "param = %p %dth off = %hu size = %hu\n",
98                                param, j, op->uop_params_off[j], param->oup_len);
99                 }
100
101                 op = update_op_next_op(op);
102         }
103 }
104
105 /**
106  * Pack parameters to update records
107  *
108  * Find and insert parameter to update records, if the parameter
109  * already exists in \a params, then just return the offset of this
110  * parameter, otherwise insert the parameter and return its offset
111  *
112  * \param[in] params    update params in which to insert parameter
113  * \param[in] new_param parameters to be inserted.
114  * \param[in] new_param_size    the size of \a new_param
115  *
116  * \retval              index inside \a params if parameter insertion
117  *                      succeeds.
118  * \retval              negative errno if it fails.
119  */
120 static unsigned int update_records_param_pack(struct update_params *params,
121                                               const void *new_param,
122                                               size_t new_param_size,
123                                               unsigned int *param_count)
124 {
125         struct object_update_param      *param;
126         unsigned int                    i;
127
128         for (i = 0; i < *param_count; i++) {
129                 struct object_update_param *param;
130
131                 param = update_params_get_param(params, i, *param_count);
132                 if ((new_param == NULL && param->oup_len == new_param_size) ||
133                     (param->oup_len == new_param_size &&
134                      memcmp(param->oup_buf, new_param, new_param_size) == 0))
135                         /* Found the parameter and return its index */
136                         return i;
137         }
138
139         param = (struct object_update_param *)((char *)params +
140                                 update_params_size(params, *param_count));
141
142         param->oup_len = new_param_size;
143         if (new_param != NULL)
144                 memcpy(param->oup_buf, new_param, new_param_size);
145
146         *param_count = *param_count + 1;
147
148         return *param_count - 1;
149 }
150
151 /**
152  * Pack update to update records
153  *
154  * Pack the update and its parameters to the update records. First it will
155  * insert parameters, get the offset of these parameter, then fill the
156  * update with these offset. If insertion exceed the maximum size of
157  * current update records, it will return -E2BIG here, and the caller might
158  * extend the update_record size \see lod_updates_pack.
159  *
160  * \param[in] env       execution environment
161  * \param[in] fid       FID of the update.
162  * \param[in] op_type   operation type of the update
163  * \param[in] ops       ur_ops in update records
164  * \param[in|out] op_count      pointer to the count of ops
165  * \param[in|out] max_op_size maximum size of the update
166  * \param[in] params    ur_params in update records
167  * \param[in|out] param_count   pointer to the count of params
168  * \param[in|out] max_param_size maximum size of the parameter
169  * \param[in] param_bufs        buffers of parameters
170  * \param[in] params_buf_count  the count of the parameter buffers
171  * \param[in] param_size        sizes of parameters
172  *
173  * \retval              0 if packing succeeds
174  * \retval              negative errno if packing fails
175  */
176 static int update_records_update_pack(const struct lu_env *env,
177                                       const struct lu_fid *fid,
178                                       enum update_type op_type,
179                                       struct update_ops *ops,
180                                       unsigned int *op_count,
181                                       size_t *max_op_size,
182                                       struct update_params *params,
183                                       unsigned int *param_count,
184                                       size_t *max_param_size,
185                                       unsigned int param_bufs_count,
186                                       const void **param_bufs,
187                                       size_t *param_sizes)
188 {
189         struct update_op        *op;
190         size_t                  total_param_sizes = 0;
191         int                     index;
192         unsigned int            i;
193
194         /* Check whether the packing exceeding the maximum update size */
195         if (unlikely(*max_op_size < update_op_size(param_bufs_count))) {
196                 CDEBUG(D_INFO, "max_op_size = %zu update_op = %zu\n",
197                        *max_op_size, update_op_size(param_bufs_count));
198                 *max_op_size = update_op_size(param_bufs_count);
199                 return -E2BIG;
200         }
201
202         for (i = 0; i < param_bufs_count; i++)
203                 total_param_sizes +=
204                         cfs_size_round(sizeof(struct object_update_param) +
205                                        param_sizes[i]);
206
207         /* Check whether the packing exceeding the maximum parameter size */
208         if (unlikely(*max_param_size < total_param_sizes)) {
209                 CDEBUG(D_INFO, "max_param_size = %zu params size = %zu\n",
210                        *max_param_size, total_param_sizes);
211
212                 *max_param_size = total_param_sizes;
213                 return -E2BIG;
214         }
215
216         op = update_ops_get_op(ops, *op_count, *op_count);
217         op->uop_fid = *fid;
218         op->uop_type = op_type;
219         op->uop_param_count = param_bufs_count;
220         for (i = 0; i < param_bufs_count; i++) {
221                 index = update_records_param_pack(params, param_bufs[i],
222                                                   param_sizes[i], param_count);
223                 if (index < 0)
224                         return index;
225
226                 CDEBUG(D_INFO, "%s %uth param offset = %d size = %zu\n",
227                        update_op_str(op_type), i, index, param_sizes[i]);
228
229                 op->uop_params_off[i] = index;
230         }
231         CDEBUG(D_INFO, "%huth "DFID" %s param_count = %u\n",
232                *op_count, PFID(fid), update_op_str(op_type), *param_count);
233
234         *op_count = *op_count + 1;
235
236         return 0;
237 }
238
239 /**
240  * Pack create update
241  *
242  * Pack create update into update records.
243  *
244  * \param[in] env       execution environment
245  * \param[in] ops       ur_ops in update records
246  * \param[in|out] op_count      pointer to the count of ops
247  * \param[in|out] max_op_size maximum size of the update
248  * \param[in] params    ur_params in update records
249  * \param[in|out] param_count   pointer to the count of params
250  * \param[in|out] max_param_size maximum size of the parameter
251  * \param[in] fid       FID of the object to be created
252  * \param[in] attr      attribute of the object to be created
253  * \param[in] hint      creation hint
254  * \param[in] dof       creation format information
255  *
256  * \retval              0 if packing succeeds.
257  * \retval              negative errno if packing fails.
258  */
259 int update_records_create_pack(const struct lu_env *env,
260                                struct update_ops *ops,
261                                unsigned int *op_count,
262                                size_t *max_ops_size,
263                                struct update_params *params,
264                                unsigned int *param_count,
265                                size_t *max_param_size,
266                                const struct lu_fid *fid,
267                                const struct lu_attr *attr,
268                                const struct dt_allocation_hint *hint,
269                                struct dt_object_format *dof)
270 {
271         size_t                  sizes[2];
272         const void              *bufs[2];
273         int                     buf_count = 0;
274         const struct lu_fid     *parent_fid = NULL;
275         struct lu_fid           tmp_fid;
276         int                     rc;
277         struct obdo             *obdo;
278
279         if (attr != NULL) {
280                 obdo = &update_env_info(env)->uti_obdo;
281                 obdo->o_valid = 0;
282                 obdo_from_la(obdo, attr, attr->la_valid);
283                 lustre_set_wire_obdo(NULL, obdo, obdo);
284                 bufs[buf_count] = obdo;
285                 sizes[buf_count] = sizeof(*obdo);
286                 buf_count++;
287         }
288
289         if (hint != NULL && hint->dah_parent != NULL) {
290                 parent_fid = lu_object_fid(&hint->dah_parent->do_lu);
291                 fid_cpu_to_le(&tmp_fid, parent_fid);
292                 bufs[buf_count] = &tmp_fid;
293                 sizes[buf_count] = sizeof(tmp_fid);
294                 buf_count++;
295         }
296
297         rc = update_records_update_pack(env, fid, OUT_CREATE, ops, op_count,
298                                         max_ops_size, params, param_count,
299                                         max_param_size, buf_count, bufs, sizes);
300         return rc;
301 }
302 EXPORT_SYMBOL(update_records_create_pack);
303
304 /**
305  * Pack attr set update
306  *
307  * Pack attr_set update into update records.
308  *
309  * \param[in] env       execution environment
310  * \param[in] ops       ur_ops in update records
311  * \param[in|out] op_count pointer to the count of ops
312  * \param[in|out] max_op_size maximum size of the update
313  * \param[in] params    ur_params in update records
314  * \param[in|out] param_count pointer to the count of params
315  * \param[in|out] max_param_size maximum size of the parameter
316  * \param[in] fid       FID of the object to set attr
317  * \param[in] attr      attribute of attr set
318  *
319  * \retval              0 if packing succeeds.
320  * \retval              negative errno if packing fails.
321  */
322 int update_records_attr_set_pack(const struct lu_env *env,
323                                  struct update_ops *ops,
324                                  unsigned int *op_count,
325                                  size_t *max_ops_size,
326                                  struct update_params *params,
327                                  unsigned int *param_count,
328                                  size_t *max_param_size,
329                                  const struct lu_fid *fid,
330                                  const struct lu_attr *attr)
331 {
332         struct obdo *obdo = &update_env_info(env)->uti_obdo;
333         size_t size = sizeof(*obdo);
334
335         obdo->o_valid = 0;
336         obdo_from_la(obdo, attr, attr->la_valid);
337         lustre_set_wire_obdo(NULL, obdo, obdo);
338         return update_records_update_pack(env, fid, OUT_ATTR_SET, ops, op_count,
339                                           max_ops_size, params, param_count,
340                                           max_param_size, 1,
341                                           (const void **)&obdo, &size);
342 }
343 EXPORT_SYMBOL(update_records_attr_set_pack);
344
345 /**
346  * Pack ref add update
347  *
348  * Pack ref add update into update records.
349  *
350  * \param[in] env       execution environment
351  * \param[in] ops       ur_ops in update records
352  * \param[in|out] op_count pointer to the count of ops
353  * \param[in|out] max_op_size maximum size of the update
354  * \param[in] params    ur_params in update records
355  * \param[in|out] param_count pointer to the count of params
356  * \param[in|out] max_param_size maximum size of the parameter
357  * \param[in] fid       FID of the object to add reference
358  *
359  * \retval              0 if packing succeeds.
360  * \retval              negative errno if packing fails.
361  */
362 int update_records_ref_add_pack(const struct lu_env *env,
363                                 struct update_ops *ops,
364                                 unsigned int *op_count,
365                                 size_t *max_ops_size,
366                                 struct update_params *params,
367                                 unsigned int *param_count,
368                                 size_t *max_param_size,
369                                 const struct lu_fid *fid)
370 {
371         return update_records_update_pack(env, fid, OUT_REF_ADD, ops, op_count,
372                                           max_ops_size, params, param_count,
373                                           max_param_size, 0, NULL, NULL);
374 }
375 EXPORT_SYMBOL(update_records_ref_add_pack);
376
377 /**
378  * Pack ref del update
379  *
380  * Pack ref del update into update records.
381  *
382  * \param[in] env       execution environment
383  * \param[in] ops       ur_ops in update records
384  * \param[in|out] op_count      pointer to the count of ops
385  * \param[in|out] max_op_size maximum size of the update
386  * \param[in] params    ur_params in update records
387  * \param[in] param_count       pointer to the count of params
388  * \param[in|out] max_param_size maximum size of the parameter
389  * \param[in] fid       FID of the object to delete reference
390  *
391  * \retval              0 if packing succeeds.
392  * \retval              negative errno if packing fails.
393  */
394 int update_records_ref_del_pack(const struct lu_env *env,
395                                 struct update_ops *ops,
396                                 unsigned int *op_count,
397                                 size_t *max_ops_size,
398                                 struct update_params *params,
399                                 unsigned int *param_count,
400                                 size_t *max_param_size,
401                                 const struct lu_fid *fid)
402 {
403         return update_records_update_pack(env, fid, OUT_REF_DEL, ops, op_count,
404                                           max_ops_size, params, param_count,
405                                           max_param_size, 0, NULL, NULL);
406 }
407 EXPORT_SYMBOL(update_records_ref_del_pack);
408
409 /**
410  * Pack object destroy update
411  *
412  * Pack object destroy update into update records.
413  *
414  * \param[in] env       execution environment
415  * \param[in] ops       ur_ops in update records
416  * \param[in|out] op_count pointer to the count of ops
417  * \param[in|out] max_op_size maximum size of the update
418  * \param[in] params    ur_params in update records
419  * \param[in|out] param_count   pointer to the count of params
420  * \param[in|out] max_param_size maximum size of the parameter
421  * \param[in] fid       FID of the object to delete reference
422  *
423  * \retval              0 if packing succeeds.
424  * \retval              negative errno if packing fails.
425  */
426 int update_records_object_destroy_pack(const struct lu_env *env,
427                                        struct update_ops *ops,
428                                        unsigned int *op_count,
429                                        size_t *max_ops_size,
430                                        struct update_params *params,
431                                        unsigned int *param_count,
432                                        size_t *max_param_size,
433                                        const struct lu_fid *fid)
434 {
435         return update_records_update_pack(env, fid, OUT_DESTROY, ops, op_count,
436                                           max_ops_size, params, param_count,
437                                           max_param_size, 0, NULL, NULL);
438 }
439 EXPORT_SYMBOL(update_records_object_destroy_pack);
440
441 /**
442  * Pack index insert update
443  *
444  * Pack index insert update into update records.
445  *
446  * \param[in] env       execution environment
447  * \param[in] ops       ur_ops in update records
448  * \param[in] op_count  pointer to the count of ops
449  * \param[in|out] max_op_size maximum size of the update
450  * \param[in] params    ur_params in update records
451  * \param[in] param_count       pointer to the count of params
452  * \param[in|out] max_param_size maximum size of the parameter
453  * \param[in] fid       FID of the object to insert index
454  * \param[in] rec       record of insertion
455  * \param[in] key       key of insertion
456  *
457  * \retval              0 if packing succeeds.
458  * \retval              negative errno if packing fails.
459  */
460 int update_records_index_insert_pack(const struct lu_env *env,
461                                      struct update_ops *ops,
462                                      unsigned int *op_count,
463                                      size_t *max_ops_size,
464                                      struct update_params *params,
465                                      unsigned int *param_count,
466                                      size_t *max_param_size,
467                                      const struct lu_fid *fid,
468                                      const struct dt_rec *rec,
469                                      const struct dt_key *key)
470 {
471         struct dt_insert_rec       *rec1 = (struct dt_insert_rec *)rec;
472         struct lu_fid              rec_fid;
473         __u32                      type = cpu_to_le32(rec1->rec_type);
474         size_t                     sizes[3] = { strlen((const char *)key) + 1,
475                                                 sizeof(rec_fid),
476                                                 sizeof(type) };
477         const void                 *bufs[3] = { key,
478                                                 &rec_fid,
479                                                 &type };
480
481         fid_cpu_to_le(&rec_fid, rec1->rec_fid);
482
483         return update_records_update_pack(env, fid, OUT_INDEX_INSERT, ops,
484                                           op_count, max_ops_size, params,
485                                           param_count, max_param_size,
486                                           3, bufs, sizes);
487 }
488 EXPORT_SYMBOL(update_records_index_insert_pack);
489
490 /**
491  * Pack index delete update
492  *
493  * Pack index delete update into update records.
494  *
495  * \param[in] env       execution environment
496  * \param[in] ops       ur_ops in update records
497  * \param[in|out] op_count      pointer to the count of ops
498  * \param[in|out] max_op_size maximum size of the update
499  * \param[in] params    ur_params in update records
500  * \param[in|ount] param_count  pointer to the count of params
501  * \param[in|out] max_param_size maximum size of the parameter
502  * \param[in] fid       FID of the object to delete index
503  * \param[in] key       key of deletion
504  *
505  * \retval              0 if packing succeeds.
506  * \retval              negative errno if packing fails.
507  */
508 int update_records_index_delete_pack(const struct lu_env *env,
509                                      struct update_ops *ops,
510                                      unsigned int *op_count,
511                                      size_t *max_ops_size,
512                                      struct update_params *params,
513                                      unsigned int *param_count,
514                                      size_t *max_param_size,
515                                      const struct lu_fid *fid,
516                                      const struct dt_key *key)
517 {
518         size_t size = strlen((const char *)key) + 1;
519
520         return update_records_update_pack(env, fid, OUT_INDEX_DELETE, ops,
521                                           op_count, max_ops_size, params,
522                                           param_count, max_param_size,
523                                           1, (const void **)&key, &size);
524 }
525 EXPORT_SYMBOL(update_records_index_delete_pack);
526
527 /**
528  * Pack xattr set update
529  *
530  * Pack xattr set update into update records.
531  *
532  * \param[in] env       execution environment
533  * \param[in] ops       ur_ops in update records
534  * \param[in|out] op_count      pointer to the count of ops
535  * \param[in|out] max_op_size maximum size of the update
536  * \param[in] params    ur_params in update records
537  * \param[in|out] param_count   pointer to the count of params
538  * \param[in|out] max_param_size maximum size of the parameter
539  * \param[in] fid       FID of the object to set xattr
540  * \param[in] buf       xattr to be set
541  * \param[in] name      name of the xattr
542  * \param[in] flag      flag for setting xattr
543  *
544  * \retval              0 if packing succeeds.
545  * \retval              negative errno if packing fails.
546  */
547 int update_records_xattr_set_pack(const struct lu_env *env,
548                                   struct update_ops *ops,
549                                   unsigned int *op_count,
550                                   size_t *max_ops_size,
551                                   struct update_params *params,
552                                   unsigned int *param_count,
553                                   size_t *max_param_size,
554                                   const struct lu_fid *fid,
555                                   const struct lu_buf *buf, const char *name,
556                                   __u32 flag)
557 {
558         size_t  sizes[3] = {strlen(name) + 1, buf->lb_len, sizeof(flag)};
559         const void *bufs[3] = {name, buf->lb_buf, &flag};
560
561         flag = cpu_to_le32(flag);
562
563         return update_records_update_pack(env, fid, OUT_XATTR_SET, ops,
564                                           op_count, max_ops_size, params,
565                                           param_count, max_param_size,
566                                           3, bufs, sizes);
567 }
568 EXPORT_SYMBOL(update_records_xattr_set_pack);
569
570 /**
571  * Pack xattr delete update
572  *
573  * Pack xattr delete update into update records.
574  *
575  * \param[in] env       execution environment
576  * \param[in] ops       ur_ops in update records
577  * \param[in|out] op_count      pointer to the count of ops
578  * \param[in|out] max_op_size maximum size of the update
579  * \param[in] params    ur_params in update records
580  * \param[in|out] param_count   pointer to the count of params
581  * \param[in|out] max_param_size maximum size of the parameter
582  * \param[in] fid       FID of the object to delete xattr
583  * \param[in] name      name of the xattr
584  *
585  * \retval              0 if packing succeeds.
586  * \retval              negative errno if packing fails.
587  */
588 int update_records_xattr_del_pack(const struct lu_env *env,
589                                   struct update_ops *ops,
590                                   unsigned int *op_count,
591                                   size_t *max_ops_size,
592                                   struct update_params *params,
593                                   unsigned int *param_count,
594                                   size_t *max_param_size,
595                                   const struct lu_fid *fid,
596                                   const char *name)
597 {
598         size_t  size = strlen(name) + 1;
599
600         return update_records_update_pack(env, fid, OUT_XATTR_DEL, ops,
601                                           op_count, max_ops_size, params,
602                                           param_count, max_param_size,
603                                           1, (const void **)&name, &size);
604 }
605 EXPORT_SYMBOL(update_records_xattr_del_pack);
606
607 /**
608  * Pack write update
609  *
610  * Pack write update into update records.
611  *
612  * \param[in] env       execution environment
613  * \param[in] ops       ur_ops in update records
614  * \param[in|out] op_count      pointer to the count of ops
615  * \param[in|out] max_op_size maximum size of the update
616  * \param[in] params    ur_params in update records
617  * \param[in|out] param_count   pointer to the count of params
618  * \param[in|out] max_param_size maximum size of the parameter
619  * \param[in] fid       FID of the object to write into
620  * \param[in] buf       buffer to write which includes an embedded size field
621  * \param[in] pos       offet in the object to start writing at
622  *
623  * \retval              0 if packing succeeds.
624  * \retval              negative errno if packing fails.
625  */
626 int update_records_write_pack(const struct lu_env *env,
627                               struct update_ops *ops,
628                               unsigned int *op_count,
629                               size_t *max_ops_size,
630                               struct update_params *params,
631                               unsigned int *param_count,
632                               size_t *max_param_size,
633                               const struct lu_fid *fid,
634                               const struct lu_buf *buf,
635                               __u64 pos)
636 {
637         size_t          sizes[2] = {buf->lb_len, sizeof(pos)};
638         const void      *bufs[2] = {buf->lb_buf, &pos};
639
640         pos = cpu_to_le64(pos);
641
642         return update_records_update_pack(env, fid, OUT_WRITE, ops,
643                                           op_count, max_ops_size, params,
644                                           param_count, max_param_size,
645                                           2, bufs, sizes);
646 }
647 EXPORT_SYMBOL(update_records_write_pack);
648
649 /**
650  * Pack punch
651  *
652  * Pack punch update into update records.
653  *
654  * \param[in] env       execution environment
655  * \param[in] ops       ur_ops in update records
656  * \param[in|out] op_count      pointer to the count of ops
657  * \param[in|out] max_op_size maximum size of the update
658  * \param[in] params    ur_params in update records
659  * \param[in|out] param_count   pointer to the count of params
660  * \param[in|out] max_param_size maximum size of the parameter
661  * \param[in] fid       FID of the object to write into
662  * \param[in] start     start offset of punch
663  * \param[in] end       end offet of punch
664  *
665  * \retval              0 if packing succeeds.
666  * \retval              negative errno if packing fails.
667  */
668 int update_records_punch_pack(const struct lu_env *env,
669                               struct update_ops *ops,
670                               unsigned int *op_count,
671                               size_t *max_ops_size,
672                               struct update_params *params,
673                               unsigned int *param_count,
674                               size_t *max_param_size,
675                               const struct lu_fid *fid,
676                               __u64 start, __u64 end)
677 {
678         size_t          sizes[2] = {sizeof(start), sizeof(end)};
679         const void      *bufs[2] = {&start, &end};
680
681         start = cpu_to_le64(start);
682         end = cpu_to_le64(end);
683
684         return update_records_update_pack(env, fid, OUT_PUNCH, ops, op_count,
685                                           max_ops_size, params, param_count,
686                                           max_param_size, 2, bufs, sizes);
687 }
688 EXPORT_SYMBOL(update_records_punch_pack);
689
690 /**
691  * Create update records in thandle_update_records
692  *
693  * Allocate update_records for thandle_update_records, the initial size
694  * will be 4KB.
695  *
696  * \param[in] tur       thandle_update_records where update_records will be
697  *                      allocated
698  * \retval              0 if allocation succeeds.
699  * \retval              negative errno if allocation fails.
700  */
701 static int tur_update_records_create(struct thandle_update_records *tur)
702 {
703         if (tur->tur_update_records != NULL)
704                 return 0;
705
706         OBD_ALLOC_LARGE(tur->tur_update_records,
707                         UPDATE_RECORDS_BUFFER_SIZE);
708
709         if (tur->tur_update_records == NULL)
710                 return -ENOMEM;
711
712         tur->tur_update_records_buf_size = UPDATE_RECORDS_BUFFER_SIZE;
713
714         return 0;
715 }
716
717 /**
718  * Extend update records
719  *
720  * Extend update_records to the new size in thandle_update_records.
721  *
722  * \param[in] tur       thandle_update_records where update_records will be
723  *                      extended.
724  * \retval              0 if extension succeeds.
725  * \retval              negative errno if extension fails.
726  */
727 int tur_update_records_extend(struct thandle_update_records *tur,
728                               size_t new_size)
729 {
730         struct llog_update_record       *record;
731
732         OBD_ALLOC_LARGE(record, new_size);
733         if (record == NULL)
734                 return -ENOMEM;
735
736         if (tur->tur_update_records != NULL) {
737                 memcpy(record, tur->tur_update_records,
738                        tur->tur_update_records_buf_size);
739                 OBD_FREE_LARGE(tur->tur_update_records,
740                                tur->tur_update_records_buf_size);
741         }
742
743         tur->tur_update_records = record;
744         tur->tur_update_records_buf_size = new_size;
745
746         return 0;
747 }
748 EXPORT_SYMBOL(tur_update_records_extend);
749
750 /**
751  * Extend update records
752  *
753  * Extend update records in thandle to make sure it is able to hold
754  * the update with certain update_op and params size.
755  *
756  * \param [in] tur      thandle_update_records to be extend
757  * \param [in] new_op_size update_op size of the update record
758  * \param [in] new_param_size params size of the update record
759  *
760  * \retval              0 if the update_records is being extended.
761  * \retval              negative errno if the update_records is not being
762  *                      extended.
763  */
764 int tur_update_extend(struct thandle_update_records *tur,
765                       size_t new_op_size, size_t new_param_size)
766 {
767         size_t record_size;
768         size_t params_size;
769         size_t extend_size;
770         int rc;
771         ENTRY;
772
773         record_size = llog_update_record_size(tur->tur_update_records);
774         /* extend update records buffer */
775         if (new_op_size > (tur->tur_update_records_buf_size - record_size -
776                            sizeof(*tur->tur_update_records))) {
777                 extend_size = round_up(new_op_size, UPDATE_RECORDS_BUFFER_SIZE);
778                 rc = tur_update_records_extend(tur,
779                                 tur->tur_update_records_buf_size +
780                                 extend_size);
781                 if (rc != 0)
782                         RETURN(rc);
783         }
784
785         /* extend parameters buffer */
786         params_size = update_params_size(tur->tur_update_params,
787                                          tur->tur_update_param_count);
788         if (new_param_size > (tur->tur_update_params_buf_size -
789                               params_size)) {
790                 extend_size = round_up(new_param_size,
791                                        UPDATE_PARAMS_BUFFER_SIZE);
792                 rc = tur_update_params_extend(tur,
793                                 tur->tur_update_params_buf_size +
794                                 extend_size);
795                 if (rc != 0)
796                         RETURN(rc);
797         }
798
799         RETURN(0);
800 }
801 EXPORT_SYMBOL(tur_update_extend);
802
803 /**
804  * Create update params in thandle_update_records
805  *
806  * Allocate update_params for thandle_update_records, the initial size
807  * will be 4KB.
808  *
809  * \param[in] tur       thandle_update_records where update_params will be
810  *                      allocated
811  * \retval              0 if allocation succeeds.
812  * \retval              negative errno if allocation fails.
813  */
814 static int tur_update_params_create(struct thandle_update_records *tur)
815 {
816         if (tur->tur_update_params != NULL)
817                 return 0;
818
819         OBD_ALLOC_LARGE(tur->tur_update_params, UPDATE_PARAMS_BUFFER_SIZE);
820         if (tur->tur_update_params == NULL)
821                 return -ENOMEM;
822
823         tur->tur_update_params_buf_size = UPDATE_PARAMS_BUFFER_SIZE;
824         return 0;
825 }
826
827 /**
828  * Extend update params
829  *
830  * Extend update_params to the new size in thandle_update_records.
831  *
832  * \param[in] tur       thandle_update_records where update_params will be
833  *                      extended.
834  * \retval              0 if extension succeeds.
835  * \retval              negative errno if extension fails.
836  */
837 int tur_update_params_extend(struct thandle_update_records *tur,
838                              size_t new_size)
839 {
840         struct update_params    *params;
841
842         OBD_ALLOC_LARGE(params, new_size);
843         if (params == NULL)
844                 return -ENOMEM;
845
846         if (tur->tur_update_params != NULL) {
847                 memcpy(params, tur->tur_update_params,
848                        tur->tur_update_params_buf_size);
849                 OBD_FREE_LARGE(tur->tur_update_params,
850                                tur->tur_update_params_buf_size);
851         }
852
853         tur->tur_update_params = params;
854         tur->tur_update_params_buf_size = new_size;
855
856         return 0;
857 }
858 EXPORT_SYMBOL(tur_update_params_extend);
859
860 /**
861  * Check and prepare whether it needs to record update.
862  *
863  * Checks if the transaction needs to record updates, and if it
864  * does, then initialize the update record buffer in the transaction.
865  *
866  * \param[in] env       execution environment
867  * \param[in] th        transaction handle
868  *
869  * \retval              0 if updates recording succeeds.
870  * \retval              negative errno if updates recording fails.
871  */
872 int check_and_prepare_update_record(const struct lu_env *env,
873                                     struct thandle *th)
874 {
875         struct thandle_update_records   *tur;
876         struct llog_update_record       *lur;
877         struct top_thandle              *top_th;
878         struct sub_thandle              *lst;
879         int                             rc;
880         bool                            record_update = false;
881         ENTRY;
882
883         top_th = container_of(th, struct top_thandle, tt_super);
884         /* Check if it needs to record updates for this transaction */
885         list_for_each_entry(lst, &top_th->tt_sub_thandle_list, st_sub_list) {
886                 if (lst->st_record_update) {
887                         record_update = true;
888                         break;
889                 }
890         }
891         if (!record_update)
892                 RETURN(0);
893
894         if (top_th->tt_update_records != NULL)
895                 RETURN(0);
896
897         tur = &update_env_info(env)->uti_tur;
898         if (tur->tur_update_records == NULL) {
899                 rc = tur_update_records_create(tur);
900                 if (rc < 0)
901                         RETURN(rc);
902         }
903
904         if (tur->tur_update_params == NULL) {
905                 rc = tur_update_params_create(tur);
906                 if (rc < 0)
907                         RETURN(rc);
908         }
909
910         lur = tur->tur_update_records;
911         lur->lur_update_rec.ur_update_count = 0;
912         lur->lur_update_rec.ur_param_count = 0;
913         lur->lur_update_rec.ur_master_transno = 0;
914         lur->lur_update_rec.ur_batchid = 0;
915         lur->lur_update_rec.ur_flags = 0;
916         lur->lur_hdr.lrh_len = LLOG_CHUNK_SIZE;
917
918         tur->tur_update_param_count = 0;
919
920         top_th->tt_update_records = tur;
921
922         RETURN(0);
923 }
924 EXPORT_SYMBOL(check_and_prepare_update_record);
925
926 /**
927  * Merge params into the update records
928  *
929  * Merge params and ops into the update records attached to top th.
930  * During transaction execution phase, parameters and update ops
931  * are collected in two different buffers (see lod_updates_pack()),
932  * then in transaction stop, it needs to be merged them in one
933  * buffer, then being written into the update log.
934  *
935  * \param[in] env       execution environment
936  * \param[in] tur       thandle update records whose updates and
937  *                      parameters are merged
938  *
939  * \retval              0 if merging succeeds.
940  * \retval              negaitive errno if merging fails.
941  */
942 int merge_params_updates_buf(const struct lu_env *env,
943                              struct thandle_update_records *tur)
944 {
945         struct llog_update_record *lur;
946         struct update_params *params;
947         size_t params_size;
948         size_t record_size;
949
950         if (tur->tur_update_records == NULL ||
951             tur->tur_update_params == NULL)
952                 return 0;
953
954         lur = tur->tur_update_records;
955         /* Extends the update records buffer if needed */
956         params_size = update_params_size(tur->tur_update_params,
957                                          tur->tur_update_param_count);
958         record_size = llog_update_record_size(lur);
959         if (cfs_size_round(record_size + params_size) >
960                                 tur->tur_update_records_buf_size) {
961                 int rc;
962
963                 rc = tur_update_records_extend(tur, record_size + params_size);
964                 if (rc < 0)
965                         return rc;
966                 lur = tur->tur_update_records;
967         }
968
969         params = update_records_get_params(&lur->lur_update_rec);
970         memcpy(params, tur->tur_update_params, params_size);
971         lur->lur_update_rec.ur_param_count = tur->tur_update_param_count;
972         return 0;
973 }
974 EXPORT_SYMBOL(merge_params_updates_buf);
975
976 static void update_key_fini(const struct lu_context *ctx,
977                             struct lu_context_key *key, void *data)
978 {
979         struct update_thread_info *info = data;
980
981         if (info->uti_tur.tur_update_records != NULL)
982                 OBD_FREE_LARGE(info->uti_tur.tur_update_records,
983                                info->uti_tur.tur_update_records_buf_size);
984         if (info->uti_tur.tur_update_params != NULL)
985                 OBD_FREE_LARGE(info->uti_tur.tur_update_params,
986                                info->uti_tur.tur_update_params_buf_size);
987
988         OBD_FREE_PTR(info);
989 }
990
991 /* context key constructor/destructor: update_key_init, update_key_fini */
992 LU_KEY_INIT(update, struct update_thread_info);
993 /* context key: update_thread_key */
994 LU_CONTEXT_KEY_DEFINE(update, LCT_MD_THREAD | LCT_MG_THREAD |
995                               LCT_DT_THREAD | LCT_TX_HANDLE | LCT_LOCAL);
996 EXPORT_SYMBOL(update_thread_key);
997 LU_KEY_INIT_GENERIC(update);
998
999 void update_info_init(void)
1000 {
1001         update_key_init_generic(&update_thread_key, NULL);
1002         lu_context_key_register(&update_thread_key);
1003 }
1004
1005 void update_info_fini(void)
1006 {
1007         lu_context_key_degister(&update_thread_key);
1008 }