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