Whamcloud - gitweb
LU-2034 changelog: redo changelog using OSD llog
[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.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2011, 2012, Intel, Inc.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * lustre/obdclass/llog_osd.c - low level llog routines on top of OSD API
37  *
38  * Author: Alexey Zhuravlev <alexey.zhuravlev@intel.com>
39  * Author: Mikhail Pershin <mike.pershin@intel.com>
40  */
41
42 #define DEBUG_SUBSYSTEM S_LOG
43
44 #ifndef EXPORT_SYMTAB
45 #define EXPORT_SYMTAB
46 #endif
47
48 #include <obd.h>
49 #include <obd_class.h>
50 #include <lustre_fid.h>
51 #include <dt_object.h>
52
53 #include "llog_internal.h"
54 #include "local_storage.h"
55
56 /*
57  * - multi-chunks or big-declaration approach
58  * - use unique sequence instead of llog sb tracking unique ids
59  * - re-use existing environment
60  * - named llog support (can be used for testing only at the present)
61  * - llog_origin_connect() work with OSD API
62  */
63
64 static int llog_osd_declare_new_object(const struct lu_env *env,
65                                        struct local_oid_storage *los,
66                                        struct dt_object *o,
67                                        struct thandle *th)
68 {
69         struct llog_thread_info *lgi = llog_info(env);
70
71         lgi->lgi_attr.la_valid = LA_MODE;
72         lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
73         lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
74
75         return local_object_declare_create(env, los, o, &lgi->lgi_attr,
76                                            &lgi->lgi_dof, th);
77 }
78
79 static int llog_osd_create_new_object(const struct lu_env *env,
80                                       struct local_oid_storage *los,
81                                       struct dt_object *o,
82                                       struct thandle *th)
83 {
84         struct llog_thread_info *lgi = llog_info(env);
85
86         lgi->lgi_attr.la_valid = LA_MODE;
87         lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
88         lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
89
90         return local_object_create(env, los, o, &lgi->lgi_attr,
91                                    &lgi->lgi_dof, th);
92 }
93
94 static int llog_osd_pad(const struct lu_env *env, struct dt_object *o,
95                         loff_t *off, int len, int index, struct thandle *th)
96 {
97         struct llog_thread_info *lgi = llog_info(env);
98         int                      rc;
99
100         ENTRY;
101
102         LASSERT(th);
103         LASSERT(off);
104         LASSERT(len >= LLOG_MIN_REC_SIZE && (len & 0x7) == 0);
105
106         lgi->lgi_tail.lrt_len = lgi->lgi_lrh.lrh_len = len;
107         lgi->lgi_tail.lrt_index = lgi->lgi_lrh.lrh_index = index;
108         lgi->lgi_lrh.lrh_type = LLOG_PAD_MAGIC;
109
110         lgi->lgi_buf.lb_buf = &lgi->lgi_lrh;
111         lgi->lgi_buf.lb_len = sizeof(lgi->lgi_lrh);
112         rc = dt_record_write(env, o, &lgi->lgi_buf, off, th);
113         if (rc) {
114                 CERROR("%s: error writing padding record: rc = %d\n",
115                        o->do_lu.lo_dev->ld_obd->obd_name, rc);
116                 RETURN(rc);
117         }
118
119         lgi->lgi_buf.lb_buf = &lgi->lgi_tail;
120         lgi->lgi_buf.lb_len = sizeof(lgi->lgi_tail);
121         *off += len - sizeof(lgi->lgi_lrh) - sizeof(lgi->lgi_tail);
122         rc = dt_record_write(env, o, &lgi->lgi_buf, off, th);
123         if (rc)
124                 CERROR("%s: error writing padding record: rc = %d\n",
125                        o->do_lu.lo_dev->ld_obd->obd_name, rc);
126
127         RETURN(rc);
128 }
129
130 static int llog_osd_write_blob(const struct lu_env *env, struct dt_object *o,
131                                struct llog_rec_hdr *rec, void *buf,
132                                loff_t *off, struct thandle *th)
133 {
134         struct llog_thread_info *lgi = llog_info(env);
135         int                      buflen = rec->lrh_len;
136         int                      rc;
137
138         ENTRY;
139
140         LASSERT(env);
141         LASSERT(o);
142
143         if (buflen == 0)
144                 CWARN("0-length record\n");
145
146         CDEBUG(D_OTHER, "write blob with type %x, buf %p/%u at off %llu\n",
147                rec->lrh_type, buf, buflen, *off);
148
149         if (!buf) {
150                 lgi->lgi_buf.lb_len = buflen;
151                 lgi->lgi_buf.lb_buf = rec;
152                 rc = dt_record_write(env, o, &lgi->lgi_buf, off, th);
153                 if (rc)
154                         CERROR("%s: error writing log record: rc = %d\n",
155                                o->do_lu.lo_dev->ld_obd->obd_name, rc);
156                 RETURN(rc);
157         }
158
159         /* the buf case */
160         rec->lrh_len = sizeof(*rec) + buflen + sizeof(lgi->lgi_tail);
161         lgi->lgi_buf.lb_len = sizeof(*rec);
162         lgi->lgi_buf.lb_buf = rec;
163         rc = dt_record_write(env, o, &lgi->lgi_buf, off, th);
164         if (rc) {
165                 CERROR("%s: error writing log hdr: rc = %d\n",
166                        o->do_lu.lo_dev->ld_obd->obd_name, rc);
167                 GOTO(out, rc);
168         }
169
170         lgi->lgi_buf.lb_len = buflen;
171         lgi->lgi_buf.lb_buf = buf;
172         rc = dt_record_write(env, o, &lgi->lgi_buf, off, th);
173         if (rc) {
174                 CERROR("%s: error writing log buffer: rc = %d\n",
175                        o->do_lu.lo_dev->ld_obd->obd_name,  rc);
176                 GOTO(out, rc);
177         }
178
179         lgi->lgi_tail.lrt_len = rec->lrh_len;
180         lgi->lgi_tail.lrt_index = rec->lrh_index;
181         lgi->lgi_buf.lb_len = sizeof(lgi->lgi_tail);
182         lgi->lgi_buf.lb_buf = &lgi->lgi_tail;
183         rc = dt_record_write(env, o, &lgi->lgi_buf, off, th);
184         if (rc)
185                 CERROR("%s: error writing log tail: rc = %d\n",
186                        o->do_lu.lo_dev->ld_obd->obd_name, rc);
187 out:
188         RETURN(rc);
189 }
190
191 static int llog_osd_read_header(const struct lu_env *env,
192                                 struct llog_handle *handle)
193 {
194         struct llog_rec_hdr     *llh_hdr;
195         struct dt_object        *o;
196         struct llog_thread_info *lgi;
197         int                      rc;
198
199         ENTRY;
200
201         LASSERT(sizeof(*handle->lgh_hdr) == LLOG_CHUNK_SIZE);
202
203         o = handle->lgh_obj;
204         LASSERT(o);
205
206         lgi = llog_info(env);
207
208         rc = dt_attr_get(env, o, &lgi->lgi_attr, NULL);
209         if (rc)
210                 RETURN(rc);
211
212         LASSERT(lgi->lgi_attr.la_valid & LA_SIZE);
213
214         if (lgi->lgi_attr.la_size == 0) {
215                 CDEBUG(D_HA, "not reading header from 0-byte log\n");
216                 RETURN(LLOG_EEMPTY);
217         }
218
219         lgi->lgi_off = 0;
220         lgi->lgi_buf.lb_buf = handle->lgh_hdr;
221         lgi->lgi_buf.lb_len = LLOG_CHUNK_SIZE;
222
223         rc = dt_record_read(env, o, &lgi->lgi_buf, &lgi->lgi_off);
224         if (rc) {
225                 CERROR("%s: error reading log header from "DFID": rc = %d\n",
226                        o->do_lu.lo_dev->ld_obd->obd_name,
227                        PFID(lu_object_fid(&o->do_lu)), rc);
228                 RETURN(rc);
229         }
230
231         llh_hdr = &handle->lgh_hdr->llh_hdr;
232         if (LLOG_REC_HDR_NEEDS_SWABBING(llh_hdr))
233                 lustre_swab_llog_hdr(handle->lgh_hdr);
234
235         if (llh_hdr->lrh_type != LLOG_HDR_MAGIC) {
236                 CERROR("%s: bad log %s "DFID" header magic: %#x "
237                        "(expected %#x)\n", o->do_lu.lo_dev->ld_obd->obd_name,
238                        handle->lgh_name ? handle->lgh_name : "",
239                        PFID(lu_object_fid(&o->do_lu)),
240                        llh_hdr->lrh_type, LLOG_HDR_MAGIC);
241                 RETURN(-EIO);
242         } else if (llh_hdr->lrh_len != LLOG_CHUNK_SIZE) {
243                 CERROR("%s: incorrectly sized log %s "DFID" header: "
244                        "%#x (expected %#x)\n"
245                        "you may need to re-run lconf --write_conf.\n",
246                        o->do_lu.lo_dev->ld_obd->obd_name,
247                        handle->lgh_name ? handle->lgh_name : "",
248                        PFID(lu_object_fid(&o->do_lu)),
249                        llh_hdr->lrh_len, LLOG_CHUNK_SIZE);
250                 RETURN(-EIO);
251         }
252
253         handle->lgh_last_idx = handle->lgh_hdr->llh_tail.lrt_index;
254
255         RETURN(0);
256 }
257
258 static int llog_osd_declare_write_rec(const struct lu_env *env,
259                                       struct llog_handle *loghandle,
260                                       struct llog_rec_hdr *rec,
261                                       int idx, struct thandle *th)
262 {
263         struct llog_thread_info *lgi = llog_info(env);
264         struct dt_object        *o;
265         int                      rc;
266
267         ENTRY;
268
269         LASSERT(env);
270         LASSERT(th);
271         LASSERT(loghandle);
272
273         o = loghandle->lgh_obj;
274         LASSERT(o);
275
276         /* each time we update header */
277         rc = dt_declare_record_write(env, o, sizeof(struct llog_log_hdr), 0,
278                                      th);
279         if (rc || idx == 0) /* if error or just header */
280                 RETURN(rc);
281
282         if (dt_object_exists(o)) {
283                 rc = dt_attr_get(env, o, &lgi->lgi_attr, BYPASS_CAPA);
284                 lgi->lgi_off = lgi->lgi_attr.la_size;
285                 LASSERT(ergo(rc == 0, lgi->lgi_attr.la_valid & LA_SIZE));
286                 if (rc)
287                         RETURN(rc);
288         } else {
289                 lgi->lgi_off = 0;
290         }
291
292         /* XXX: implement declared window or multi-chunks approach */
293         rc = dt_declare_record_write(env, o, 32 * 1024, lgi->lgi_off, th);
294
295         RETURN(rc);
296 }
297
298 /* returns negative in on error; 0 if success && reccookie == 0; 1 otherwise */
299 /* appends if idx == -1, otherwise overwrites record idx. */
300 static int llog_osd_write_rec(const struct lu_env *env,
301                               struct llog_handle *loghandle,
302                               struct llog_rec_hdr *rec,
303                               struct llog_cookie *reccookie, int cookiecount,
304                               void *buf, int idx, struct thandle *th)
305 {
306         struct llog_thread_info *lgi = llog_info(env);
307         struct llog_log_hdr     *llh;
308         int                      reclen = rec->lrh_len;
309         int                      index, rc;
310         struct llog_rec_tail    *lrt;
311         struct dt_object        *o;
312         size_t                   left;
313
314         ENTRY;
315
316         LASSERT(env);
317         llh = loghandle->lgh_hdr;
318         LASSERT(llh);
319         o = loghandle->lgh_obj;
320         LASSERT(o);
321         LASSERT(th);
322
323         CDEBUG(D_OTHER, "new record %x to "DFID"\n",
324                rec->lrh_type, PFID(lu_object_fid(&o->do_lu)));
325
326         /* record length should not bigger than LLOG_CHUNK_SIZE */
327         if (buf)
328                 rc = (reclen > LLOG_CHUNK_SIZE - sizeof(struct llog_rec_hdr) -
329                       sizeof(struct llog_rec_tail)) ? -E2BIG : 0;
330         else
331                 rc = (reclen > LLOG_CHUNK_SIZE) ? -E2BIG : 0;
332         if (rc)
333                 RETURN(rc);
334
335         rc = dt_attr_get(env, o, &lgi->lgi_attr, NULL);
336         if (rc)
337                 RETURN(rc);
338
339         if (buf)
340                 /* write_blob adds header and tail to lrh_len. */
341                 reclen = sizeof(*rec) + rec->lrh_len +
342                          sizeof(struct llog_rec_tail);
343
344         if (idx != -1) {
345                 /* no header: only allowed to insert record 1 */
346                 if (idx != 1 && lgi->lgi_attr.la_size == 0)
347                         LBUG();
348
349                 if (idx && llh->llh_size && llh->llh_size != rec->lrh_len)
350                         RETURN(-EINVAL);
351
352                 if (!ext2_test_bit(idx, llh->llh_bitmap))
353                         CERROR("%s: modify unset record %u\n",
354                                o->do_lu.lo_dev->ld_obd->obd_name, idx);
355                 if (idx != rec->lrh_index)
356                         CERROR("%s: index mismatch %d %u\n",
357                                o->do_lu.lo_dev->ld_obd->obd_name, idx,
358                                rec->lrh_index);
359
360                 lgi->lgi_off = 0;
361                 rc = llog_osd_write_blob(env, o, &llh->llh_hdr, NULL,
362                                          &lgi->lgi_off, th);
363                 /* we are done if we only write the header or on error */
364                 if (rc || idx == 0)
365                         RETURN(rc);
366
367                 if (buf) {
368                         /* We assume that caller has set lgh_cur_* */
369                         lgi->lgi_off = loghandle->lgh_cur_offset;
370                         CDEBUG(D_OTHER,
371                                "modify record "LPX64": idx:%d/%u/%d, len:%u "
372                                "offset %llu\n",
373                                loghandle->lgh_id.lgl_oid, idx, rec->lrh_index,
374                                loghandle->lgh_cur_idx, rec->lrh_len,
375                                (long long)(lgi->lgi_off - sizeof(*llh)));
376                         if (rec->lrh_index != loghandle->lgh_cur_idx) {
377                                 CERROR("%s: modify idx mismatch %u/%d\n",
378                                        o->do_lu.lo_dev->ld_obd->obd_name, idx,
379                                        loghandle->lgh_cur_idx);
380                                 RETURN(-EFAULT);
381                         }
382                 } else {
383                         /* Assumes constant lrh_len */
384                         lgi->lgi_off = sizeof(*llh) + (idx - 1) * reclen;
385                 }
386
387                 rc = llog_osd_write_blob(env, o, rec, buf, &lgi->lgi_off, th);
388                 if (rc == 0 && reccookie) {
389                         reccookie->lgc_lgl = loghandle->lgh_id;
390                         reccookie->lgc_index = idx;
391                         rc = 1;
392                 }
393                 RETURN(rc);
394         }
395
396         /* Make sure that records don't cross a chunk boundary, so we can
397          * process them page-at-a-time if needed.  If it will cross a chunk
398          * boundary, write in a fake (but referenced) entry to pad the chunk.
399          *
400          * We know that llog_current_log() will return a loghandle that is
401          * big enough to hold reclen, so all we care about is padding here.
402          */
403         LASSERT(lgi->lgi_attr.la_valid & LA_SIZE);
404         lgi->lgi_off = lgi->lgi_attr.la_size;
405         left = LLOG_CHUNK_SIZE - (lgi->lgi_off & (LLOG_CHUNK_SIZE - 1));
406         /* NOTE: padding is a record, but no bit is set */
407         if (left != 0 && left != reclen &&
408             left < (reclen + LLOG_MIN_REC_SIZE)) {
409                 index = loghandle->lgh_last_idx + 1;
410                 rc = llog_osd_pad(env, o, &lgi->lgi_off, left, index, th);
411                 if (rc)
412                         RETURN(rc);
413                 loghandle->lgh_last_idx++; /*for pad rec*/
414         }
415         /* if it's the last idx in log file, then return -ENOSPC */
416         if (loghandle->lgh_last_idx >= LLOG_BITMAP_SIZE(llh) - 1)
417                 RETURN(-ENOSPC);
418
419         loghandle->lgh_last_idx++;
420         index = loghandle->lgh_last_idx;
421         LASSERT(index < LLOG_BITMAP_SIZE(llh));
422         rec->lrh_index = index;
423         if (buf == NULL) {
424                 lrt = (struct llog_rec_tail *)((char *)rec + rec->lrh_len -
425                                                sizeof(*lrt));
426                 lrt->lrt_len = rec->lrh_len;
427                 lrt->lrt_index = rec->lrh_index;
428         }
429         /* The caller should make sure only 1 process access the lgh_last_idx,
430          * Otherwise it might hit the assert.*/
431         LASSERT(index < LLOG_BITMAP_SIZE(llh));
432         cfs_spin_lock(&loghandle->lgh_hdr_lock);
433         if (ext2_set_bit(index, llh->llh_bitmap)) {
434                 CERROR("%s: index %u already set in log bitmap\n",
435                        o->do_lu.lo_dev->ld_obd->obd_name, index);
436                 cfs_spin_unlock(&loghandle->lgh_hdr_lock);
437                 LBUG(); /* should never happen */
438         }
439         llh->llh_count++;
440         cfs_spin_unlock(&loghandle->lgh_hdr_lock);
441         llh->llh_tail.lrt_index = index;
442
443         lgi->lgi_off = 0;
444         rc = llog_osd_write_blob(env, o, &llh->llh_hdr, NULL, &lgi->lgi_off,
445                                  th);
446         if (rc)
447                 RETURN(rc);
448
449         rc = dt_attr_get(env, o, &lgi->lgi_attr, NULL);
450         if (rc)
451                 RETURN(rc);
452         LASSERT(lgi->lgi_attr.la_valid & LA_SIZE);
453         lgi->lgi_off = lgi->lgi_attr.la_size;
454
455         rc = llog_osd_write_blob(env, o, rec, buf, &lgi->lgi_off, th);
456         if (rc)
457                 RETURN(rc);
458
459         CDEBUG(D_RPCTRACE, "added record "LPX64": idx: %u, %u\n",
460                loghandle->lgh_id.lgl_oid, index, rec->lrh_len);
461         if (rc == 0 && reccookie) {
462                 reccookie->lgc_lgl = loghandle->lgh_id;
463                 reccookie->lgc_index = index;
464                 if ((rec->lrh_type == MDS_UNLINK_REC) ||
465                     (rec->lrh_type == MDS_SETATTR64_REC))
466                         reccookie->lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
467                 else if (rec->lrh_type == OST_SZ_REC)
468                         reccookie->lgc_subsys = LLOG_SIZE_ORIG_CTXT;
469                 else
470                         reccookie->lgc_subsys = -1;
471                 rc = 1;
472         }
473         RETURN(rc);
474 }
475
476 /* We can skip reading at least as many log blocks as the number of
477  * minimum sized log records we are skipping.  If it turns out
478  * that we are not far enough along the log (because the
479  * actual records are larger than minimum size) we just skip
480  * some more records.
481  */
482 static void llog_skip_over(__u64 *off, int curr, int goal)
483 {
484         if (goal <= curr)
485                 return;
486         *off = (*off + (goal - curr - 1) * LLOG_MIN_REC_SIZE) &
487                 ~(LLOG_CHUNK_SIZE - 1);
488 }
489
490 /* sets:
491  *  - cur_offset to the furthest point read in the log file
492  *  - cur_idx to the log index preceeding cur_offset
493  * returns -EIO/-EINVAL on error
494  */
495 static int llog_osd_next_block(const struct lu_env *env,
496                                struct llog_handle *loghandle, int *cur_idx,
497                                int next_idx, __u64 *cur_offset, void *buf,
498                                int len)
499 {
500         struct llog_thread_info *lgi = llog_info(env);
501         struct dt_object        *o;
502         struct dt_device        *dt;
503         int                      rc;
504
505         ENTRY;
506
507         LASSERT(env);
508         LASSERT(lgi);
509
510         if (len == 0 || len & (LLOG_CHUNK_SIZE - 1))
511                 RETURN(-EINVAL);
512
513         CDEBUG(D_OTHER, "looking for log index %u (cur idx %u off "LPU64")\n",
514                next_idx, *cur_idx, *cur_offset);
515
516         LASSERT(loghandle);
517         LASSERT(loghandle->lgh_ctxt);
518
519         o = loghandle->lgh_obj;
520         LASSERT(o);
521         LASSERT(dt_object_exists(o));
522         dt = lu2dt_dev(o->do_lu.lo_dev);
523         LASSERT(dt);
524
525         rc = dt_attr_get(env, o, &lgi->lgi_attr, BYPASS_CAPA);
526         if (rc)
527                 GOTO(out, rc);
528
529         while (*cur_offset < lgi->lgi_attr.la_size) {
530                 struct llog_rec_hdr     *rec, *last_rec;
531                 struct llog_rec_tail    *tail;
532
533                 llog_skip_over(cur_offset, *cur_idx, next_idx);
534
535                 /* read up to next LLOG_CHUNK_SIZE block */
536                 lgi->lgi_buf.lb_len = LLOG_CHUNK_SIZE -
537                                       (*cur_offset & (LLOG_CHUNK_SIZE - 1));
538                 lgi->lgi_buf.lb_buf = buf;
539                 rc = dt_read(env, o, &lgi->lgi_buf, cur_offset);
540                 if (rc < 0) {
541                         CERROR("%s: can't read llog block from log "DFID
542                                " offset "LPU64": rc = %d\n",
543                                o->do_lu.lo_dev->ld_obd->obd_name,
544                                PFID(lu_object_fid(&o->do_lu)), *cur_offset,
545                                rc);
546                         GOTO(out, rc);
547                 }
548
549                 if (rc < len) {
550                         /* signal the end of the valid buffer to
551                          * llog_process */
552                         memset(buf + rc, 0, len - rc);
553                 }
554
555                 if (rc == 0) /* end of file, nothing to do */
556                         GOTO(out, rc);
557
558                 if (rc < sizeof(*tail)) {
559                         CERROR("%s: invalid llog block at log id "LPU64"/%u "
560                                "offset "LPU64"\n",
561                                o->do_lu.lo_dev->ld_obd->obd_name,
562                                loghandle->lgh_id.lgl_oid,
563                                loghandle->lgh_id.lgl_ogen, *cur_offset);
564                         GOTO(out, rc = -EINVAL);
565                 }
566
567                 rec = buf;
568                 if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
569                         lustre_swab_llog_rec(rec);
570
571                 tail = (struct llog_rec_tail *)((char *)buf + rc -
572                                                 sizeof(struct llog_rec_tail));
573                 /* get the last record in block */
574                 last_rec = (struct llog_rec_hdr *)((char *)buf + rc -
575                                                    le32_to_cpu(tail->lrt_len));
576
577                 if (LLOG_REC_HDR_NEEDS_SWABBING(last_rec))
578                         lustre_swab_llog_rec(last_rec);
579                 LASSERT(last_rec->lrh_index == tail->lrt_index);
580
581                 *cur_idx = tail->lrt_index;
582
583                 /* this shouldn't happen */
584                 if (tail->lrt_index == 0) {
585                         CERROR("%s: invalid llog tail at log id "LPU64"/%u "
586                                "offset "LPU64"\n",
587                                o->do_lu.lo_dev->ld_obd->obd_name,
588                                loghandle->lgh_id.lgl_oid,
589                                loghandle->lgh_id.lgl_ogen, *cur_offset);
590                         GOTO(out, rc = -EINVAL);
591                 }
592                 if (tail->lrt_index < next_idx)
593                         continue;
594
595                 /* sanity check that the start of the new buffer is no farther
596                  * than the record that we wanted.  This shouldn't happen. */
597                 if (rec->lrh_index > next_idx) {
598                         CERROR("%s: missed desired record? %u > %u\n",
599                                o->do_lu.lo_dev->ld_obd->obd_name,
600                                rec->lrh_index, next_idx);
601                         GOTO(out, rc = -ENOENT);
602                 }
603                 GOTO(out, rc = 0);
604         }
605         GOTO(out, rc = -EIO);
606 out:
607         return rc;
608 }
609
610 static int llog_osd_prev_block(const struct lu_env *env,
611                                struct llog_handle *loghandle,
612                                int prev_idx, void *buf, int len)
613 {
614         struct llog_thread_info *lgi = llog_info(env);
615         struct dt_object        *o;
616         struct dt_device        *dt;
617         loff_t                   cur_offset;
618         int                      rc;
619
620         ENTRY;
621
622         if (len == 0 || len & (LLOG_CHUNK_SIZE - 1))
623                 RETURN(-EINVAL);
624
625         CDEBUG(D_OTHER, "looking for log index %u\n", prev_idx);
626
627         LASSERT(loghandle);
628         LASSERT(loghandle->lgh_ctxt);
629
630         o = loghandle->lgh_obj;
631         LASSERT(o);
632         LASSERT(dt_object_exists(o));
633         dt = lu2dt_dev(o->do_lu.lo_dev);
634         LASSERT(dt);
635
636         cur_offset = LLOG_CHUNK_SIZE;
637         llog_skip_over(&cur_offset, 0, prev_idx);
638
639         rc = dt_attr_get(env, o, &lgi->lgi_attr, BYPASS_CAPA);
640         if (rc)
641                 GOTO(out, rc);
642
643         while (cur_offset < lgi->lgi_attr.la_size) {
644                 struct llog_rec_hdr     *rec, *last_rec;
645                 struct llog_rec_tail    *tail;
646
647                 lgi->lgi_buf.lb_len = len;
648                 lgi->lgi_buf.lb_buf = buf;
649                 rc = dt_read(env, o, &lgi->lgi_buf, &cur_offset);
650                 if (rc < 0) {
651                         CERROR("%s: can't read llog block from log "DFID
652                                " offset "LPU64": rc = %d\n",
653                                o->do_lu.lo_dev->ld_obd->obd_name,
654                                PFID(lu_object_fid(&o->do_lu)), cur_offset, rc);
655                         GOTO(out, rc);
656                 }
657
658                 if (rc == 0) /* end of file, nothing to do */
659                         GOTO(out, rc);
660
661                 if (rc < sizeof(*tail)) {
662                         CERROR("%s: invalid llog block at log id "LPU64"/%u "
663                                "offset "LPU64"\n",
664                                o->do_lu.lo_dev->ld_obd->obd_name,
665                                loghandle->lgh_id.lgl_oid,
666                                loghandle->lgh_id.lgl_ogen, cur_offset);
667                         GOTO(out, rc = -EINVAL);
668                 }
669
670                 rec = buf;
671                 if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
672                         lustre_swab_llog_rec(rec);
673
674                 tail = (struct llog_rec_tail *)((char *)buf + rc -
675                                                 sizeof(struct llog_rec_tail));
676                 /* get the last record in block */
677                 last_rec = (struct llog_rec_hdr *)((char *)buf + rc -
678                                                    le32_to_cpu(tail->lrt_len));
679
680                 if (LLOG_REC_HDR_NEEDS_SWABBING(last_rec))
681                         lustre_swab_llog_rec(last_rec);
682                 LASSERT(last_rec->lrh_index == tail->lrt_index);
683
684                 /* this shouldn't happen */
685                 if (tail->lrt_index == 0) {
686                         CERROR("%s: invalid llog tail at log id "LPU64"/%u "
687                                "offset "LPU64"\n",
688                                o->do_lu.lo_dev->ld_obd->obd_name,
689                                loghandle->lgh_id.lgl_oid,
690                                loghandle->lgh_id.lgl_ogen, cur_offset);
691                         GOTO(out, rc = -EINVAL);
692                 }
693                 if (tail->lrt_index < prev_idx)
694                         continue;
695
696                 /* sanity check that the start of the new buffer is no farther
697                  * than the record that we wanted.  This shouldn't happen. */
698                 if (rec->lrh_index > prev_idx) {
699                         CERROR("%s: missed desired record? %u > %u\n",
700                                o->do_lu.lo_dev->ld_obd->obd_name,
701                                rec->lrh_index, prev_idx);
702                         GOTO(out, rc = -ENOENT);
703                 }
704                 GOTO(out, rc = 0);
705         }
706         GOTO(out, rc = -EIO);
707 out:
708         return rc;
709 }
710
711 struct dt_object *llog_osd_dir_get(const struct lu_env *env,
712                                    struct llog_ctxt *ctxt)
713 {
714         struct dt_device        *dt;
715         struct dt_thread_info   *dti = dt_info(env);
716         struct dt_object        *dir;
717         int                      rc;
718
719         dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
720         if (ctxt->loc_dir == NULL) {
721                 rc = dt_root_get(env, dt, &dti->dti_fid);
722                 if (rc)
723                         return ERR_PTR(rc);
724                 dir = dt_locate(env, dt, &dti->dti_fid);
725         } else {
726                 lu_object_get(&ctxt->loc_dir->do_lu);
727                 dir = ctxt->loc_dir;
728         }
729
730         return dir;
731 }
732
733 static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle,
734                          struct llog_logid *logid, char *name,
735                          enum llog_open_param open_param)
736 {
737         struct llog_thread_info         *lgi = llog_info(env);
738         struct llog_ctxt                *ctxt = handle->lgh_ctxt;
739         struct dt_object                *o;
740         struct dt_device                *dt;
741         struct ls_device                *ls;
742         struct local_oid_storage        *los;
743         int                              rc = 0;
744
745         ENTRY;
746
747         LASSERT(env);
748         LASSERT(ctxt);
749         LASSERT(ctxt->loc_exp);
750         LASSERT(ctxt->loc_exp->exp_obd);
751         dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
752         LASSERT(dt);
753
754         ls = ls_device_get(dt);
755         if (IS_ERR(ls))
756                 RETURN(PTR_ERR(ls));
757
758         cfs_mutex_lock(&ls->ls_los_mutex);
759         los = dt_los_find(ls, FID_SEQ_LLOG);
760         cfs_mutex_unlock(&ls->ls_los_mutex);
761         LASSERT(los);
762         ls_device_put(env, ls);
763
764         LASSERT(handle);
765
766         if (logid != NULL) {
767                 logid_to_fid(logid, &lgi->lgi_fid);
768         } else if (name) {
769                 struct dt_object *llog_dir;
770
771                 llog_dir = llog_osd_dir_get(env, ctxt);
772                 if (IS_ERR(llog_dir))
773                         GOTO(out, rc = PTR_ERR(llog_dir));
774                 dt_read_lock(env, llog_dir, 0);
775                 rc = dt_lookup_dir(env, llog_dir, name, &lgi->lgi_fid);
776                 dt_read_unlock(env, llog_dir);
777                 lu_object_put(env, &llog_dir->do_lu);
778                 if (rc == -ENOENT && open_param == LLOG_OPEN_NEW) {
779                         /* generate fid for new llog */
780                         rc = local_object_fid_generate(env, los,
781                                                        &lgi->lgi_fid);
782                 }
783                 if (rc < 0)
784                         GOTO(out, rc);
785                 OBD_ALLOC(handle->lgh_name, strlen(name) + 1);
786                 if (handle->lgh_name)
787                         strcpy(handle->lgh_name, name);
788                 else
789                         GOTO(out, rc = -ENOMEM);
790         } else {
791                 LASSERTF(open_param & LLOG_OPEN_NEW, "%#x\n", open_param);
792                 /* generate fid for new llog */
793                 rc = local_object_fid_generate(env, los, &lgi->lgi_fid);
794                 if (rc < 0)
795                         GOTO(out, rc);
796         }
797
798         o = ls_locate(env, ls, &lgi->lgi_fid);
799         if (IS_ERR(o))
800                 GOTO(out_name, rc = PTR_ERR(o));
801
802         /* No new llog is expected but doesn't exist */
803         if (open_param != LLOG_OPEN_NEW && !dt_object_exists(o))
804                 GOTO(out_put, rc = -ENOENT);
805
806         fid_to_logid(&lgi->lgi_fid, &handle->lgh_id);
807         handle->lgh_obj = o;
808         handle->private_data = los;
809         LASSERT(handle->lgh_ctxt);
810
811         RETURN(rc);
812
813 out_put:
814         lu_object_put(env, &o->do_lu);
815 out_name:
816         if (handle->lgh_name != NULL)
817                 OBD_FREE(handle->lgh_name, strlen(name) + 1);
818 out:
819         dt_los_put(los);
820         RETURN(rc);
821 }
822
823 static int llog_osd_exist(struct llog_handle *handle)
824 {
825         LASSERT(handle->lgh_obj);
826         return (dt_object_exists(handle->lgh_obj) &&
827                 !lu_object_is_dying(handle->lgh_obj->do_lu.lo_header));
828 }
829
830 static int llog_osd_declare_create(const struct lu_env *env,
831                                    struct llog_handle *res, struct thandle *th)
832 {
833         struct llog_thread_info         *lgi = llog_info(env);
834         struct local_oid_storage        *los;
835         struct dt_object                *o;
836         int                              rc;
837
838         ENTRY;
839
840         LASSERT(res->lgh_obj);
841         LASSERT(th);
842
843         /* object can be created by another thread */
844         o = res->lgh_obj;
845         if (dt_object_exists(o))
846                 RETURN(0);
847
848         los = res->private_data;
849         LASSERT(los);
850
851         rc = llog_osd_declare_new_object(env, los, o, th);
852         if (rc)
853                 RETURN(rc);
854
855         rc = dt_declare_record_write(env, o, LLOG_CHUNK_SIZE, 0, th);
856         if (rc)
857                 RETURN(rc);
858
859         if (res->lgh_name) {
860                 struct dt_object *llog_dir;
861
862                 llog_dir = llog_osd_dir_get(env, res->lgh_ctxt);
863                 if (IS_ERR(llog_dir))
864                         RETURN(PTR_ERR(llog_dir));
865                 dt_declare_ref_add(env, o, th);
866                 logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
867                 rc = dt_declare_insert(env, llog_dir,
868                                        (struct dt_rec *)&lgi->lgi_fid,
869                                        (struct dt_key *)res->lgh_name, th);
870                 lu_object_put(env, &llog_dir->do_lu);
871                 if (rc)
872                         CERROR("%s: can't declare named llog %s: rc = %d\n",
873                                o->do_lu.lo_dev->ld_obd->obd_name,
874                                res->lgh_name, rc);
875         }
876         RETURN(rc);
877 }
878
879 /* This is a callback from the llog_* functions.
880  * Assumes caller has already pushed us into the kernel context. */
881 static int llog_osd_create(const struct lu_env *env, struct llog_handle *res,
882                            struct thandle *th)
883 {
884         struct llog_thread_info *lgi = llog_info(env);
885         struct local_oid_storage *los;
886         struct dt_object        *o;
887         int                      rc = 0;
888
889         ENTRY;
890
891         LASSERT(env);
892         o = res->lgh_obj;
893         LASSERT(o);
894
895         /* llog can be already created */
896         if (dt_object_exists(o))
897                 RETURN(-EEXIST);
898
899         los = res->private_data;
900         LASSERT(los);
901
902         dt_write_lock(env, o, 0);
903         if (!dt_object_exists(o))
904                 rc = llog_osd_create_new_object(env, los, o, th);
905         else
906                 rc = -EEXIST;
907         if (res->lgh_name)
908                 dt_ref_add(env, o, th);
909         dt_write_unlock(env, o);
910         if (rc)
911                 RETURN(rc);
912
913         if (res->lgh_name) {
914                 struct dt_object *llog_dir;
915
916                 llog_dir = llog_osd_dir_get(env, res->lgh_ctxt);
917                 if (IS_ERR(llog_dir))
918                         RETURN(PTR_ERR(llog_dir));
919
920                 logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
921                 dt_read_lock(env, llog_dir, 0);
922                 rc = dt_insert(env, llog_dir,
923                                (struct dt_rec *)&lgi->lgi_fid,
924                                (struct dt_key *)res->lgh_name,
925                                th, BYPASS_CAPA, 1);
926                 dt_read_unlock(env, llog_dir);
927                 lu_object_put(env, &llog_dir->do_lu);
928                 if (rc)
929                         CERROR("%s: can't create named llog %s: rc = %d\n",
930                                o->do_lu.lo_dev->ld_obd->obd_name,
931                                res->lgh_name, rc);
932         }
933         RETURN(rc);
934 }
935
936 static int llog_osd_close(const struct lu_env *env, struct llog_handle *handle)
937 {
938         struct local_oid_storage        *los;
939         int                              rc = 0;
940
941         ENTRY;
942
943         LASSERT(handle->lgh_obj);
944
945         lu_object_put(env, &handle->lgh_obj->do_lu);
946
947         los = handle->private_data;
948         LASSERT(los);
949         dt_los_put(los);
950
951         if (handle->lgh_name)
952                 OBD_FREE(handle->lgh_name, strlen(handle->lgh_name) + 1);
953
954         RETURN(rc);
955 }
956
957 static int llog_osd_destroy(const struct lu_env *env,
958                             struct llog_handle *loghandle)
959 {
960         struct llog_ctxt        *ctxt;
961         struct dt_object        *o, *llog_dir = NULL;
962         struct dt_device        *d;
963         struct thandle          *th;
964         char                    *name = NULL;
965         int                      rc;
966
967         ENTRY;
968
969         ctxt = loghandle->lgh_ctxt;
970         LASSERT(ctxt);
971
972         o = loghandle->lgh_obj;
973         LASSERT(o);
974
975         d = lu2dt_dev(o->do_lu.lo_dev);
976         LASSERT(d);
977         LASSERT(d == ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt);
978
979         th = dt_trans_create(env, d);
980         if (IS_ERR(th))
981                 RETURN(PTR_ERR(th));
982
983         if (loghandle->lgh_name) {
984                 llog_dir = llog_osd_dir_get(env, ctxt);
985                 if (IS_ERR(llog_dir))
986                         GOTO(out_trans, rc = PTR_ERR(llog_dir));
987
988                 dt_declare_ref_del(env, o, th);
989                 name = loghandle->lgh_name;
990                 rc = dt_declare_delete(env, llog_dir,
991                                        (struct dt_key *)name, th);
992                 if (rc)
993                         GOTO(out_trans, rc);
994         }
995
996         dt_declare_ref_del(env, o, th);
997
998         rc = dt_declare_destroy(env, o, th);
999         if (rc)
1000                 GOTO(out_trans, rc);
1001
1002         rc = dt_trans_start_local(env, d, th);
1003         if (rc)
1004                 GOTO(out_trans, rc);
1005
1006         dt_write_lock(env, o, 0);
1007         if (dt_object_exists(o)) {
1008                 if (name) {
1009                         dt_ref_del(env, o, th);
1010                         dt_read_lock(env, llog_dir, 0);
1011                         rc = dt_delete(env, llog_dir,
1012                                        (struct dt_key *) name,
1013                                        th, BYPASS_CAPA);
1014                         dt_read_unlock(env, llog_dir);
1015                         if (rc) {
1016                                 CERROR("%s: can't remove llog %s: rc = %d\n",
1017                                        o->do_lu.lo_dev->ld_obd->obd_name,
1018                                        name, rc);
1019                                 GOTO(out_unlock, rc);
1020                         }
1021                 }
1022                 dt_ref_del(env, o, th);
1023                 rc = dt_destroy(env, o, th);
1024                 if (rc)
1025                         GOTO(out_unlock, rc);
1026         }
1027 out_unlock:
1028         dt_write_unlock(env, o);
1029 out_trans:
1030         dt_trans_stop(env, d, th);
1031         if (llog_dir != NULL)
1032                 lu_object_put(env, &llog_dir->do_lu);
1033         RETURN(rc);
1034 }
1035
1036 static int llog_osd_setup(const struct lu_env *env, struct obd_device *obd,
1037                           struct obd_llog_group *olg, int ctxt_idx,
1038                           struct obd_device *disk_obd)
1039 {
1040         struct local_oid_storage        *los;
1041         struct llog_thread_info         *lgi = llog_info(env);
1042         struct llog_ctxt                *ctxt;
1043         int                              rc = 0;
1044
1045         ENTRY;
1046
1047         LASSERT(obd);
1048         LASSERT(olg->olg_ctxts[ctxt_idx]);
1049
1050         ctxt = llog_ctxt_get(olg->olg_ctxts[ctxt_idx]);
1051         LASSERT(ctxt);
1052
1053         /* initialize data allowing to generate new fids,
1054          * literally we need a sequece */
1055         lgi->lgi_fid.f_seq = FID_SEQ_LLOG;
1056         lgi->lgi_fid.f_oid = 1;
1057         lgi->lgi_fid.f_ver = 0;
1058         rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt,
1059                                     &lgi->lgi_fid, &los);
1060         llog_ctxt_put(ctxt);
1061         return rc;
1062 }
1063
1064 static int llog_osd_cleanup(const struct lu_env *env, struct llog_ctxt *ctxt)
1065 {
1066         struct dt_device                *dt;
1067         struct ls_device                *ls;
1068         struct local_oid_storage        *los;
1069
1070         LASSERT(ctxt->loc_exp->exp_obd);
1071         dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
1072         ls = ls_device_get(dt);
1073         if (IS_ERR(ls))
1074                 RETURN(PTR_ERR(ls));
1075
1076         cfs_mutex_lock(&ls->ls_los_mutex);
1077         los = dt_los_find(ls, FID_SEQ_LLOG);
1078         cfs_mutex_unlock(&ls->ls_los_mutex);
1079         if (los != NULL) {
1080                 dt_los_put(los);
1081                 local_oid_storage_fini(env, los);
1082         }
1083         ls_device_put(env, ls);
1084         return 0;
1085 }
1086
1087 struct llog_operations llog_osd_ops = {
1088         .lop_next_block         = llog_osd_next_block,
1089         .lop_prev_block         = llog_osd_prev_block,
1090         .lop_read_header        = llog_osd_read_header,
1091         .lop_destroy            = llog_osd_destroy,
1092         .lop_setup              = llog_osd_setup,
1093         .lop_cleanup            = llog_osd_cleanup,
1094         .lop_open               = llog_osd_open,
1095         .lop_exist              = llog_osd_exist,
1096         .lop_declare_create     = llog_osd_declare_create,
1097         .lop_create             = llog_osd_create,
1098         .lop_declare_write_rec  = llog_osd_declare_write_rec,
1099         .lop_write_rec          = llog_osd_write_rec,
1100         .lop_close              = llog_osd_close,
1101 };
1102 EXPORT_SYMBOL(llog_osd_ops);
1103
1104 /* reads the catalog list */
1105 int llog_osd_get_cat_list(const struct lu_env *env, struct dt_device *d,
1106                           int idx, int count, struct llog_catid *idarray)
1107 {
1108         struct llog_thread_info *lgi = llog_info(env);
1109         struct dt_object        *o = NULL;
1110         struct thandle          *th;
1111         int                      rc, size;
1112
1113         ENTRY;
1114
1115         LASSERT(d);
1116
1117         size = sizeof(*idarray) * count;
1118         lgi->lgi_off = idx *  sizeof(*idarray);
1119
1120         lu_local_obj_fid(&lgi->lgi_fid, LLOG_CATALOGS_OID);
1121         o = dt_locate(env, d, &lgi->lgi_fid);
1122         if (IS_ERR(o))
1123                 RETURN(PTR_ERR(o));
1124
1125         if (!dt_object_exists(o)) {
1126                 th = dt_trans_create(env, d);
1127                 if (IS_ERR(th))
1128                         GOTO(out, rc = PTR_ERR(th));
1129
1130                 lgi->lgi_attr.la_valid = LA_MODE;
1131                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
1132                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
1133
1134                 rc = dt_declare_create(env, o, &lgi->lgi_attr, NULL,
1135                                        &lgi->lgi_dof, th);
1136                 if (rc)
1137                         GOTO(out_trans, rc);
1138
1139                 rc = dt_trans_start_local(env, d, th);
1140                 if (rc)
1141                         GOTO(out_trans, rc);
1142
1143                 dt_write_lock(env, o, 0);
1144                 if (!dt_object_exists(o))
1145                         rc = dt_create(env, o, &lgi->lgi_attr, NULL,
1146                                        &lgi->lgi_dof, th);
1147                 dt_write_unlock(env, o);
1148 out_trans:
1149                 dt_trans_stop(env, d, th);
1150                 if (rc)
1151                         GOTO(out, rc);
1152         }
1153
1154         rc = dt_attr_get(env, o, &lgi->lgi_attr, BYPASS_CAPA);
1155         if (rc)
1156                 GOTO(out, rc);
1157
1158         if (!S_ISREG(lgi->lgi_attr.la_mode)) {
1159                 CERROR("%s: CATALOGS is not a regular file!: mode = %o\n",
1160                        o->do_lu.lo_dev->ld_obd->obd_name,
1161                        lgi->lgi_attr.la_mode);
1162                 GOTO(out, rc = -ENOENT);
1163         }
1164
1165         CDEBUG(D_CONFIG, "cat list: disk size=%d, read=%d\n",
1166                (int)lgi->lgi_attr.la_size, size);
1167
1168         /* return just number of llogs */
1169         if (idarray == NULL) {
1170                 rc = lgi->lgi_attr.la_size / sizeof(*idarray);
1171                 GOTO(out, rc);
1172         }
1173
1174         /* read for new ost index or for empty file */
1175         memset(idarray, 0, size);
1176         if (lgi->lgi_attr.la_size < lgi->lgi_off + size)
1177                 GOTO(out, rc = 0);
1178         if (lgi->lgi_attr.la_size < lgi->lgi_off + size)
1179                 size = lgi->lgi_attr.la_size - lgi->lgi_off;
1180
1181         lgi->lgi_buf.lb_buf = idarray;
1182         lgi->lgi_buf.lb_len = size;
1183         rc = dt_record_read(env, o, &lgi->lgi_buf, &lgi->lgi_off);
1184         if (rc) {
1185                 CERROR("%s: error reading CATALOGS: rc = %d\n",
1186                        o->do_lu.lo_dev->ld_obd->obd_name,  rc);
1187                 GOTO(out, rc);
1188         }
1189
1190         EXIT;
1191 out:
1192         lu_object_put(env, &o->do_lu);
1193         RETURN(rc);
1194 }
1195 EXPORT_SYMBOL(llog_osd_get_cat_list);
1196
1197 /* writes the cat list */
1198 int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d,
1199                           int idx, int count, struct llog_catid *idarray)
1200 {
1201         struct llog_thread_info *lgi = llog_info(env);
1202         struct dt_object        *o = NULL;
1203         struct thandle          *th;
1204         int                      rc, size;
1205
1206         if (!count)
1207                 RETURN(0);
1208
1209         LASSERT(d);
1210
1211         size = sizeof(*idarray) * count;
1212         lgi->lgi_off = idx * sizeof(*idarray);
1213
1214         lu_local_obj_fid(&lgi->lgi_fid, LLOG_CATALOGS_OID);
1215         o = dt_locate(env, d, &lgi->lgi_fid);
1216         if (IS_ERR(o))
1217                 RETURN(PTR_ERR(o));
1218
1219         if (!dt_object_exists(o))
1220                 GOTO(out, rc = -ENOENT);
1221
1222         rc = dt_attr_get(env, o, &lgi->lgi_attr, BYPASS_CAPA);
1223         if (rc)
1224                 GOTO(out, rc);
1225
1226         if (!S_ISREG(lgi->lgi_attr.la_mode)) {
1227                 CERROR("%s: CATALOGS is not a regular file!: mode = %o\n",
1228                        o->do_lu.lo_dev->ld_obd->obd_name,
1229                        lgi->lgi_attr.la_mode);
1230                 GOTO(out, rc = -ENOENT);
1231         }
1232
1233         th = dt_trans_create(env, d);
1234         if (IS_ERR(th))
1235                 GOTO(out, rc = PTR_ERR(th));
1236
1237         rc = dt_declare_record_write(env, o, size, lgi->lgi_off, th);
1238         if (rc)
1239                 GOTO(out, rc);
1240
1241         rc = dt_trans_start_local(env, d, th);
1242         if (rc)
1243                 GOTO(out_trans, rc);
1244
1245         lgi->lgi_buf.lb_buf = idarray;
1246         lgi->lgi_buf.lb_len = size;
1247         rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
1248         if (rc)
1249                 CDEBUG(D_INODE, "error writeing CATALOGS: rc = %d\n", rc);
1250 out_trans:
1251         dt_trans_stop(env, d, th);
1252 out:
1253         lu_object_put(env, &o->do_lu);
1254         RETURN(rc);
1255 }
1256 EXPORT_SYMBOL(llog_osd_put_cat_list);