Whamcloud - gitweb
LU-14976 ptlrpc: align function names with param names
[fs/lustre-release.git] / lustre / obdclass / llog_osd.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) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2012, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  */
31 /*
32  * lustre/obdclass/llog_osd.c
33  *
34  * Low level llog routines on top of OSD API
35  *
36  * This file provides set of methods for llog operations on top of
37  * dt_device. It contains all supported llog_operations interfaces and
38  * supplimental functions.
39  *
40  * Author: Alexey Zhuravlev <alexey.zhuravlev@intel.com>
41  * Author: Mikhail Pershin <mike.pershin@intel.com>
42  */
43
44 #define DEBUG_SUBSYSTEM S_LOG
45
46 #include <linux/delay.h>
47
48 #include <dt_object.h>
49 #include <llog_swab.h>
50 #include <lustre_fid.h>
51 #include <obd.h>
52 #include <obd_class.h>
53
54 #include "llog_internal.h"
55 #include "local_storage.h"
56
57 /**
58  * Implementation of the llog_operations::lop_declare_create
59  *
60  * This function is a wrapper over local_storage API function
61  * local_object_declare_create().
62  *
63  * \param[in] env       execution environment
64  * \param[in] los       local_storage for bottom storage device
65  * \param[in] o         dt_object to create
66  * \param[in] th        current transaction handle
67  *
68  * \retval              0 on successful declaration of the new object
69  * \retval              negative error if declaration was failed
70  */
71 static int llog_osd_declare_new_object(const struct lu_env *env,
72                                        struct local_oid_storage *los,
73                                        struct dt_object *o,
74                                        struct thandle *th)
75 {
76         struct llog_thread_info *lgi = llog_info(env);
77
78         lgi->lgi_attr.la_valid = LA_MODE;
79         lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
80         lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
81
82         return local_object_declare_create(env, los, o, &lgi->lgi_attr,
83                                            &lgi->lgi_dof, th);
84 }
85
86 /**
87  * Implementation of the llog_operations::lop_create
88  *
89  * This function is a wrapper over local_storage API function
90  * local_object_create().
91  *
92  * \param[in] env       execution environment
93  * \param[in] los       local_storage for bottom storage device
94  * \param[in] o         dt_object to create
95  * \param[in] th        current transaction handle
96  *
97  * \retval              0 on successful creation of the new object
98  * \retval              negative error if creation was failed
99  */
100 static int llog_osd_create_new_object(const struct lu_env *env,
101                                       struct local_oid_storage *los,
102                                       struct dt_object *o,
103                                       struct thandle *th)
104 {
105         struct llog_thread_info *lgi = llog_info(env);
106
107         lgi->lgi_attr.la_valid = LA_MODE;
108         lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
109         lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
110
111         return local_object_create(env, los, o, &lgi->lgi_attr,
112                                    &lgi->lgi_dof, th);
113 }
114
115 /**
116  * Implementation of the llog_operations::lop_exist
117  *
118  * This function checks that llog exists on storage.
119  *
120  * \param[in] handle    llog handle of the current llog
121  *
122  * \retval              true if llog object exists and is not just destroyed
123  * \retval              false if llog doesn't exist or just destroyed
124  */
125 static int llog_osd_exist(struct llog_handle *handle)
126 {
127         LASSERT(handle->lgh_obj);
128         return dt_object_exists(handle->lgh_obj) && !handle->lgh_destroyed;
129 }
130
131 static void *rec_tail(struct llog_rec_hdr *rec)
132 {
133         return (void *)((char *)rec + rec->lrh_len -
134                         sizeof(struct llog_rec_tail));
135 }
136
137 /**
138  * Write a padding record to the llog
139  *
140  * This function writes a padding record to the end of llog. That may
141  * be needed if llog contains records of variable size, e.g. config logs
142  * or changelogs.
143  * The padding record just aligns llog to the llog chunk_size boundary if
144  * the current record doesn't fit in the remaining space.
145  *
146  * It allocates full length to avoid two separate writes for header and tail.
147  * Such 2-steps scheme needs extra protection and complex error handling.
148  *
149  * \param[in]     env   execution environment
150  * \param[in]     o     dt_object to create
151  * \param[in,out] off   pointer to the padding start offset
152  * \param[in]     len   padding length
153  * \param[in]     index index of the padding record in a llog
154  * \param[in]     th    current transaction handle
155  *
156  * \retval              0 on successful padding write
157  * \retval              negative error if write failed
158  */
159 static int llog_osd_pad(const struct lu_env *env, struct dt_object *o,
160                         loff_t *off, int len, int index, struct thandle *th)
161 {
162         struct llog_thread_info *lgi = llog_info(env);
163         struct llog_rec_hdr     *rec;
164         struct llog_rec_tail    *tail;
165         int                      rc;
166
167         ENTRY;
168
169         LASSERT(th);
170         LASSERT(off);
171         LASSERT(len >= LLOG_MIN_REC_SIZE && (len & 0x7) == 0);
172
173         OBD_ALLOC(rec, len);
174         if (rec == NULL)
175                 RETURN(-ENOMEM);
176
177         rec->lrh_len = len;
178         rec->lrh_index = index;
179         rec->lrh_type = LLOG_PAD_MAGIC;
180
181         tail = rec_tail(rec);
182         tail->lrt_len = len;
183         tail->lrt_index = index;
184
185         lgi->lgi_buf.lb_buf = rec;
186         lgi->lgi_buf.lb_len = len;
187         rc = dt_record_write(env, o, &lgi->lgi_buf, off, th);
188         if (rc)
189                 CERROR("%s: error writing padding record: rc = %d\n",
190                        o->do_lu.lo_dev->ld_obd->obd_name, rc);
191
192         OBD_FREE(rec, len);
193         RETURN(rc);
194 }
195
196 /**
197  * Implementation of the llog_operations::lop_read_header
198  *
199  * This function reads the current llog header from the bottom storage
200  * device.
201  *
202  * \param[in] env       execution environment
203  * \param[in] handle    llog handle of the current llog
204  *
205  * \retval              0 on successful header read
206  * \retval              negative error if read failed
207  */
208 static int llog_osd_read_header(const struct lu_env *env,
209                                 struct llog_handle *handle)
210 {
211         struct llog_rec_hdr     *llh_hdr;
212         struct dt_object        *o;
213         struct llog_thread_info *lgi;
214         enum llog_flag           flags;
215         int                      rc;
216
217         ENTRY;
218
219         o = handle->lgh_obj;
220         LASSERT(o);
221
222         lgi = llog_info(env);
223
224         dt_read_lock(env, o, 0);
225
226         rc = dt_attr_get(env, o, &lgi->lgi_attr);
227         if (rc)
228                 GOTO(unlock, rc);
229
230         LASSERT(lgi->lgi_attr.la_valid & LA_SIZE);
231
232         if (lgi->lgi_attr.la_size == 0) {
233                 CDEBUG(D_HA, "not reading header from 0-byte log\n");
234                 GOTO(unlock, rc = LLOG_EEMPTY);
235         }
236
237         flags = handle->lgh_hdr->llh_flags;
238
239         lgi->lgi_off = 0;
240         lgi->lgi_buf.lb_buf = handle->lgh_hdr;
241         lgi->lgi_buf.lb_len = handle->lgh_hdr_size;
242         rc = dt_read(env, o, &lgi->lgi_buf, &lgi->lgi_off);
243         llh_hdr = &handle->lgh_hdr->llh_hdr;
244         if (rc < sizeof(*llh_hdr) || rc < llh_hdr->lrh_len) {
245                 CERROR("%s: error reading "DFID" log header size %d: rc = %d\n",
246                        o->do_lu.lo_dev->ld_obd->obd_name,
247                        PFID(lu_object_fid(&o->do_lu)), rc < 0 ? 0 : rc,
248                        -EFAULT);
249
250                 if (rc >= 0)
251                         rc = -EFAULT;
252
253                 GOTO(unlock, rc);
254         }
255
256         if (LLOG_REC_HDR_NEEDS_SWABBING(llh_hdr))
257                 lustre_swab_llog_hdr(handle->lgh_hdr);
258
259         if (llh_hdr->lrh_type != LLOG_HDR_MAGIC) {
260                 CERROR("%s: bad log %s "DFID" header magic: %#x "
261                        "(expected %#x)\n", o->do_lu.lo_dev->ld_obd->obd_name,
262                        handle->lgh_name ? handle->lgh_name : "",
263                        PFID(lu_object_fid(&o->do_lu)),
264                        llh_hdr->lrh_type, LLOG_HDR_MAGIC);
265                 GOTO(unlock, rc = -EIO);
266         } else if (llh_hdr->lrh_len < LLOG_MIN_CHUNK_SIZE ||
267                    llh_hdr->lrh_len > handle->lgh_hdr_size) {
268                 CERROR("%s: incorrectly sized log %s "DFID" header: "
269                        "%#x (expected at least %#x)\n"
270                        "you may need to re-run lconf --write_conf.\n",
271                        o->do_lu.lo_dev->ld_obd->obd_name,
272                        handle->lgh_name ? handle->lgh_name : "",
273                        PFID(lu_object_fid(&o->do_lu)),
274                        llh_hdr->lrh_len, LLOG_MIN_CHUNK_SIZE);
275                 GOTO(unlock, rc = -EIO);
276         } else if (LLOG_HDR_TAIL(handle->lgh_hdr)->lrt_index >
277                    LLOG_HDR_BITMAP_SIZE(handle->lgh_hdr) ||
278                    LLOG_HDR_TAIL(handle->lgh_hdr)->lrt_len !=
279                         llh_hdr->lrh_len) {
280                 CERROR("%s: incorrectly sized log %s "DFID" tailer: "
281                        "%#x : rc = %d\n",
282                        o->do_lu.lo_dev->ld_obd->obd_name,
283                        handle->lgh_name ? handle->lgh_name : "",
284                        PFID(lu_object_fid(&o->do_lu)),
285                        LLOG_HDR_TAIL(handle->lgh_hdr)->lrt_len, -EIO);
286                 GOTO(unlock, rc = -EIO);
287         }
288
289         handle->lgh_hdr->llh_flags |= (flags & LLOG_F_EXT_MASK);
290         handle->lgh_last_idx = LLOG_HDR_TAIL(handle->lgh_hdr)->lrt_index;
291         rc = 0;
292
293 unlock:
294         dt_read_unlock(env, o);
295         RETURN(rc);
296 }
297
298 /**
299  * Implementation of the llog_operations::lop_declare_write
300  *
301  * This function declares the new record write.
302  *
303  * \param[in] env       execution environment
304  * \param[in] loghandle llog handle of the current llog
305  * \param[in] rec       llog record header. This is a real header of the full
306  *                      llog record to write. This is the beginning of buffer
307  *                      to write, the length of buffer is stored in
308  *                      \a rec::lrh_len
309  * \param[in] idx       index of the llog record. If \a idx == -1 then this is
310  *                      append case, otherwise \a idx is the index of record
311  *                      to modify
312  * \param[in] th        current transaction handle
313  *
314  * \retval              0 on successful declaration
315  * \retval              negative error if declaration failed
316  */
317 static int llog_osd_declare_write_rec(const struct lu_env *env,
318                                       struct llog_handle *loghandle,
319                                       struct llog_rec_hdr *rec,
320                                       int idx, struct thandle *th)
321 {
322         struct llog_thread_info *lgi = llog_info(env);
323         __u32                   chunk_size;
324         struct dt_object        *o;
325         int                      rc;
326
327         ENTRY;
328
329         LASSERT(env);
330         LASSERT(th);
331         LASSERT(loghandle);
332         LASSERT(rec);
333         LASSERT(rec->lrh_len <= loghandle->lgh_ctxt->loc_chunk_size);
334
335         o = loghandle->lgh_obj;
336         LASSERT(o);
337
338         chunk_size = loghandle->lgh_ctxt->loc_chunk_size;
339         lgi->lgi_buf.lb_len = chunk_size;
340         lgi->lgi_buf.lb_buf = NULL;
341         /* each time we update header */
342         rc = dt_declare_record_write(env, o, &lgi->lgi_buf, 0,
343                                      th);
344         if (rc || idx == 0) /* if error or just header */
345                 RETURN(rc);
346
347         /**
348          * the pad record can be inserted so take into account double
349          * record size
350          */
351         lgi->lgi_buf.lb_len = chunk_size * 2;
352         lgi->lgi_buf.lb_buf = NULL;
353         /* XXX: implement declared window or multi-chunks approach */
354         rc = dt_declare_record_write(env, o, &lgi->lgi_buf, -1, th);
355
356         RETURN(rc);
357 }
358
359 /**
360  * Implementation of the llog_operations::lop_write
361  *
362  * This function writes the new record in the llog or modify the existed one.
363  *
364  * \param[in]  env              execution environment
365  * \param[in]  loghandle        llog handle of the current llog
366  * \param[in]  rec              llog record header. This is a real header of
367  *                              the full llog record to write. This is
368  *                              the beginning of buffer to write, the length
369  *                              of buffer is stored in \a rec::lrh_len
370  * \param[in,out] reccookie     pointer to the cookie to return back if needed.
371  *                              It is used for further cancel of this llog
372  *                              record.
373  * \param[in]  idx              index of the llog record. If \a idx == -1 then
374  *                              this is append case, otherwise \a idx is
375  *                              the index of record to modify
376  * \param[in]  th               current transaction handle
377  *
378  * \retval                      0 on successful write && \a reccookie == NULL
379  *                              1 on successful write && \a reccookie != NULL
380  * \retval                      negative error if write failed
381  */
382 static int llog_osd_write_rec(const struct lu_env *env,
383                               struct llog_handle *loghandle,
384                               struct llog_rec_hdr *rec,
385                               struct llog_cookie *reccookie,
386                               int idx, struct thandle *th)
387 {
388         struct llog_thread_info *lgi = llog_info(env);
389         struct llog_log_hdr *llh;
390         int reclen = rec->lrh_len;
391         int index, rc;
392         struct llog_rec_tail *lrt;
393         struct dt_object *o;
394         __u32 chunk_size;
395         size_t left;
396         __u32 orig_last_idx;
397         bool pad = false;
398         ENTRY;
399
400         llh = loghandle->lgh_hdr;
401         o = loghandle->lgh_obj;
402
403         chunk_size = llh->llh_hdr.lrh_len;
404         CDEBUG(D_OTHER, "new record %x to "DFID"\n",
405                rec->lrh_type, PFID(lu_object_fid(&o->do_lu)));
406
407         if (!llog_osd_exist(loghandle))
408                 RETURN(-ENOENT);
409
410         /* record length should not bigger than  */
411         if (reclen > loghandle->lgh_hdr->llh_hdr.lrh_len)
412                 RETURN(-E2BIG);
413
414         /* sanity check for fixed-records llog */
415         if (idx != LLOG_HEADER_IDX && (llh->llh_flags & LLOG_F_IS_FIXSIZE)) {
416                 LASSERT(llh->llh_size != 0);
417                 LASSERT(llh->llh_size == reclen);
418         }
419
420         /* return error if osp object is stale */
421         if (idx != LLOG_HEADER_IDX && dt_object_stale(o))
422                 RETURN(-ESTALE);
423         rc = dt_attr_get(env, o, &lgi->lgi_attr);
424         if (rc)
425                 RETURN(rc);
426
427         /**
428          * The modification case.
429          * If idx set then the record with that index must be modified.
430          * There are three cases possible:
431          * 1) the common case is the llog header update (idx == 0)
432          * 2) the llog record modification during llog process.
433          *    This is indicated by the \a loghandle::lgh_cur_idx > 0.
434          *    In that case the \a loghandle::lgh_cur_offset
435          * 3) otherwise this is assumed that llog consist of records of
436          *    fixed size, i.e. catalog. The llog header must has llh_size
437          *    field equal to record size. The record offset is calculated
438          *    just by /a idx value
439          *
440          * During modification we don't need extra header update because
441          * the bitmap and record count are not changed. The record header
442          * and tail remains the same too.
443          */
444         if (idx != LLOG_NEXT_IDX) {
445                 /* llog can be empty only when first record is being written */
446                 LASSERT(ergo(idx > 0, lgi->lgi_attr.la_size > 0));
447
448                 if (!test_bit_le(idx, LLOG_HDR_BITMAP(llh))) {
449                         CERROR("%s: modify unset record %u\n",
450                                o->do_lu.lo_dev->ld_obd->obd_name, idx);
451                         RETURN(-ENOENT);
452                 }
453
454                 if (idx != rec->lrh_index) {
455                         CERROR("%s: modify index mismatch %d %u\n",
456                                o->do_lu.lo_dev->ld_obd->obd_name, idx,
457                                rec->lrh_index);
458                         RETURN(-EFAULT);
459                 }
460
461                 if (idx == LLOG_HEADER_IDX) {
462                         /* llog header update */
463                         __u32   *bitmap = LLOG_HDR_BITMAP(llh);
464
465                         lgi->lgi_off = 0;
466
467                         /* If it does not indicate the bitmap index
468                          * (reccookie == NULL), then it means update
469                          * the whole update header. Otherwise only
470                          * update header and bits needs to be updated,
471                          * and in DNE cases, it will signaficantly
472                          * shrink the RPC size.
473                          * see distribute_txn_cancel_records()*/
474                         if (reccookie == NULL) {
475                                 lgi->lgi_buf.lb_len = reclen;
476                                 lgi->lgi_buf.lb_buf = rec;
477                                 rc = dt_record_write(env, o, &lgi->lgi_buf,
478                                                      &lgi->lgi_off, th);
479                                 RETURN(rc);
480                         }
481
482                         /* update the header */
483                         lgi->lgi_buf.lb_len = llh->llh_bitmap_offset;
484                         lgi->lgi_buf.lb_buf = llh;
485                         rc = dt_record_write(env, o, &lgi->lgi_buf,
486                                              &lgi->lgi_off, th);
487                         if (rc != 0)
488                                 RETURN(rc);
489
490                         /* update the bitmap */
491                         index = reccookie->lgc_index;
492                         lgi->lgi_off = llh->llh_bitmap_offset +
493                                       (index / (sizeof(*bitmap) * 8)) *
494                                                         sizeof(*bitmap);
495                         lgi->lgi_buf.lb_len = sizeof(*bitmap);
496                         lgi->lgi_buf.lb_buf =
497                                         &bitmap[index/(sizeof(*bitmap)*8)];
498                         rc = dt_record_write(env, o, &lgi->lgi_buf,
499                                              &lgi->lgi_off, th);
500
501                         RETURN(rc);
502                 } else if (llh->llh_flags & LLOG_F_IS_FIXSIZE) {
503                         lgi->lgi_off = llh->llh_hdr.lrh_len +
504                                        (idx - 1) * reclen;
505                 } else if (reccookie != NULL && reccookie->lgc_index > 0) {
506                         /**
507                          * The lgc_offset can be used only if index is
508                          * the same.
509                          */
510                         if (idx != reccookie->lgc_index) {
511                                 CERROR("%s: modify index mismatch %d %d\n",
512                                        o->do_lu.lo_dev->ld_obd->obd_name, idx,
513                                        reccookie->lgc_index);
514                                 RETURN(-EFAULT);
515                         }
516
517                         lgi->lgi_off = reccookie->lgc_offset;
518                         CDEBUG(D_OTHER, "modify record "DFID": idx:%u, "
519                                "len:%u offset %llu\n",
520                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid), idx,
521                                rec->lrh_len, (long long)lgi->lgi_off);
522                 } else {
523                         /* This can be result of lgh_cur_idx is not set during
524                          * llog processing or llh_size is not set to proper
525                          * record size for fixed records llog. Therefore it is
526                          * impossible to get record offset. */
527                         CERROR("%s: can't get record offset, idx:%d, "
528                                "len:%u.\n", o->do_lu.lo_dev->ld_obd->obd_name,
529                                idx, rec->lrh_len);
530                         RETURN(-EFAULT);
531                 }
532
533                 /* update only data, header and tail remain the same */
534                 lgi->lgi_off += sizeof(struct llog_rec_hdr);
535                 lgi->lgi_buf.lb_len = REC_DATA_LEN(rec);
536                 lgi->lgi_buf.lb_buf = REC_DATA(rec);
537                 rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
538                 if (rc == 0 && reccookie) {
539                         reccookie->lgc_lgl = loghandle->lgh_id;
540                         reccookie->lgc_index = idx;
541                         rc = 1;
542                 }
543                 RETURN(rc);
544         }
545
546         /**
547          * The append case.
548          * The most common case of using llog. The new index is assigned to
549          * the new record, new bit is set in llog bitmap and llog count is
550          * incremented.
551          *
552          * Make sure that records don't cross a chunk boundary, so we can
553          * process them page-at-a-time if needed.  If it will cross a chunk
554          * boundary, write in a fake (but referenced) entry to pad the chunk.
555          */
556
557
558         /* simulate ENOSPC when new plain llog is being added to the
559          * catalog */
560         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_LLOG_CREATE_FAILED2) &&
561             llh->llh_flags & LLOG_F_IS_CAT)
562                 RETURN(-ENOSPC);
563
564         LASSERT(lgi->lgi_attr.la_valid & LA_SIZE);
565         orig_last_idx = loghandle->lgh_last_idx;
566         lgi->lgi_off = lgi->lgi_attr.la_size;
567
568         if (loghandle->lgh_max_size > 0 &&
569             lgi->lgi_off >= loghandle->lgh_max_size) {
570                 CDEBUG(D_OTHER, "llog is getting too large (%u > %u) at %u "
571                        DFID"\n", (unsigned)lgi->lgi_off,
572                        loghandle->lgh_max_size, (int)loghandle->lgh_last_idx,
573                        PFID(&loghandle->lgh_id.lgl_oi.oi_fid));
574                 /* this is to signal that this llog is full */
575                 loghandle->lgh_last_idx = LLOG_HDR_BITMAP_SIZE(llh) - 1;
576                 RETURN(-ENOSPC);
577         }
578
579         left = chunk_size - (lgi->lgi_off & (chunk_size - 1));
580         /* NOTE: padding is a record, but no bit is set */
581         if (left != 0 && left != reclen &&
582             left < (reclen + LLOG_MIN_REC_SIZE)) {
583                 index = loghandle->lgh_last_idx + 1;
584                 rc = llog_osd_pad(env, o, &lgi->lgi_off, left, index, th);
585                 if (rc)
586                         RETURN(rc);
587
588                 loghandle->lgh_last_idx++; /* for pad rec */
589                 pad = true;
590         }
591         /* if it's the last idx in log file, then return -ENOSPC
592          * or wrap around if a catalog */
593         if (llog_is_full(loghandle) ||
594             unlikely(llh->llh_flags & LLOG_F_IS_CAT &&
595                      OBD_FAIL_PRECHECK(OBD_FAIL_CAT_RECORDS) &&
596                      loghandle->lgh_last_idx >= cfs_fail_val)) {
597                 if (llh->llh_flags & LLOG_F_IS_CAT)
598                         loghandle->lgh_last_idx = 0;
599                 else
600                         RETURN(-ENOSPC);
601         }
602
603         down_write(&loghandle->lgh_last_sem);
604         /* increment the last_idx along with llh_tail index, they should
605          * be equal for a llog lifetime */
606         loghandle->lgh_last_idx++;
607         index = loghandle->lgh_last_idx;
608         LLOG_HDR_TAIL(llh)->lrt_index = index;
609         /**
610          * NB: the caller should make sure only 1 process access
611          * the lgh_last_idx, e.g. append should be exclusive.
612          * Otherwise it might hit the assert.
613          */
614         LASSERT(index < LLOG_HDR_BITMAP_SIZE(llh));
615         rec->lrh_index = index;
616         lrt = rec_tail(rec);
617         lrt->lrt_len = rec->lrh_len;
618         lrt->lrt_index = rec->lrh_index;
619
620         /* the lgh_hdr_mutex protects llog header data from concurrent
621          * update/cancel, the llh_count and llh_bitmap are protected */
622         mutex_lock(&loghandle->lgh_hdr_mutex);
623         if (__test_and_set_bit_le(index, LLOG_HDR_BITMAP(llh))) {
624                 CERROR("%s: index %u already set in llog bitmap "DFID"\n",
625                        o->do_lu.lo_dev->ld_obd->obd_name, index,
626                        PFID(lu_object_fid(&o->do_lu)));
627                 mutex_unlock(&loghandle->lgh_hdr_mutex);
628                 LBUG(); /* should never happen */
629         }
630         llh->llh_count++;
631
632         if (!(llh->llh_flags & LLOG_F_IS_FIXSIZE)) {
633                 /* Update the minimum size of the llog record */
634                 if (llh->llh_size == 0)
635                         llh->llh_size = reclen;
636                 else if (reclen < llh->llh_size)
637                         llh->llh_size = reclen;
638         }
639
640         /*
641          * readers (e.g. llog_osd_read_header()) must not find
642          * llog updated partially (bitmap/counter claims record,
643          * but a record hasn't been added yet) as this results
644          * in EIO.
645          */
646         dt_write_lock(env, o, 0);
647
648         if (lgi->lgi_attr.la_size == 0) {
649                 lgi->lgi_off = 0;
650                 lgi->lgi_buf.lb_len = llh->llh_hdr.lrh_len;
651                 lgi->lgi_buf.lb_buf = &llh->llh_hdr;
652                 rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
653                 if (rc != 0)
654                         GOTO(out_unlock, rc);
655         } else {
656                 __u32   *bitmap = LLOG_HDR_BITMAP(llh);
657
658                 /* Note: If this is not initialization (size == 0), then do not
659                  * write the whole header (8k bytes), only update header/tail
660                  * and bits needs to be updated. Because this update might be
661                  * part of cross-MDT operation, which needs to write these
662                  * updates into the update log(32KB limit) and also pack inside
663                  * the RPC (1MB limit), if we write 8K for each operation, which
664                  * will cost a lot space, and keep us adding more updates to one
665                  * update log.*/
666                 lgi->lgi_off = 0;
667                 lgi->lgi_buf.lb_len = llh->llh_bitmap_offset;
668                 lgi->lgi_buf.lb_buf = &llh->llh_hdr;
669                 rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
670                 if (rc != 0)
671                         GOTO(out_unlock, rc);
672
673                 lgi->lgi_off = llh->llh_bitmap_offset +
674                               (index / (sizeof(*bitmap) * 8)) * sizeof(*bitmap);
675                 lgi->lgi_buf.lb_len = sizeof(*bitmap);
676                 lgi->lgi_buf.lb_buf = &bitmap[index/(sizeof(*bitmap)*8)];
677                 rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
678                 if (rc != 0)
679                         GOTO(out_unlock, rc);
680
681                 lgi->lgi_off =  (unsigned long)LLOG_HDR_TAIL(llh) -
682                                 (unsigned long)llh;
683                 lgi->lgi_buf.lb_len = sizeof(llh->llh_tail);
684                 lgi->lgi_buf.lb_buf = LLOG_HDR_TAIL(llh);
685                 rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
686                 if (rc != 0)
687                         GOTO(out_unlock, rc);
688         }
689         if (OBD_FAIL_PRECHECK(OBD_FAIL_LLOG_PAUSE_AFTER_PAD) && pad) {
690                 /* a window for concurrent llog reader, see LU-12577 */
691                 OBD_FAIL_TIMEOUT(OBD_FAIL_LLOG_PAUSE_AFTER_PAD,
692                                  cfs_fail_val ?: 1);
693         }
694
695 out_unlock:
696         /* unlock here for remote object */
697         mutex_unlock(&loghandle->lgh_hdr_mutex);
698         if (rc) {
699                 dt_write_unlock(env, o);
700                 GOTO(out, rc);
701         }
702
703         if (OBD_FAIL_PRECHECK(OBD_FAIL_LLOG_PROCESS_TIMEOUT) &&
704            cfs_fail_val == (unsigned int)(loghandle->lgh_id.lgl_oi.oi.oi_id &
705                                           0xFFFFFFFF)) {
706                 OBD_RACE(OBD_FAIL_LLOG_PROCESS_TIMEOUT);
707                 msleep(1 * MSEC_PER_SEC);
708         }
709         /* computed index can be used to determine offset for fixed-size
710          * records. This also allows to handle Catalog wrap around case */
711         if (llh->llh_flags & LLOG_F_IS_FIXSIZE) {
712                 lgi->lgi_off = llh->llh_hdr.lrh_len + (index - 1) * reclen;
713         } else {
714                 rc = dt_attr_get(env, o, &lgi->lgi_attr);
715                 if (rc) {
716                         dt_write_unlock(env, o);
717                         GOTO(out, rc);
718                 }
719
720                 LASSERT(lgi->lgi_attr.la_valid & LA_SIZE);
721                 lgi->lgi_off = max_t(__u64, lgi->lgi_attr.la_size,
722                                      lgi->lgi_off);
723         }
724
725         lgi->lgi_buf.lb_len = reclen;
726         lgi->lgi_buf.lb_buf = rec;
727         rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
728
729         dt_write_unlock(env, o);
730         if (rc < 0)
731                 GOTO(out, rc);
732
733         up_write(&loghandle->lgh_last_sem);
734
735         CDEBUG(D_HA, "added record "DFID".%u, %u off%llu\n",
736                PFID(lu_object_fid(&o->do_lu)), index, rec->lrh_len,
737                lgi->lgi_off);
738         if (reccookie != NULL) {
739                 reccookie->lgc_lgl = loghandle->lgh_id;
740                 reccookie->lgc_index = index;
741                 if ((rec->lrh_type == MDS_UNLINK_REC) ||
742                     (rec->lrh_type == MDS_SETATTR64_REC))
743                         reccookie->lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
744                 else if (rec->lrh_type == OST_SZ_REC)
745                         reccookie->lgc_subsys = LLOG_SIZE_ORIG_CTXT;
746                 else
747                         reccookie->lgc_subsys = -1;
748                 rc = 1;
749         }
750         RETURN(rc);
751 out:
752         /* cleanup llog for error case */
753         mutex_lock(&loghandle->lgh_hdr_mutex);
754         clear_bit_le(index, LLOG_HDR_BITMAP(llh));
755         llh->llh_count--;
756         mutex_unlock(&loghandle->lgh_hdr_mutex);
757
758         /* restore llog last_idx */
759         if (dt_object_remote(o)) {
760                 loghandle->lgh_last_idx = orig_last_idx;
761         } else if (--loghandle->lgh_last_idx == 0 &&
762             (llh->llh_flags & LLOG_F_IS_CAT) && llh->llh_cat_idx != 0) {
763                 /* catalog had just wrap-around case */
764                 loghandle->lgh_last_idx = LLOG_HDR_BITMAP_SIZE(llh) - 1;
765         }
766
767         LLOG_HDR_TAIL(llh)->lrt_index = loghandle->lgh_last_idx;
768         up_write(&loghandle->lgh_last_sem);
769
770         RETURN(rc);
771 }
772
773 /**
774  * We can skip reading at least as many log blocks as the number of
775  * minimum sized log records we are skipping.  If it turns out
776  * that we are not far enough along the log (because the
777  * actual records are larger than minimum size) we just skip
778  * some more records.
779  *
780  * Note: in llog_process_thread, it will use bitmap offset as
781  * the index to locate the record, which also includs some pad
782  * records, whose record size is very small, and it also does not
783  * consider pad record when recording minimum record size (otherwise
784  * min_record size might be too small), so in some rare cases,
785  * it might skip too much record for @goal, see llog_osd_next_block().
786  *
787  * When force_mini_rec is true, it means we have to use LLOG_MIN_REC_SIZE
788  * as the min record size to skip over, usually because in the previous
789  * try, it skip too much record, see loog_osd_next(prev)_block().
790  */
791 static inline void llog_skip_over(struct llog_handle *lgh, __u64 *off,
792                                   int curr, int goal, __u32 chunk_size,
793                                   bool force_mini_rec)
794 {
795         struct llog_log_hdr *llh = lgh->lgh_hdr;
796
797         /* Goal should not bigger than the record count */
798         if (goal > lgh->lgh_last_idx)
799                 goal = lgh->lgh_last_idx;
800
801         if (goal > curr) {
802                 if (llh->llh_flags & LLOG_F_IS_FIXSIZE) {
803                         *off = chunk_size + (goal - 1) * llh->llh_size;
804                 } else {
805                         __u64 min_rec_size = LLOG_MIN_REC_SIZE;
806
807                         if (llh->llh_size > 0 && !force_mini_rec)
808                                 min_rec_size = llh->llh_size;
809
810                         *off = *off + (goal - curr - 1) * min_rec_size;
811                 }
812         }
813         /* always align with lower chunk boundary*/
814         *off &= ~(chunk_size - 1);
815 }
816
817 /**
818  * Remove optional fields that the client doesn't expect.
819  * This is typically in order to ensure compatibility with older clients.
820  * It is assumed that since we exclusively remove fields, the block will be
821  * big enough to handle the remapped records. It is also assumed that records
822  * of a block have the same format (i.e.: the same features enabled).
823  *
824  * \param[in,out]    hdr           Header of the block of records to remap.
825  * \param[in,out]    last_hdr      Last header, don't read past this point.
826  * \param[in]        flags         Flags describing the fields to keep.
827  * \param[in]        extra_flags   Flags describing the extra fields to keep.
828  */
829 static void changelog_block_trim_ext(struct llog_rec_hdr *hdr,
830                                      struct llog_rec_hdr *last_hdr,
831                                      struct llog_handle *loghandle)
832 {
833         enum changelog_rec_flags flags = CLF_SUPPORTED;
834         enum changelog_rec_extra_flags extra_flags = CLFE_SUPPORTED;
835
836         if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_XATTR))
837                 extra_flags &= ~CLFE_XATTR;
838         if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_OMODE))
839                 extra_flags &= ~CLFE_OPEN;
840         if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_NID))
841                 extra_flags &= ~CLFE_NID;
842         if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_X_UIDGID))
843                 extra_flags &= ~CLFE_UIDGID;
844         if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_EXTRA_FLAGS))
845                 flags &= ~CLF_EXTRA_FLAGS;
846         if (!(loghandle->lgh_hdr->llh_flags & LLOG_F_EXT_JOBID))
847                 flags &= ~CLF_JOBID;
848
849         if (flags == CLF_SUPPORTED && extra_flags == CLFE_SUPPORTED)
850                 return;
851
852         if (hdr->lrh_type != CHANGELOG_REC)
853                 return;
854
855         do {
856                 struct changelog_rec *rec = (struct changelog_rec *)(hdr + 1);
857                 enum changelog_rec_extra_flags xflag = CLFE_INVALID;
858
859                 if (flags & CLF_EXTRA_FLAGS &&
860                     rec->cr_flags & CLF_EXTRA_FLAGS) {
861                         xflag = changelog_rec_extra_flags(rec)->cr_extra_flags &
862                                 extra_flags;
863                 }
864
865                 if (unlikely(hdr->lrh_len == 0)) {
866                         /* It is corruption case, we cannot know the next rec,
867                          * jump to the last one directly to avoid dead loop. */
868                         LCONSOLE(D_WARNING, "Hit invalid llog record: "
869                                  "idx %u, type %u, id %u\n",
870                                  hdr->lrh_index, hdr->lrh_type, hdr->lrh_id);
871                         hdr = llog_rec_hdr_next(last_hdr);
872                         if (unlikely(hdr == last_hdr))
873                                 LCONSOLE(D_WARNING, "The last record crashed: "
874                                          "idx %u, type %u, id %u\n",
875                                          hdr->lrh_index, hdr->lrh_type,
876                                          hdr->lrh_id);
877                         break;
878                 }
879
880                 changelog_remap_rec(rec, rec->cr_flags & flags, xflag);
881                 hdr = llog_rec_hdr_next(hdr);
882                 /* Yield CPU to avoid soft-lockup if there are too many records
883                  * to be handled. */
884                 cond_resched();
885         } while ((char *)hdr <= (char *)last_hdr);
886 }
887
888 /**
889  * Implementation of the llog_operations::lop_next_block
890  *
891  * This function finds the the next llog block to return which contains
892  * record with required index. It is main part of llog processing.
893  *
894  * \param[in]     env           execution environment
895  * \param[in]     loghandle     llog handle of the current llog
896  * \param[in,out] cur_idx       index preceeding cur_offset
897  * \param[in]     next_idx      target index to find
898  * \param[in,out] cur_offset    furtherst point read in the file
899  * \param[in]     buf           pointer to data buffer to fill
900  * \param[in]     len           required len to read, it is
901  *                              usually llog chunk_size.
902  *
903  * \retval                      0 on successful buffer read
904  * \retval                      negative value on error
905  */
906 static int llog_osd_next_block(const struct lu_env *env,
907                                struct llog_handle *loghandle, int *cur_idx,
908                                int next_idx, __u64 *cur_offset, void *buf,
909                                int len)
910 {
911         struct llog_thread_info *lgi = llog_info(env);
912         struct dt_object        *o;
913         struct dt_device        *dt;
914         int                      rc;
915         __u32                   chunk_size;
916         int last_idx = *cur_idx;
917         __u64 last_offset = *cur_offset;
918         bool force_mini_rec = false;
919
920         ENTRY;
921
922         LASSERT(env);
923         LASSERT(lgi);
924
925         chunk_size = loghandle->lgh_hdr->llh_hdr.lrh_len;
926         if (len == 0 || len & (chunk_size - 1))
927                 RETURN(-EINVAL);
928
929         LASSERT(loghandle);
930         LASSERT(loghandle->lgh_ctxt);
931
932         if (OBD_FAIL_PRECHECK(OBD_FAIL_MDS_CHANGELOG_DEL) &&
933             cfs_fail_val == ((unsigned long)loghandle & 0xFFFFFFFF)) {
934                 OBD_RACE(OBD_FAIL_MDS_CHANGELOG_DEL);
935                 msleep(MSEC_PER_SEC >> 2);
936         }
937
938         o = loghandle->lgh_obj;
939         LASSERT(o);
940         dt_read_lock(env, o, 0);
941         if (!llog_osd_exist(loghandle))
942                 GOTO(out, rc = -ESTALE); //object was destroyed
943
944         dt = lu2dt_dev(o->do_lu.lo_dev);
945         LASSERT(dt);
946
947         rc = dt_attr_get(env, o, &lgi->lgi_attr);
948         if (rc)
949                 GOTO(out, rc);
950
951         CDEBUG(D_OTHER,
952                "looking for log index %u (cur idx %u off %llu), size %llu\n",
953                next_idx, *cur_idx,
954                *cur_offset, lgi->lgi_attr.la_size);
955
956         while (*cur_offset < lgi->lgi_attr.la_size) {
957                 struct llog_rec_hdr     *rec, *last_rec;
958                 struct llog_rec_tail    *tail;
959
960                 llog_skip_over(loghandle, cur_offset, *cur_idx,
961                                next_idx, chunk_size, force_mini_rec);
962
963                 /* read up to next llog chunk_size block */
964                 lgi->lgi_buf.lb_len = chunk_size -
965                                       (*cur_offset & (chunk_size - 1));
966                 lgi->lgi_buf.lb_buf = buf;
967
968                 rc = dt_read(env, o, &lgi->lgi_buf, cur_offset);
969                 if (rc < 0) {
970                         if (rc == -EBADR && !force_mini_rec)
971                                 goto retry;
972
973                         CERROR("%s: can't read llog block from log "DFID
974                                " offset %llu: rc = %d\n",
975                                o->do_lu.lo_dev->ld_obd->obd_name,
976                                PFID(lu_object_fid(&o->do_lu)), *cur_offset,
977                                rc);
978                         GOTO(out, rc);
979                 }
980
981                 if (rc < len) {
982                         /* signal the end of the valid buffer to
983                          * llog_process */
984                         memset(buf + rc, 0, len - rc);
985                 }
986
987                 if (rc == 0) { /* end of file, nothing to do */
988                         if (!force_mini_rec)
989                                 goto retry;
990                         GOTO(out, rc);
991                 }
992
993                 if (rc < sizeof(*tail)) {
994                         if (!force_mini_rec)
995                                 goto retry;
996
997                         CERROR("%s: invalid llog block at log id "DFID":%x "
998                                "offset %llu\n",
999                                o->do_lu.lo_dev->ld_obd->obd_name,
1000                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
1001                                loghandle->lgh_id.lgl_ogen, *cur_offset);
1002                         GOTO(out, rc = -EINVAL);
1003                 }
1004
1005                 rec = buf;
1006                 if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
1007                         lustre_swab_llog_rec(rec);
1008                 tail = (struct llog_rec_tail *)((char *)buf + rc -
1009                                                 sizeof(struct llog_rec_tail));
1010
1011                 if (llog_verify_record(loghandle, rec)) {
1012                         /*
1013                          * the block seems corrupted. make a pad record so the
1014                          * caller can skip the block and try with the next one
1015                          */
1016                         rec->lrh_len = rc;
1017                         rec->lrh_index = next_idx;
1018                         rec->lrh_type = LLOG_PAD_MAGIC;
1019
1020                         tail = rec_tail(rec);
1021                         tail->lrt_len = rc;
1022                         tail->lrt_index = next_idx;
1023
1024                         GOTO(out, rc = 0);
1025                 }
1026
1027                 /* get the last record in block */
1028                 last_rec = (struct llog_rec_hdr *)((char *)buf + rc -
1029                                                    tail->lrt_len);
1030
1031                 if (LLOG_REC_HDR_NEEDS_SWABBING(last_rec))
1032                         lustre_swab_llog_rec(last_rec);
1033
1034                 if (last_rec->lrh_index != tail->lrt_index) {
1035                         CERROR("%s: invalid llog tail at log id "DFID":%x offset %llu last_rec idx %u tail idx %u lrt len %u read_size %d\n",
1036                                o->do_lu.lo_dev->ld_obd->obd_name,
1037                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
1038                                loghandle->lgh_id.lgl_ogen, *cur_offset,
1039                                last_rec->lrh_index, tail->lrt_index,
1040                                tail->lrt_len, rc);
1041                         GOTO(out, rc = -EINVAL);
1042                 }
1043
1044                 *cur_idx = tail->lrt_index;
1045
1046                 /* this shouldn't happen */
1047                 if (tail->lrt_index == 0) {
1048                         CERROR("%s: invalid llog tail at log id "DFID":%x "
1049                                "offset %llu bytes %d\n",
1050                                o->do_lu.lo_dev->ld_obd->obd_name,
1051                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
1052                                loghandle->lgh_id.lgl_ogen, *cur_offset, rc);
1053                         GOTO(out, rc = -EINVAL);
1054                 }
1055                 if (tail->lrt_index < next_idx) {
1056                         last_idx = *cur_idx;
1057                         last_offset = *cur_offset;
1058                         continue;
1059                 }
1060
1061                 /* sanity check that the start of the new buffer is no farther
1062                  * than the record that we wanted.  This shouldn't happen. */
1063                 if (next_idx && rec->lrh_index > next_idx) {
1064                         if (!force_mini_rec && next_idx > last_idx)
1065                                 goto retry;
1066
1067                         CERROR("%s: missed desired record? %u > %u\n",
1068                                o->do_lu.lo_dev->ld_obd->obd_name,
1069                                rec->lrh_index, next_idx);
1070                         GOTO(out, rc = -ENOENT);
1071                 }
1072
1073                 /* Trim unsupported extensions for compat w/ older clients */
1074                 changelog_block_trim_ext(rec, last_rec, loghandle);
1075
1076                 GOTO(out, rc = 0);
1077
1078 retry:
1079                 /* Note: because there are some pad records in the
1080                  * llog, so llog_skip_over() might skip too much
1081                  * records, let's try skip again with minimum record */
1082                 force_mini_rec = true;
1083                 *cur_offset = last_offset;
1084                 *cur_idx = last_idx;
1085         }
1086         GOTO(out, rc = -EIO);
1087 out:
1088         dt_read_unlock(env, o);
1089         return rc;
1090 }
1091
1092 /**
1093  * Implementation of the llog_operations::lop_prev_block
1094  *
1095  * This function finds the llog block to return which contains
1096  * record with required index but in reverse order - from end of llog
1097  * to the beginning.
1098  * It is main part of reverse llog processing.
1099  *
1100  * \param[in] env       execution environment
1101  * \param[in] loghandle llog handle of the current llog
1102  * \param[in] prev_idx  target index to find
1103  * \param[in] buf       pointer to data buffer to fill
1104  * \param[in] len       required len to read, it is llog_chunk_size usually.
1105  *
1106  * \retval              0 on successful buffer read
1107  * \retval              negative value on error
1108  */
1109 static int llog_osd_prev_block(const struct lu_env *env,
1110                                struct llog_handle *loghandle,
1111                                int prev_idx, void *buf, int len)
1112 {
1113         struct llog_thread_info *lgi = llog_info(env);
1114         struct dt_object        *o;
1115         struct dt_device        *dt;
1116         loff_t                   cur_offset;
1117         __u32                   chunk_size;
1118         int                      rc;
1119
1120         ENTRY;
1121
1122         chunk_size = loghandle->lgh_hdr->llh_hdr.lrh_len;
1123         if (len == 0 || len & (chunk_size - 1))
1124                 RETURN(-EINVAL);
1125
1126         CDEBUG(D_OTHER, "looking for log index %u\n", prev_idx);
1127
1128         LASSERT(loghandle);
1129         LASSERT(loghandle->lgh_ctxt);
1130
1131         o = loghandle->lgh_obj;
1132         LASSERT(o);
1133         dt_read_lock(env, o, 0);
1134         if (!llog_osd_exist(loghandle))
1135                 GOTO(out, rc = -ESTALE);
1136
1137         dt = lu2dt_dev(o->do_lu.lo_dev);
1138         LASSERT(dt);
1139
1140         /* Let's only use mini record size for previous block read
1141          * for now XXX */
1142         cur_offset = chunk_size;
1143         llog_skip_over(loghandle, &cur_offset, 0, prev_idx,
1144                        chunk_size, true);
1145
1146         rc = dt_attr_get(env, o, &lgi->lgi_attr);
1147         if (rc)
1148                 GOTO(out, rc);
1149
1150         while (cur_offset < lgi->lgi_attr.la_size) {
1151                 struct llog_rec_hdr     *rec, *last_rec;
1152                 struct llog_rec_tail    *tail;
1153
1154                 lgi->lgi_buf.lb_len = len;
1155                 lgi->lgi_buf.lb_buf = buf;
1156                 rc = dt_read(env, o, &lgi->lgi_buf, &cur_offset);
1157                 if (rc < 0) {
1158                         CERROR("%s: can't read llog block from log "DFID
1159                                " offset %llu: rc = %d\n",
1160                                o->do_lu.lo_dev->ld_obd->obd_name,
1161                                PFID(lu_object_fid(&o->do_lu)), cur_offset, rc);
1162                         GOTO(out, rc);
1163                 }
1164
1165                 if (rc == 0) /* end of file, nothing to do */
1166                         GOTO(out, rc);
1167
1168                 if (rc < sizeof(*tail)) {
1169                         CERROR("%s: invalid llog block at log id "DFID":%x "
1170                                "offset %llu\n",
1171                                o->do_lu.lo_dev->ld_obd->obd_name,
1172                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
1173                                loghandle->lgh_id.lgl_ogen, cur_offset);
1174                         GOTO(out, rc = -EINVAL);
1175                 }
1176
1177                 rec = buf;
1178                 if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
1179                         lustre_swab_llog_rec(rec);
1180
1181                 tail = (struct llog_rec_tail *)((char *)buf + rc -
1182                                                 sizeof(struct llog_rec_tail));
1183                 /* get the last record in block */
1184                 last_rec = (struct llog_rec_hdr *)((char *)buf + rc -
1185                                                    le32_to_cpu(tail->lrt_len));
1186
1187                 if (LLOG_REC_HDR_NEEDS_SWABBING(last_rec))
1188                         lustre_swab_llog_rec(last_rec);
1189                 LASSERT(last_rec->lrh_index == tail->lrt_index);
1190
1191                 /* this shouldn't happen */
1192                 if (tail->lrt_index == 0) {
1193                         CERROR("%s: invalid llog tail at log id "DFID":%x "
1194                                "offset %llu\n",
1195                                o->do_lu.lo_dev->ld_obd->obd_name,
1196                                PFID(&loghandle->lgh_id.lgl_oi.oi_fid),
1197                                loghandle->lgh_id.lgl_ogen, cur_offset);
1198                         GOTO(out, rc = -EINVAL);
1199                 }
1200                 if (tail->lrt_index < prev_idx)
1201                         continue;
1202
1203                 /* sanity check that the start of the new buffer is no farther
1204                  * than the record that we wanted.  This shouldn't happen. */
1205                 if (rec->lrh_index > prev_idx) {
1206                         CERROR("%s: missed desired record? %u > %u\n",
1207                                o->do_lu.lo_dev->ld_obd->obd_name,
1208                                rec->lrh_index, prev_idx);
1209                         GOTO(out, rc = -ENOENT);
1210                 }
1211
1212                 /* Trim unsupported extensions for compat w/ older clients */
1213                 changelog_block_trim_ext(rec, last_rec, loghandle);
1214
1215                 GOTO(out, rc = 0);
1216         }
1217         GOTO(out, rc = -EIO);
1218 out:
1219         dt_read_unlock(env, o);
1220         return rc;
1221 }
1222
1223 /**
1224  * This is helper function to get llog directory object. It is used by named
1225  * llog operations to find/insert/delete llog entry from llog directory.
1226  *
1227  * \param[in] env       execution environment
1228  * \param[in] ctxt      llog context
1229  *
1230  * \retval              dt_object of llog directory
1231  * \retval              ERR_PTR of negative value on error
1232  */
1233 static struct dt_object *llog_osd_dir_get(const struct lu_env *env,
1234                                           struct llog_ctxt *ctxt)
1235 {
1236         struct dt_device        *dt;
1237         struct dt_thread_info   *dti = dt_info(env);
1238         struct dt_object        *dir;
1239         int                      rc;
1240
1241         dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
1242         if (ctxt->loc_dir == NULL) {
1243                 rc = dt_root_get(env, dt, &dti->dti_fid);
1244                 if (rc)
1245                         return ERR_PTR(rc);
1246                 dir = dt_locate(env, dt, &dti->dti_fid);
1247
1248                 if (!IS_ERR(dir) && !dt_try_as_dir(env, dir)) {
1249                         dt_object_put(env, dir);
1250                         return ERR_PTR(-ENOTDIR);
1251                 }
1252         } else {
1253                 lu_object_get(&ctxt->loc_dir->do_lu);
1254                 dir = ctxt->loc_dir;
1255         }
1256
1257         return dir;
1258 }
1259
1260 /**
1261  * Implementation of the llog_operations::lop_open
1262  *
1263  * This function opens the llog by its logid or by name, it may open also
1264  * non existent llog and assing then new id to it.
1265  * The llog_open/llog_close pair works similar to lu_object_find/put,
1266  * the object may not exist prior open. The result of open is just dt_object
1267  * in the llog header.
1268  *
1269  * \param[in] env               execution environment
1270  * \param[in] handle            llog handle of the current llog
1271  * \param[in] logid             logid of llog to open (nameless llog)
1272  * \param[in] name              name of llog to open (named llog)
1273  * \param[in] open_param
1274  *                              LLOG_OPEN_NEW - new llog, may not exist
1275  *                              LLOG_OPEN_EXIST - old llog, must exist
1276  *
1277  * \retval                      0 on successful open, llog_handle::lgh_obj
1278  *                              contains the dt_object of the llog.
1279  * \retval                      negative value on error
1280  */
1281 static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle,
1282                          struct llog_logid *logid, char *name,
1283                          enum llog_open_param open_param)
1284 {
1285         struct llog_thread_info         *lgi = llog_info(env);
1286         struct llog_ctxt                *ctxt = handle->lgh_ctxt;
1287         struct dt_object                *o;
1288         struct dt_device                *dt;
1289         struct ls_device                *ls;
1290         struct local_oid_storage        *los = NULL;
1291         int                              rc = 0;
1292         bool new_id = false;
1293
1294         ENTRY;
1295
1296         LASSERT(env);
1297         LASSERT(ctxt);
1298         LASSERT(ctxt->loc_exp);
1299         LASSERT(ctxt->loc_exp->exp_obd);
1300         dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
1301         LASSERT(dt);
1302         if (ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1303                 struct lu_object_conf conf = { 0 };
1304                 if (logid != NULL) {
1305                         logid_to_fid(logid, &lgi->lgi_fid);
1306                 } else {
1307                         /* If logid == NULL, then it means the caller needs
1308                          * to allocate new FID (llog_cat_declare_add_rec()). */
1309                         rc = dt_fid_alloc(env, dt, &lgi->lgi_fid, NULL, NULL);
1310                         if (rc < 0)
1311                                 RETURN(rc);
1312                         rc = 0;
1313                         conf.loc_flags = LOC_F_NEW;
1314                 }
1315
1316                 o = dt_locate_at(env, dt, &lgi->lgi_fid,
1317                                  dt->dd_lu_dev.ld_site->ls_top_dev, &conf);
1318                 if (IS_ERR(o))
1319                         RETURN(PTR_ERR(o));
1320
1321                 goto after_open;
1322         }
1323
1324         ls = ls_device_get(dt);
1325         if (IS_ERR(ls))
1326                 RETURN(PTR_ERR(ls));
1327
1328         mutex_lock(&ls->ls_los_mutex);
1329         los = dt_los_find(ls, name != NULL ? FID_SEQ_LLOG_NAME : FID_SEQ_LLOG);
1330         mutex_unlock(&ls->ls_los_mutex);
1331         LASSERT(los);
1332         ls_device_put(env, ls);
1333
1334         LASSERT(handle);
1335
1336         if (logid != NULL) {
1337                 logid_to_fid(logid, &lgi->lgi_fid);
1338         } else if (name) {
1339                 struct dt_object *llog_dir;
1340
1341                 llog_dir = llog_osd_dir_get(env, ctxt);
1342                 if (IS_ERR(llog_dir))
1343                         GOTO(out, rc = PTR_ERR(llog_dir));
1344                 dt_read_lock(env, llog_dir, 0);
1345                 rc = dt_lookup_dir(env, llog_dir, name, &lgi->lgi_fid);
1346                 dt_read_unlock(env, llog_dir);
1347                 dt_object_put(env, llog_dir);
1348                 if (rc == -ENOENT && open_param == LLOG_OPEN_NEW) {
1349                         /* generate fid for new llog */
1350                         rc = local_object_fid_generate(env, los,
1351                                                        &lgi->lgi_fid);
1352                         new_id = true;
1353                 }
1354                 if (rc < 0)
1355                         GOTO(out, rc);
1356                 OBD_ALLOC(handle->lgh_name, strlen(name) + 1);
1357                 if (handle->lgh_name)
1358                         strcpy(handle->lgh_name, name);
1359                 else
1360                         GOTO(out, rc = -ENOMEM);
1361         } else {
1362                 LASSERTF(open_param & LLOG_OPEN_NEW, "%#x\n", open_param);
1363                 /* generate fid for new llog */
1364 generate:
1365                 rc = local_object_fid_generate(env, los, &lgi->lgi_fid);
1366                 if (rc < 0)
1367                         GOTO(out, rc);
1368                 new_id = true;
1369         }
1370         if (OBD_FAIL_PRECHECK(OBD_FAIL_MDS_LLOG_UMOUNT_RACE) &&
1371             cfs_fail_val == 1) {
1372                 cfs_fail_val = 2;
1373                 OBD_RACE(OBD_FAIL_MDS_LLOG_UMOUNT_RACE);
1374                 msleep(MSEC_PER_SEC);
1375         }
1376         o = ls_locate(env, ls, &lgi->lgi_fid, NULL);
1377         if (IS_ERR(o))
1378                 GOTO(out_name, rc = PTR_ERR(o));
1379
1380         if (dt_object_exists(o) && new_id) {
1381                 /* llog exists with just generated ID, e.g. some old llog file
1382                  * still is in use or is orphan, drop a warn and skip it. */
1383                 CDEBUG(D_INFO, "%s: llog exists with the same FID: "DFID
1384                        ", skipping\n",
1385                        o->do_lu.lo_dev->ld_obd->obd_name,
1386                        PFID(lu_object_fid(&o->do_lu)));
1387                 dt_object_put(env, o);
1388                 /* just skip this llog ID, we shouldn't delete it because we
1389                  * don't know exactly what is its purpose and state. */
1390                 goto generate;
1391         }
1392
1393 after_open:
1394         /* No new llog is expected but doesn't exist */
1395         if (open_param != LLOG_OPEN_NEW && !dt_object_exists(o)) {
1396                 CDEBUG(D_INFO, "%s: llog FID: "DFID" obj %p doesn`t exist\n",
1397                        o->do_lu.lo_dev->ld_obd->obd_name,
1398                        PFID(lu_object_fid(&o->do_lu)), o);
1399                 GOTO(out_put, rc = -ENOENT);
1400         }
1401         fid_to_logid(&lgi->lgi_fid, &handle->lgh_id);
1402         handle->lgh_obj = o;
1403         handle->private_data = los;
1404         LASSERT(handle->lgh_ctxt);
1405
1406         RETURN(rc);
1407
1408 out_put:
1409         dt_object_put(env, o);
1410 out_name:
1411         if (handle->lgh_name != NULL)
1412                 OBD_FREE(handle->lgh_name, strlen(name) + 1);
1413 out:
1414         if (los != NULL)
1415                 dt_los_put(los);
1416         RETURN(rc);
1417 }
1418
1419 /**
1420  * Get dir for regular fid log object
1421  *
1422  * Get directory for regular fid log object, and these regular fid log
1423  * object will be inserted under this directory, to satisfy the FS
1424  * consistency check, e2fsck etc.
1425  *
1426  * \param [in] env      execution environment
1427  * \param [in] dto      llog object
1428  *
1429  * \retval              pointer to the directory if it is found.
1430  * \retval              ERR_PTR(negative errno) if it fails.
1431  */
1432 struct dt_object *llog_osd_get_regular_fid_dir(const struct lu_env *env,
1433                                                struct dt_object *dto)
1434 {
1435         struct llog_thread_info *lgi = llog_info(env);
1436         struct seq_server_site *ss = dto->do_lu.lo_dev->ld_site->ld_seq_site;
1437         struct lu_seq_range     *range = &lgi->lgi_range;
1438         struct lu_fid           *dir_fid = &lgi->lgi_fid;
1439         struct dt_object        *dir;
1440         int                     rc;
1441         ENTRY;
1442
1443         fld_range_set_any(range);
1444         LASSERT(ss != NULL);
1445         rc = ss->ss_server_fld->lsf_seq_lookup(env, ss->ss_server_fld,
1446                                    fid_seq(lu_object_fid(&dto->do_lu)), range);
1447         if (rc < 0)
1448                 RETURN(ERR_PTR(rc));
1449
1450         lu_update_log_dir_fid(dir_fid, range->lsr_index);
1451         dir = dt_locate(env, lu2dt_dev(dto->do_lu.lo_dev), dir_fid);
1452         if (IS_ERR(dir))
1453                 RETURN(dir);
1454
1455         if (!dt_try_as_dir(env, dir)) {
1456                 dt_object_put(env, dir);
1457                 RETURN(ERR_PTR(-ENOTDIR));
1458         }
1459
1460         RETURN(dir);
1461 }
1462
1463 /**
1464  * Add llog object with regular FID to name entry
1465  *
1466  * Add llog object with regular FID to name space, and each llog
1467  * object on each MDT will be /update_log_dir/[seq:oid:ver],
1468  * so to satisfy the namespace consistency check, e2fsck etc.
1469  *
1470  * \param [in] env      execution environment
1471  * \param [in] dto      llog object
1472  * \param [in] th       thandle
1473  * \param [in] declare  if it is declare or execution
1474  *
1475  * \retval              0 if insertion succeeds.
1476  * \retval              negative errno if insertion fails.
1477  */
1478 static int
1479 llog_osd_regular_fid_add_name_entry(const struct lu_env *env,
1480                                     struct dt_object *dto,
1481                                     struct thandle *th, bool declare)
1482 {
1483         struct llog_thread_info *lgi = llog_info(env);
1484         const struct lu_fid     *fid = lu_object_fid(&dto->do_lu);
1485         struct dt_insert_rec    *rec = &lgi->lgi_dt_rec;
1486         struct dt_object        *dir;
1487         char                    *name = lgi->lgi_name;
1488         int                     rc;
1489         ENTRY;
1490
1491         if (!fid_is_norm(fid))
1492                 RETURN(0);
1493
1494         dir = llog_osd_get_regular_fid_dir(env, dto);
1495         if (IS_ERR(dir))
1496                 RETURN(PTR_ERR(dir));
1497
1498         rec->rec_fid = fid;
1499         rec->rec_type = S_IFREG;
1500         snprintf(name, sizeof(lgi->lgi_name), DFID, PFID(fid));
1501         dt_write_lock(env, dir, 0);
1502         if (declare) {
1503                 rc = dt_declare_insert(env, dir, (struct dt_rec *)rec,
1504                                (struct dt_key *)name, th);
1505         } else {
1506                 rc = dt_insert(env, dir, (struct dt_rec *)rec,
1507                                (struct dt_key *)name, th);
1508         }
1509         dt_write_unlock(env, dir);
1510
1511         dt_object_put(env, dir);
1512         RETURN(rc);
1513 }
1514
1515
1516 /**
1517  * Implementation of the llog_operations::lop_declare_create
1518  *
1519  * This function declares the llog create. It declares also name insert
1520  * into llog directory in case of named llog.
1521  *
1522  * \param[in] env       execution environment
1523  * \param[in] res       llog handle of the current llog
1524  * \param[in] th        current transaction handle
1525  *
1526  * \retval              0 on successful create declaration
1527  * \retval              negative value on error
1528  */
1529 static int llog_osd_declare_create(const struct lu_env *env,
1530                                    struct llog_handle *res, struct thandle *th)
1531 {
1532         struct llog_thread_info         *lgi = llog_info(env);
1533         struct dt_insert_rec            *rec = &lgi->lgi_dt_rec;
1534         struct local_oid_storage        *los;
1535         struct dt_object                *o;
1536         int                              rc;
1537
1538         ENTRY;
1539
1540         LASSERT(res->lgh_obj);
1541         LASSERT(th);
1542
1543         /* object can be created by another thread */
1544         o = res->lgh_obj;
1545         if (dt_object_exists(o))
1546                 RETURN(0);
1547
1548         if (res->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1549                 struct llog_thread_info *lgi = llog_info(env);
1550
1551                 lgi->lgi_attr.la_valid = LA_MODE | LA_SIZE;
1552                 lgi->lgi_attr.la_size = 0;
1553                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
1554                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
1555
1556                 rc = dt_declare_create(env, o, &lgi->lgi_attr, NULL,
1557                                        &lgi->lgi_dof, th);
1558                 if (rc < 0)
1559                         RETURN(rc);
1560
1561
1562                 rc = llog_osd_regular_fid_add_name_entry(env, o, th, true);
1563
1564                 RETURN(rc);
1565         }
1566         los = res->private_data;
1567         LASSERT(los);
1568
1569         rc = llog_osd_declare_new_object(env, los, o, th);
1570         if (rc)
1571                 RETURN(rc);
1572
1573         /* do not declare header initialization here as it's declared
1574          * in llog_osd_declare_write_rec() which is always called */
1575
1576         if (res->lgh_name) {
1577                 struct dt_object *llog_dir;
1578
1579                 llog_dir = llog_osd_dir_get(env, res->lgh_ctxt);
1580                 if (IS_ERR(llog_dir))
1581                         RETURN(PTR_ERR(llog_dir));
1582                 logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
1583                 rec->rec_fid = &lgi->lgi_fid;
1584                 rec->rec_type = S_IFREG;
1585                 rc = dt_declare_insert(env, llog_dir,
1586                                        (struct dt_rec *)rec,
1587                                        (struct dt_key *)res->lgh_name, th);
1588                 dt_object_put(env, llog_dir);
1589                 if (rc)
1590                         CERROR("%s: can't declare named llog %s: rc = %d\n",
1591                                o->do_lu.lo_dev->ld_obd->obd_name,
1592                                res->lgh_name, rc);
1593         }
1594         RETURN(rc);
1595 }
1596
1597 /**
1598  * Implementation of the llog_operations::lop_create
1599  *
1600  * This function creates the llog according with llog_handle::lgh_obj
1601  * and llog_handle::lgh_name.
1602  *
1603  * \param[in] env       execution environment
1604  * \param[in] res       llog handle of the current llog
1605  * \param[in] th        current transaction handle
1606  *
1607  * \retval              0 on successful create
1608  * \retval              negative value on error
1609  */
1610 static int llog_osd_create(const struct lu_env *env, struct llog_handle *res,
1611                            struct thandle *th)
1612 {
1613         struct llog_thread_info *lgi = llog_info(env);
1614         struct dt_insert_rec    *rec = &lgi->lgi_dt_rec;
1615         struct local_oid_storage *los;
1616         struct dt_object        *o;
1617         int                      rc = 0;
1618
1619         ENTRY;
1620
1621         LASSERT(env);
1622         o = res->lgh_obj;
1623         LASSERT(o);
1624
1625         /* llog can be already created */
1626         if (dt_object_exists(o))
1627                 RETURN(-EEXIST);
1628
1629         if (res->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1630                 struct llog_thread_info *lgi = llog_info(env);
1631
1632                 lgi->lgi_attr.la_valid = LA_MODE | LA_SIZE | LA_TYPE;
1633                 lgi->lgi_attr.la_size = 0;
1634                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
1635                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
1636
1637                 dt_write_lock(env, o, 0);
1638                 rc = dt_create(env, o, &lgi->lgi_attr, NULL,
1639                                &lgi->lgi_dof, th);
1640                 dt_write_unlock(env, o);
1641                 if (rc < 0)
1642                         RETURN(rc);
1643
1644                 rc = llog_osd_regular_fid_add_name_entry(env, o, th, false);
1645
1646                 RETURN(rc);
1647         }
1648
1649         los = res->private_data;
1650         LASSERT(los);
1651
1652         dt_write_lock(env, o, 0);
1653         if (!dt_object_exists(o))
1654                 rc = llog_osd_create_new_object(env, los, o, th);
1655         else
1656                 rc = -EEXIST;
1657
1658         dt_write_unlock(env, o);
1659         if (rc)
1660                 RETURN(rc);
1661
1662         if (res->lgh_name) {
1663                 struct dt_object *llog_dir;
1664
1665                 llog_dir = llog_osd_dir_get(env, res->lgh_ctxt);
1666                 if (IS_ERR(llog_dir))
1667                         RETURN(PTR_ERR(llog_dir));
1668
1669                 logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
1670                 rec->rec_fid = &lgi->lgi_fid;
1671                 rec->rec_type = S_IFREG;
1672                 dt_read_lock(env, llog_dir, 0);
1673                 rc = dt_insert(env, llog_dir, (struct dt_rec *)rec,
1674                                (struct dt_key *)res->lgh_name, th);
1675                 dt_read_unlock(env, llog_dir);
1676                 dt_object_put(env, llog_dir);
1677                 if (rc)
1678                         CERROR("%s: can't create named llog %s: rc = %d\n",
1679                                o->do_lu.lo_dev->ld_obd->obd_name,
1680                                res->lgh_name, rc);
1681         }
1682         RETURN(rc);
1683 }
1684
1685 /**
1686  * Implementation of the llog_operations::lop_close
1687  *
1688  * This function closes the llog. It just put llog object and referenced
1689  * local storage.
1690  *
1691  * \param[in] env       execution environment
1692  * \param[in] handle    llog handle of the current llog
1693  *
1694  * \retval              0 on successful llog close
1695  * \retval              negative value on error
1696  */
1697 static int llog_osd_close(const struct lu_env *env, struct llog_handle *handle)
1698 {
1699         struct local_oid_storage        *los;
1700         int                              rc = 0;
1701
1702         ENTRY;
1703
1704         LASSERT(handle->lgh_obj);
1705
1706         if (handle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1707                 /* Remove the object from the cache, otherwise it may
1708                  * hold LOD being released during cleanup process */
1709                 dt_object_put_nocache(env, handle->lgh_obj);
1710                 LASSERT(handle->private_data == NULL);
1711                 RETURN(rc);
1712         } else {
1713                 dt_object_put(env, handle->lgh_obj);
1714         }
1715         los = handle->private_data;
1716         LASSERT(los);
1717         dt_los_put(los);
1718
1719         if (handle->lgh_name)
1720                 OBD_FREE(handle->lgh_name, strlen(handle->lgh_name) + 1);
1721
1722         RETURN(rc);
1723 }
1724
1725 /**
1726  * delete llog object name entry
1727  *
1728  * Delete llog object (with regular FID) from name space (under
1729  * update_log_dir).
1730  *
1731  * \param [in] env      execution environment
1732  * \param [in] dto      llog object
1733  * \param [in] th       thandle
1734  * \param [in] declare  if it is declare or execution
1735  *
1736  * \retval              0 if deletion succeeds.
1737  * \retval              negative errno if deletion fails.
1738  */
1739 static int
1740 llog_osd_regular_fid_del_name_entry(const struct lu_env *env,
1741                                     struct dt_object *dto,
1742                                     struct thandle *th, bool declare)
1743 {
1744         struct llog_thread_info *lgi = llog_info(env);
1745         const struct lu_fid     *fid = lu_object_fid(&dto->do_lu);
1746         struct dt_object        *dir;
1747         char                    *name = lgi->lgi_name;
1748         int                     rc;
1749         ENTRY;
1750
1751         if (!fid_is_norm(fid))
1752                 RETURN(0);
1753
1754         dir = llog_osd_get_regular_fid_dir(env, dto);
1755         if (IS_ERR(dir))
1756                 RETURN(PTR_ERR(dir));
1757
1758         snprintf(name, sizeof(lgi->lgi_name), DFID, PFID(fid));
1759         dt_write_lock(env, dir, 0);
1760         if (declare) {
1761                 rc = dt_declare_delete(env, dir, (struct dt_key *)name,
1762                                        th);
1763         } else {
1764                 rc = dt_delete(env, dir, (struct dt_key *)name, th);
1765         }
1766         dt_write_unlock(env, dir);
1767
1768         dt_object_put(env, dir);
1769         RETURN(rc);
1770 }
1771
1772 /**
1773  * Implementation of the llog_operations::lop_declare_destroy
1774  *
1775  * This function declare destroys the llog and deletes also entry in the
1776  * llog directory in case of named llog. Llog should be opened prior that.
1777  *
1778  * \param[in] env               execution environment
1779  * \param[in] loghandle llog handle of the current llog
1780  *
1781  * \retval              0 on successful destroy
1782  * \retval              negative value on error
1783  */
1784 static int llog_osd_declare_destroy(const struct lu_env *env,
1785                                     struct llog_handle *loghandle,
1786                                     struct thandle *th)
1787 {
1788         struct llog_ctxt        *ctxt;
1789         struct dt_object        *o, *llog_dir = NULL;
1790         int                      rc;
1791
1792         ENTRY;
1793
1794         ctxt = loghandle->lgh_ctxt;
1795         LASSERT(ctxt);
1796
1797         o = loghandle->lgh_obj;
1798         LASSERT(o);
1799
1800         if (loghandle->lgh_name) {
1801                 llog_dir = llog_osd_dir_get(env, ctxt);
1802                 if (IS_ERR(llog_dir))
1803                         RETURN(PTR_ERR(llog_dir));
1804
1805                 rc = dt_declare_delete(env, llog_dir,
1806                                        (struct dt_key *)loghandle->lgh_name,
1807                                        th);
1808                 if (rc < 0)
1809                         GOTO(out_put, rc);
1810         }
1811
1812         rc = dt_declare_ref_del(env, o, th);
1813         if (rc < 0)
1814                 GOTO(out_put, rc);
1815
1816         rc = dt_declare_destroy(env, o, th);
1817         if (rc < 0)
1818                 GOTO(out_put, rc);
1819
1820         if (loghandle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1821                 rc = llog_osd_regular_fid_del_name_entry(env, o, th, true);
1822                 if (rc < 0)
1823                         GOTO(out_put, rc);
1824         }
1825
1826 out_put:
1827         if (!(IS_ERR_OR_NULL(llog_dir)))
1828                 dt_object_put(env, llog_dir);
1829
1830         RETURN(rc);
1831 }
1832
1833
1834 /**
1835  * Implementation of the llog_operations::lop_destroy
1836  *
1837  * This function destroys the llog and deletes also entry in the
1838  * llog directory in case of named llog. Llog should be opened prior that.
1839  * Destroy method is not part of external transaction and does everything
1840  * inside.
1841  *
1842  * \param[in] env               execution environment
1843  * \param[in] loghandle llog handle of the current llog
1844  *
1845  * \retval              0 on successful destroy
1846  * \retval              negative value on error
1847  */
1848 static int llog_osd_destroy(const struct lu_env *env,
1849                             struct llog_handle *loghandle, struct thandle *th)
1850 {
1851         struct llog_ctxt        *ctxt;
1852         struct dt_object        *o, *llog_dir = NULL;
1853         int                      rc;
1854
1855         ENTRY;
1856
1857         ctxt = loghandle->lgh_ctxt;
1858         LASSERT(ctxt != NULL);
1859
1860         o = loghandle->lgh_obj;
1861         LASSERT(o != NULL);
1862
1863         dt_write_lock(env, o, 0);
1864         if (!llog_osd_exist(loghandle))
1865                 GOTO(out_unlock, rc = 0);
1866
1867         if (loghandle->lgh_name) {
1868                 llog_dir = llog_osd_dir_get(env, ctxt);
1869                 if (IS_ERR(llog_dir))
1870                         GOTO(out_unlock, rc = PTR_ERR(llog_dir));
1871
1872                 dt_read_lock(env, llog_dir, 0);
1873                 rc = dt_delete(env, llog_dir,
1874                                (struct dt_key *)loghandle->lgh_name,
1875                                th);
1876                 dt_read_unlock(env, llog_dir);
1877                 if (rc) {
1878                         CERROR("%s: can't remove llog %s: rc = %d\n",
1879                                o->do_lu.lo_dev->ld_obd->obd_name,
1880                                loghandle->lgh_name, rc);
1881                         GOTO(out_unlock, rc);
1882                 }
1883         }
1884
1885         dt_ref_del(env, o, th);
1886         rc = dt_destroy(env, o, th);
1887         if (rc < 0)
1888                 GOTO(out_unlock, rc);
1889
1890         loghandle->lgh_destroyed = true;
1891         if (loghandle->lgh_ctxt->loc_flags & LLOG_CTXT_FLAG_NORMAL_FID) {
1892                 rc = llog_osd_regular_fid_del_name_entry(env, o, th, false);
1893                 if (rc < 0)
1894                         GOTO(out_unlock, rc);
1895         }
1896
1897 out_unlock:
1898         dt_write_unlock(env, o);
1899         if (!(IS_ERR_OR_NULL(llog_dir)))
1900                 dt_object_put(env, llog_dir);
1901         RETURN(rc);
1902 }
1903
1904 /**
1905  * Implementation of the llog_operations::lop_setup
1906  *
1907  * This function setup the llog on local storage.
1908  *
1909  * \param[in] env       execution environment
1910  * \param[in] obd       obd device the llog belongs to
1911  * \param[in] olg       the llog group, it is always zero group now.
1912  * \param[in] ctxt_idx  the llog index, it defines the purpose of this llog.
1913  *                      Every new llog type have to use own index.
1914  * \param[in] disk_obd  the storage obd, where llog is stored.
1915  *
1916  * \retval              0 on successful llog setup
1917  * \retval              negative value on error
1918  */
1919 static int llog_osd_setup(const struct lu_env *env, struct obd_device *obd,
1920                           struct obd_llog_group *olg, int ctxt_idx,
1921                           struct obd_device *disk_obd)
1922 {
1923         struct llog_thread_info         *lgi = llog_info(env);
1924         struct llog_ctxt                *ctxt;
1925         int                              rc = 0;
1926         ENTRY;
1927
1928         LASSERT(obd);
1929         LASSERT(olg->olg_ctxts[ctxt_idx]);
1930
1931         ctxt = llog_ctxt_get(olg->olg_ctxts[ctxt_idx]);
1932         LASSERT(ctxt);
1933
1934         if (disk_obd == NULL)
1935                 GOTO(out, rc = 0);
1936
1937         /* initialize data allowing to generate new fids,
1938          * literally we need a sequece */
1939         lgi->lgi_fid.f_seq = FID_SEQ_LLOG;
1940         lgi->lgi_fid.f_oid = 1;
1941         lgi->lgi_fid.f_ver = 0;
1942         rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt,
1943                                     &lgi->lgi_fid,
1944                                     &ctxt->loc_los_nameless);
1945         if (rc != 0)
1946                 GOTO(out, rc);
1947
1948         lgi->lgi_fid.f_seq = FID_SEQ_LLOG_NAME;
1949         lgi->lgi_fid.f_oid = 1;
1950         lgi->lgi_fid.f_ver = 0;
1951         rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt,
1952                                     &lgi->lgi_fid,
1953                                     &ctxt->loc_los_named);
1954         if (rc != 0) {
1955                 local_oid_storage_fini(env, ctxt->loc_los_nameless);
1956                 ctxt->loc_los_nameless = NULL;
1957         }
1958
1959         GOTO(out, rc);
1960
1961 out:
1962         llog_ctxt_put(ctxt);
1963         return rc;
1964 }
1965
1966 /**
1967  * Implementation of the llog_operations::lop_cleanup
1968  *
1969  * This function cleanups the llog on local storage.
1970  *
1971  * \param[in] env       execution environment
1972  * \param[in] ctxt      the llog context
1973  *
1974  * \retval              0
1975  */
1976 static int llog_osd_cleanup(const struct lu_env *env, struct llog_ctxt *ctxt)
1977 {
1978         if (ctxt->loc_los_nameless != NULL) {
1979                 local_oid_storage_fini(env, ctxt->loc_los_nameless);
1980                 ctxt->loc_los_nameless = NULL;
1981         }
1982
1983         if (ctxt->loc_los_named != NULL) {
1984                 local_oid_storage_fini(env, ctxt->loc_los_named);
1985                 ctxt->loc_los_named = NULL;
1986         }
1987
1988         return 0;
1989 }
1990
1991 const struct llog_operations llog_osd_ops = {
1992         .lop_next_block         = llog_osd_next_block,
1993         .lop_prev_block         = llog_osd_prev_block,
1994         .lop_read_header        = llog_osd_read_header,
1995         .lop_declare_destroy    = llog_osd_declare_destroy,
1996         .lop_destroy            = llog_osd_destroy,
1997         .lop_setup              = llog_osd_setup,
1998         .lop_cleanup            = llog_osd_cleanup,
1999         .lop_open               = llog_osd_open,
2000         .lop_exist              = llog_osd_exist,
2001         .lop_declare_create     = llog_osd_declare_create,
2002         .lop_create             = llog_osd_create,
2003         .lop_declare_write_rec  = llog_osd_declare_write_rec,
2004         .lop_write_rec          = llog_osd_write_rec,
2005         .lop_close              = llog_osd_close,
2006 };
2007 EXPORT_SYMBOL(llog_osd_ops);
2008
2009 const struct llog_operations llog_common_cat_ops = {
2010         .lop_next_block         = llog_osd_next_block,
2011         .lop_prev_block         = llog_osd_prev_block,
2012         .lop_read_header        = llog_osd_read_header,
2013         .lop_declare_destroy    = llog_osd_declare_destroy,
2014         .lop_destroy            = llog_osd_destroy,
2015         .lop_setup              = llog_osd_setup,
2016         .lop_cleanup            = llog_osd_cleanup,
2017         .lop_open               = llog_osd_open,
2018         .lop_exist              = llog_osd_exist,
2019         .lop_declare_create     = llog_osd_declare_create,
2020         .lop_create             = llog_osd_create,
2021         .lop_declare_write_rec  = llog_osd_declare_write_rec,
2022         .lop_write_rec          = llog_osd_write_rec,
2023         .lop_close              = llog_osd_close,
2024         .lop_add                = llog_cat_add_rec,
2025         .lop_declare_add        = llog_cat_declare_add_rec,
2026 };
2027 EXPORT_SYMBOL(llog_common_cat_ops);
2028
2029 /**
2030  * Read the special file which contains the list of llog catalogs IDs
2031  *
2032  * This function reads the CATALOGS file which contains the array of llog
2033  * catalogs IDs. The main purpose of this file is to store OSP llogs indexed
2034  * by OST/MDT number.
2035  *
2036  * \param[in]  env              execution environment
2037  * \param[in]  d                corresponding storage device
2038  * \param[in]  idx              position to start from, usually OST/MDT index
2039  * \param[in]  count            how many catalog IDs to read
2040  * \param[out] idarray          the buffer for the data. If it is NULL then
2041  *                              function returns just number of catalog IDs
2042  *                              in the file.
2043  * \param[in]  fid              LLOG_CATALOGS_OID for CATALOG object
2044  *
2045  * \retval                      0 on successful read of catalog IDs
2046  * \retval                      negative value on error
2047  * \retval                      positive value which is number of records in
2048  *                              the file if \a idarray is NULL
2049  */
2050 int llog_osd_get_cat_list(const struct lu_env *env, struct dt_device *d,
2051                           int idx, int count, struct llog_catid *idarray,
2052                           const struct lu_fid *fid)
2053 {
2054         struct llog_thread_info *lgi = llog_info(env);
2055         struct dt_object        *o = NULL;
2056         struct thandle          *th;
2057         int                      rc, size;
2058
2059         ENTRY;
2060
2061         LASSERT(d);
2062
2063         size = sizeof(*idarray) * count;
2064         lgi->lgi_off = idx *  sizeof(*idarray);
2065
2066         lgi->lgi_fid = *fid;
2067         o = dt_locate(env, d, &lgi->lgi_fid);
2068         if (IS_ERR(o))
2069                 RETURN(PTR_ERR(o));
2070
2071         if (!dt_object_exists(o)) {
2072                 th = dt_trans_create(env, d);
2073                 if (IS_ERR(th))
2074                         GOTO(out, rc = PTR_ERR(th));
2075
2076                 lgi->lgi_attr.la_valid = LA_MODE | LA_TYPE;
2077                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
2078                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
2079
2080                 th->th_wait_submit = 1;
2081                 /* Make the llog object creation synchronization, so
2082                  * it will be reliable to the reference, especially
2083                  * for remote reference */
2084                 th->th_sync = 1;
2085
2086                 rc = dt_declare_create(env, o, &lgi->lgi_attr, NULL,
2087                                        &lgi->lgi_dof, th);
2088                 if (rc)
2089                         GOTO(out_trans, rc);
2090
2091                 rc = dt_trans_start_local(env, d, th);
2092                 if (rc)
2093                         GOTO(out_trans, rc);
2094
2095                 dt_write_lock(env, o, 0);
2096                 if (!dt_object_exists(o))
2097                         rc = dt_create(env, o, &lgi->lgi_attr, NULL,
2098                                        &lgi->lgi_dof, th);
2099                 dt_write_unlock(env, o);
2100 out_trans:
2101                 dt_trans_stop(env, d, th);
2102                 if (rc)
2103                         GOTO(out, rc);
2104         }
2105
2106         rc = dt_attr_get(env, o, &lgi->lgi_attr);
2107         if (rc)
2108                 GOTO(out, rc);
2109
2110         if (!S_ISREG(lgi->lgi_attr.la_mode)) {
2111                 CERROR("%s: CATALOGS is not a regular file!: mode = %o\n",
2112                        o->do_lu.lo_dev->ld_obd->obd_name,
2113                        lgi->lgi_attr.la_mode);
2114                 GOTO(out, rc = -ENOENT);
2115         }
2116
2117         CDEBUG(D_CONFIG, "cat list: disk size=%d, read=%d\n",
2118                (int)lgi->lgi_attr.la_size, size);
2119
2120         /* return just number of llogs */
2121         if (idarray == NULL) {
2122                 rc = lgi->lgi_attr.la_size / sizeof(*idarray);
2123                 GOTO(out, rc);
2124         }
2125
2126         /* read for new ost index or for empty file */
2127         memset(idarray, 0, size);
2128         if (lgi->lgi_attr.la_size <= lgi->lgi_off)
2129                 GOTO(out, rc = 0);
2130         if (lgi->lgi_attr.la_size < lgi->lgi_off + size)
2131                 size = lgi->lgi_attr.la_size - lgi->lgi_off;
2132
2133         lgi->lgi_buf.lb_buf = idarray;
2134         lgi->lgi_buf.lb_len = size;
2135         rc = dt_record_read(env, o, &lgi->lgi_buf, &lgi->lgi_off);
2136         /* -EFAULT means the llog is a sparse file. This is not an error
2137          * after arbitrary OST index is supported. */
2138         if (rc < 0 && rc != -EFAULT) {
2139                 CERROR("%s: error reading CATALOGS: rc = %d\n",
2140                        o->do_lu.lo_dev->ld_obd->obd_name,  rc);
2141                 GOTO(out, rc);
2142         }
2143
2144         EXIT;
2145 out:
2146         dt_object_put(env, o);
2147         RETURN(rc);
2148 }
2149 EXPORT_SYMBOL(llog_osd_get_cat_list);
2150
2151 /**
2152  * Write the special file which contains the list of llog catalogs IDs
2153  *
2154  * This function writes the CATALOG file which contains the array of llog
2155  * catalogs IDs. It is used mostly to store OSP llogs indexed by OST/MDT
2156  * number.
2157  *
2158  * \param[in]  env      execution environment
2159  * \param[in]  d        corresponding storage device
2160  * \param[in]  idx      position to start from, usually OST/MDT index
2161  * \param[in]  count    how many catalog IDs to write
2162  * \param[out] idarray  the buffer with the data to write.
2163  * \param[in]  fid      LLOG_CATALOGS_OID for CATALOG object
2164  *
2165  * \retval              0 on successful write of catalog IDs
2166  * \retval              negative value on error
2167  */
2168 int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d,
2169                           int idx, int count, struct llog_catid *idarray,
2170                           const struct lu_fid *fid)
2171 {
2172         struct llog_thread_info *lgi = llog_info(env);
2173         struct dt_object        *o = NULL;
2174         struct thandle          *th;
2175         int                      rc, size;
2176
2177         if (count == 0)
2178                 RETURN(0);
2179
2180         LASSERT(d);
2181
2182         size = sizeof(*idarray) * count;
2183         lgi->lgi_off = idx * sizeof(*idarray);
2184         lgi->lgi_fid = *fid;
2185
2186         o = dt_locate(env, d, &lgi->lgi_fid);
2187         if (IS_ERR(o))
2188                 RETURN(PTR_ERR(o));
2189
2190         if (!dt_object_exists(o))
2191                 GOTO(out, rc = -ENOENT);
2192
2193         rc = dt_attr_get(env, o, &lgi->lgi_attr);
2194         if (rc)
2195                 GOTO(out, rc);
2196
2197         if (!S_ISREG(lgi->lgi_attr.la_mode)) {
2198                 CERROR("%s: CATALOGS is not a regular file!: mode = %o\n",
2199                        o->do_lu.lo_dev->ld_obd->obd_name,
2200                        lgi->lgi_attr.la_mode);
2201                 GOTO(out, rc = -ENOENT);
2202         }
2203
2204         th = dt_trans_create(env, d);
2205         if (IS_ERR(th))
2206                 GOTO(out, rc = PTR_ERR(th));
2207
2208         lgi->lgi_buf.lb_len = size;
2209         lgi->lgi_buf.lb_buf = idarray;
2210         rc = dt_declare_record_write(env, o, &lgi->lgi_buf, lgi->lgi_off, th);
2211         if (rc)
2212                 GOTO(out_trans, rc);
2213
2214         /* For update log, this happens during initialization,
2215          * see lod_sub_prep_llog(), and we need make sure catlog
2216          * file ID is written to catlist file(committed) before
2217          * cross-MDT operation write update records to catlog FILE,
2218          * otherwise, during failover these update records might
2219          * missing */
2220         if (fid_is_update_log(fid))
2221                 th->th_sync = 1;
2222
2223         rc = dt_trans_start_local(env, d, th);
2224         if (rc)
2225                 GOTO(out_trans, rc);
2226
2227         th->th_wait_submit = 1;
2228
2229         rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
2230         if (rc)
2231                 CDEBUG(D_INODE, "can't write CATALOGS at index %d: rc = %d\n",
2232                        idx, rc);
2233 out_trans:
2234         dt_trans_stop(env, d, th);
2235 out:
2236         dt_object_put(env, o);
2237         RETURN(rc);
2238 }
2239 EXPORT_SYMBOL(llog_osd_put_cat_list);