Whamcloud - gitweb
4ebe3a80db9edae250cf009263d77c20bed3214f
[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 static int llog_osd_open(const struct lu_env *env, struct llog_handle *handle,
712                          struct llog_logid *logid, char *name,
713                          enum llog_open_param open_param)
714 {
715         struct llog_thread_info         *lgi = llog_info(env);
716         struct llog_ctxt                *ctxt = handle->lgh_ctxt;
717         struct dt_object                *o;
718         struct dt_device                *dt;
719         struct ls_device                *ls;
720         struct local_oid_storage        *los;
721         int                              rc = 0;
722
723         ENTRY;
724
725         LASSERT(env);
726         LASSERT(ctxt);
727         LASSERT(ctxt->loc_exp);
728         LASSERT(ctxt->loc_exp->exp_obd);
729         dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
730         LASSERT(dt);
731
732         ls = ls_device_get(dt);
733         if (IS_ERR(ls))
734                 RETURN(PTR_ERR(ls));
735
736         cfs_mutex_lock(&ls->ls_los_mutex);
737         los = dt_los_find(ls, FID_SEQ_LLOG);
738         cfs_mutex_unlock(&ls->ls_los_mutex);
739         LASSERT(los);
740         ls_device_put(env, ls);
741
742         LASSERT(handle);
743
744         if (logid != NULL) {
745                 logid_to_fid(logid, &lgi->lgi_fid);
746         } else if (name) {
747                 LASSERT(ctxt->loc_dir);
748                 dt_read_lock(env, ctxt->loc_dir, 0);
749                 rc = dt_lookup_dir(env, ctxt->loc_dir, name, &lgi->lgi_fid);
750                 dt_read_unlock(env, ctxt->loc_dir);
751                 if (rc == -ENOENT && open_param == LLOG_OPEN_NEW) {
752                         /* generate fid for new llog */
753                         rc = local_object_fid_generate(env, los,
754                                                        &lgi->lgi_fid);
755                 }
756                 if (rc < 0)
757                         GOTO(out, rc);
758                 OBD_ALLOC(handle->lgh_name, strlen(name) + 1);
759                 if (handle->lgh_name)
760                         strcpy(handle->lgh_name, name);
761                 else
762                         GOTO(out, rc = -ENOMEM);
763         } else {
764                 LASSERTF(open_param & LLOG_OPEN_NEW, "%#x\n", open_param);
765                 /* generate fid for new llog */
766                 rc = local_object_fid_generate(env, los, &lgi->lgi_fid);
767                 if (rc < 0)
768                         GOTO(out, rc);
769         }
770
771         o = ls_locate(env, ls, &lgi->lgi_fid);
772         if (IS_ERR(o))
773                 GOTO(out_name, rc = PTR_ERR(o));
774
775         /* No new llog is expected but doesn't exist */
776         if (open_param != LLOG_OPEN_NEW && !dt_object_exists(o))
777                 GOTO(out_put, rc = -ENOENT);
778
779         fid_to_logid(&lgi->lgi_fid, &handle->lgh_id);
780         handle->lgh_obj = o;
781         handle->private_data = los;
782         LASSERT(handle->lgh_ctxt);
783
784         RETURN(rc);
785
786 out_put:
787         lu_object_put(env, &o->do_lu);
788 out_name:
789         if (handle->lgh_name != NULL)
790                 OBD_FREE(handle->lgh_name, strlen(name) + 1);
791 out:
792         dt_los_put(los);
793         RETURN(rc);
794 }
795
796 static int llog_osd_exist(struct llog_handle *handle)
797 {
798         LASSERT(handle->lgh_obj);
799         return (dt_object_exists(handle->lgh_obj) &&
800                 !lu_object_is_dying(handle->lgh_obj->do_lu.lo_header));
801 }
802
803 static int llog_osd_declare_create(const struct lu_env *env,
804                                    struct llog_handle *res, struct thandle *th)
805 {
806         struct llog_thread_info         *lgi = llog_info(env);
807         struct local_oid_storage        *los;
808         struct dt_object                *o;
809         int                              rc;
810
811         ENTRY;
812
813         LASSERT(res->lgh_obj);
814         LASSERT(th);
815
816         /* object can be created by another thread */
817         o = res->lgh_obj;
818         if (dt_object_exists(o))
819                 RETURN(0);
820
821         los = res->private_data;
822         LASSERT(los);
823
824         rc = llog_osd_declare_new_object(env, los, o, th);
825         if (rc)
826                 RETURN(rc);
827
828         rc = dt_declare_record_write(env, o, LLOG_CHUNK_SIZE, 0, th);
829         if (rc)
830                 RETURN(rc);
831
832         if (res->lgh_name) {
833                 LASSERT(res->lgh_ctxt->loc_dir);
834                 dt_declare_ref_add(env, o, th);
835                 logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
836                 rc = dt_declare_insert(env, res->lgh_ctxt->loc_dir,
837                                        (struct dt_rec *)&lgi->lgi_fid,
838                                        (struct dt_key *)res->lgh_name, th);
839                 if (rc)
840                         CERROR("%s: can't declare named llog %s: rc = %d\n",
841                                o->do_lu.lo_dev->ld_obd->obd_name,
842                                res->lgh_name, rc);
843         }
844         RETURN(rc);
845 }
846
847 /* This is a callback from the llog_* functions.
848  * Assumes caller has already pushed us into the kernel context. */
849 static int llog_osd_create(const struct lu_env *env, struct llog_handle *res,
850                            struct thandle *th)
851 {
852         struct llog_thread_info *lgi = llog_info(env);
853         struct local_oid_storage *los;
854         struct dt_object        *o;
855         int                      rc = 0;
856
857         ENTRY;
858
859         LASSERT(env);
860         o = res->lgh_obj;
861         LASSERT(o);
862
863         /* llog can be already created */
864         if (dt_object_exists(o))
865                 RETURN(-EEXIST);
866
867         los = res->private_data;
868         LASSERT(los);
869
870         dt_write_lock(env, o, 0);
871         if (!dt_object_exists(o))
872                 rc = llog_osd_create_new_object(env, los, o, th);
873         else
874                 rc = -EEXIST;
875         if (res->lgh_name)
876                 dt_ref_add(env, o, th);
877         dt_write_unlock(env, o);
878         if (rc)
879                 RETURN(rc);
880
881         if (res->lgh_name) {
882                 LASSERT(res->lgh_ctxt->loc_dir);
883                 logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
884                 dt_read_lock(env, res->lgh_ctxt->loc_dir, 0);
885                 rc = dt_insert(env, res->lgh_ctxt->loc_dir,
886                                (struct dt_rec *)&lgi->lgi_fid,
887                                (struct dt_key *)res->lgh_name,
888                                th, BYPASS_CAPA, 1);
889                 dt_read_unlock(env, res->lgh_ctxt->loc_dir);
890                 if (rc)
891                         CERROR("%s: can't create named llog %s: rc = %d\n",
892                                o->do_lu.lo_dev->ld_obd->obd_name,
893                                res->lgh_name, rc);
894         }
895         RETURN(rc);
896 }
897
898 static int llog_osd_close(const struct lu_env *env, struct llog_handle *handle)
899 {
900         struct local_oid_storage        *los;
901         int                              rc = 0;
902
903         ENTRY;
904
905         LASSERT(handle->lgh_obj);
906
907         lu_object_put(env, &handle->lgh_obj->do_lu);
908
909         los = handle->private_data;
910         LASSERT(los);
911         dt_los_put(los);
912
913         if (handle->lgh_name)
914                 OBD_FREE(handle->lgh_name, strlen(handle->lgh_name) + 1);
915
916         RETURN(rc);
917 }
918
919 static int llog_osd_destroy(const struct lu_env *env,
920                             struct llog_handle *loghandle)
921 {
922         struct llog_ctxt        *ctxt;
923         struct dt_object        *o;
924         struct dt_device        *d;
925         struct thandle          *th;
926         char                    *name = NULL;
927         int                      rc;
928
929         ENTRY;
930
931         ctxt = loghandle->lgh_ctxt;
932         LASSERT(ctxt);
933
934         o = loghandle->lgh_obj;
935         LASSERT(o);
936
937         d = lu2dt_dev(o->do_lu.lo_dev);
938         LASSERT(d);
939         LASSERT(d == ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt);
940
941         th = dt_trans_create(env, d);
942         if (IS_ERR(th))
943                 RETURN(PTR_ERR(th));
944
945         if (loghandle->lgh_name) {
946                 LASSERT(ctxt->loc_dir);
947                 dt_declare_ref_del(env, o, th);
948                 name = loghandle->lgh_name;
949                 rc = dt_declare_delete(env, ctxt->loc_dir,
950                                        (struct dt_key *)name, th);
951                 if (rc)
952                         GOTO(out_trans, rc);
953         }
954
955         dt_declare_ref_del(env, o, th);
956
957         rc = dt_declare_destroy(env, o, th);
958         if (rc)
959                 GOTO(out_trans, rc);
960
961         rc = dt_trans_start_local(env, d, th);
962         if (rc)
963                 GOTO(out_trans, rc);
964
965         dt_write_lock(env, o, 0);
966         if (dt_object_exists(o)) {
967                 if (name) {
968                         dt_ref_del(env, o, th);
969                         dt_read_lock(env, ctxt->loc_dir, 0);
970                         rc = dt_delete(env, ctxt->loc_dir,
971                                        (struct dt_key *) name,
972                                        th, BYPASS_CAPA);
973                         dt_read_unlock(env, ctxt->loc_dir);
974                         if (rc) {
975                                 CERROR("%s: can't remove llog %s: rc = %d\n",
976                                        o->do_lu.lo_dev->ld_obd->obd_name,
977                                        name, rc);
978                                 GOTO(out_unlock, rc);
979                         }
980                 }
981                 dt_ref_del(env, o, th);
982                 rc = dt_destroy(env, o, th);
983                 if (rc)
984                         GOTO(out_unlock, rc);
985         }
986 out_unlock:
987         dt_write_unlock(env, o);
988 out_trans:
989         dt_trans_stop(env, d, th);
990         RETURN(rc);
991 }
992
993 static int llog_osd_setup(const struct lu_env *env, struct obd_device *obd,
994                           struct obd_llog_group *olg, int ctxt_idx,
995                           struct obd_device *disk_obd)
996 {
997         struct local_oid_storage        *los;
998         struct llog_thread_info         *lgi = llog_info(env);
999         struct llog_ctxt                *ctxt;
1000         int                              rc = 0;
1001
1002         ENTRY;
1003
1004         LASSERT(obd);
1005         LASSERT(olg->olg_ctxts[ctxt_idx]);
1006
1007         ctxt = llog_ctxt_get(olg->olg_ctxts[ctxt_idx]);
1008         LASSERT(ctxt);
1009
1010         /* initialize data allowing to generate new fids,
1011          * literally we need a sequece */
1012         lgi->lgi_fid.f_seq = FID_SEQ_LLOG;
1013         lgi->lgi_fid.f_oid = 1;
1014         lgi->lgi_fid.f_ver = 0;
1015         rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt,
1016                                     &lgi->lgi_fid, &los);
1017         llog_ctxt_put(ctxt);
1018         return rc;
1019 }
1020
1021 static int llog_osd_cleanup(const struct lu_env *env, struct llog_ctxt *ctxt)
1022 {
1023         struct dt_device                *dt;
1024         struct ls_device                *ls;
1025         struct local_oid_storage        *los;
1026
1027         LASSERT(ctxt->loc_exp->exp_obd);
1028         dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
1029         ls = ls_device_get(dt);
1030         if (IS_ERR(ls))
1031                 RETURN(PTR_ERR(ls));
1032
1033         cfs_mutex_lock(&ls->ls_los_mutex);
1034         los = dt_los_find(ls, FID_SEQ_LLOG);
1035         cfs_mutex_unlock(&ls->ls_los_mutex);
1036         if (los != NULL) {
1037                 dt_los_put(los);
1038                 local_oid_storage_fini(env, los);
1039         }
1040         ls_device_put(env, ls);
1041         return 0;
1042 }
1043
1044 struct llog_operations llog_osd_ops = {
1045         .lop_next_block         = llog_osd_next_block,
1046         .lop_prev_block         = llog_osd_prev_block,
1047         .lop_read_header        = llog_osd_read_header,
1048         .lop_destroy            = llog_osd_destroy,
1049         .lop_setup              = llog_osd_setup,
1050         .lop_cleanup            = llog_osd_cleanup,
1051         .lop_open               = llog_osd_open,
1052         .lop_exist              = llog_osd_exist,
1053         .lop_declare_create     = llog_osd_declare_create,
1054         .lop_create             = llog_osd_create,
1055         .lop_declare_write_rec  = llog_osd_declare_write_rec,
1056         .lop_write_rec          = llog_osd_write_rec,
1057         .lop_close              = llog_osd_close,
1058 };
1059 EXPORT_SYMBOL(llog_osd_ops);
1060
1061 /* reads the catalog list */
1062 int llog_osd_get_cat_list(const struct lu_env *env, struct dt_device *d,
1063                           int idx, int count, struct llog_catid *idarray)
1064 {
1065         struct llog_thread_info *lgi = llog_info(env);
1066         struct dt_object        *o = NULL;
1067         struct thandle          *th;
1068         int                      rc, size;
1069
1070         ENTRY;
1071
1072         LASSERT(d);
1073
1074         size = sizeof(*idarray) * count;
1075         lgi->lgi_off = idx *  sizeof(*idarray);
1076
1077         lu_local_obj_fid(&lgi->lgi_fid, LLOG_CATALOGS_OID);
1078         o = dt_locate(env, d, &lgi->lgi_fid);
1079         if (IS_ERR(o))
1080                 RETURN(PTR_ERR(o));
1081
1082         if (!dt_object_exists(o)) {
1083                 th = dt_trans_create(env, d);
1084                 if (IS_ERR(th))
1085                         GOTO(out, rc = PTR_ERR(th));
1086
1087                 lgi->lgi_attr.la_valid = LA_MODE;
1088                 lgi->lgi_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
1089                 lgi->lgi_dof.dof_type = dt_mode_to_dft(S_IFREG);
1090
1091                 rc = dt_declare_create(env, o, &lgi->lgi_attr, NULL,
1092                                        &lgi->lgi_dof, th);
1093                 if (rc)
1094                         GOTO(out_trans, rc);
1095
1096                 rc = dt_trans_start_local(env, d, th);
1097                 if (rc)
1098                         GOTO(out_trans, rc);
1099
1100                 dt_write_lock(env, o, 0);
1101                 if (!dt_object_exists(o))
1102                         rc = dt_create(env, o, &lgi->lgi_attr, NULL,
1103                                        &lgi->lgi_dof, th);
1104                 dt_write_unlock(env, o);
1105 out_trans:
1106                 dt_trans_stop(env, d, th);
1107                 if (rc)
1108                         GOTO(out, rc);
1109         }
1110
1111         rc = dt_attr_get(env, o, &lgi->lgi_attr, BYPASS_CAPA);
1112         if (rc)
1113                 GOTO(out, rc);
1114
1115         if (!S_ISREG(lgi->lgi_attr.la_mode)) {
1116                 CERROR("%s: CATALOGS is not a regular file!: mode = %o\n",
1117                        o->do_lu.lo_dev->ld_obd->obd_name,
1118                        lgi->lgi_attr.la_mode);
1119                 GOTO(out, rc = -ENOENT);
1120         }
1121
1122         CDEBUG(D_CONFIG, "cat list: disk size=%d, read=%d\n",
1123                (int)lgi->lgi_attr.la_size, size);
1124
1125         /* return just number of llogs */
1126         if (idarray == NULL) {
1127                 rc = lgi->lgi_attr.la_size / sizeof(*idarray);
1128                 GOTO(out, rc);
1129         }
1130
1131         /* read for new ost index or for empty file */
1132         memset(idarray, 0, size);
1133         if (lgi->lgi_attr.la_size < lgi->lgi_off + size)
1134                 GOTO(out, rc = 0);
1135         if (lgi->lgi_attr.la_size < lgi->lgi_off + size)
1136                 size = lgi->lgi_attr.la_size - lgi->lgi_off;
1137
1138         lgi->lgi_buf.lb_buf = idarray;
1139         lgi->lgi_buf.lb_len = size;
1140         rc = dt_record_read(env, o, &lgi->lgi_buf, &lgi->lgi_off);
1141         if (rc) {
1142                 CERROR("%s: error reading CATALOGS: rc = %d\n",
1143                        o->do_lu.lo_dev->ld_obd->obd_name,  rc);
1144                 GOTO(out, rc);
1145         }
1146
1147         EXIT;
1148 out:
1149         lu_object_put(env, &o->do_lu);
1150         RETURN(rc);
1151 }
1152 EXPORT_SYMBOL(llog_osd_get_cat_list);
1153
1154 /* writes the cat list */
1155 int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d,
1156                           int idx, int count, struct llog_catid *idarray)
1157 {
1158         struct llog_thread_info *lgi = llog_info(env);
1159         struct dt_object        *o = NULL;
1160         struct thandle          *th;
1161         int                      rc, size;
1162
1163         if (!count)
1164                 RETURN(0);
1165
1166         LASSERT(d);
1167
1168         size = sizeof(*idarray) * count;
1169         lgi->lgi_off = idx * sizeof(*idarray);
1170
1171         lu_local_obj_fid(&lgi->lgi_fid, LLOG_CATALOGS_OID);
1172         o = dt_locate(env, d, &lgi->lgi_fid);
1173         if (IS_ERR(o))
1174                 RETURN(PTR_ERR(o));
1175
1176         if (!dt_object_exists(o))
1177                 GOTO(out, rc = -ENOENT);
1178
1179         rc = dt_attr_get(env, o, &lgi->lgi_attr, BYPASS_CAPA);
1180         if (rc)
1181                 GOTO(out, rc);
1182
1183         if (!S_ISREG(lgi->lgi_attr.la_mode)) {
1184                 CERROR("%s: CATALOGS is not a regular file!: mode = %o\n",
1185                        o->do_lu.lo_dev->ld_obd->obd_name,
1186                        lgi->lgi_attr.la_mode);
1187                 GOTO(out, rc = -ENOENT);
1188         }
1189
1190         th = dt_trans_create(env, d);
1191         if (IS_ERR(th))
1192                 GOTO(out, rc = PTR_ERR(th));
1193
1194         rc = dt_declare_record_write(env, o, size, lgi->lgi_off, th);
1195         if (rc)
1196                 GOTO(out, rc);
1197
1198         rc = dt_trans_start_local(env, d, th);
1199         if (rc)
1200                 GOTO(out_trans, rc);
1201
1202         lgi->lgi_buf.lb_buf = idarray;
1203         lgi->lgi_buf.lb_len = size;
1204         rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
1205         if (rc)
1206                 CDEBUG(D_INODE, "error writeing CATALOGS: rc = %d\n", rc);
1207 out_trans:
1208         dt_trans_stop(env, d, th);
1209 out:
1210         lu_object_put(env, &o->do_lu);
1211         RETURN(rc);
1212 }
1213 EXPORT_SYMBOL(llog_osd_put_cat_list);