Whamcloud - gitweb
4f0e45be2ce6195036198ad1563b52e7db8833d1
[fs/lustre-release.git] / lustre / obdclass / llog_lvfs.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
8  * This program is free software; you can redistribute it and/or modify
9  * it under the terms of the GNU General Public License version 2 only,
10  * as published by the Free Software Foundation.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * General Public License version 2 for more details (a copy is included
16  * in the LICENSE file that accompanied this code).
17  *
18  * You should have received a copy of the GNU General Public License
19  * version 2 along with this program; If not, see
20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
21  *
22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23  * CA 95054 USA or visit www.sun.com if you need additional information or
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
30  * Use is subject to license terms.
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_lvfs.c
37  *
38  * OST<->MDS recovery logging infrastructure.
39  * Invariants in implementation:
40  * - we do not share logs among different OST<->MDS connections, so that
41  *   if an OST or MDS fails it need only look at log(s) relevant to itself
42  *
43  * Author: Andreas Dilger <adilger@clusterfs.com>
44  */
45
46 #define DEBUG_SUBSYSTEM S_LOG
47
48 #ifndef EXPORT_SYMTAB
49 #define EXPORT_SYMTAB
50 #endif
51
52 #ifndef __KERNEL__
53 #include <liblustre.h>
54 #endif
55
56 #include <obd.h>
57 #include <obd_class.h>
58 #include <lustre_log.h>
59 #include <obd_ost.h>
60 #include <libcfs/list.h>
61 #include <lvfs.h>
62 #include <lustre_fsfilt.h>
63 #include <lustre_disk.h>
64 #include "llog_internal.h"
65
66 #if defined(__KERNEL__) && defined(LLOG_LVFS)
67
68 static int llog_lvfs_pad(struct obd_device *obd, struct l_file *file,
69                                 int len, int index)
70 {
71         struct llog_rec_hdr rec = { 0 };
72         struct llog_rec_tail tail;
73         int rc;
74         ENTRY;
75
76         LASSERT(len >= LLOG_MIN_REC_SIZE && (len & 0x7) == 0);
77
78         tail.lrt_len = rec.lrh_len = len;
79         tail.lrt_index = rec.lrh_index = index;
80         rec.lrh_type = LLOG_PAD_MAGIC;
81
82         rc = fsfilt_write_record(obd, file, &rec, sizeof(rec), &file->f_pos, 0);
83         if (rc) {
84                 CERROR("error writing padding record: rc %d\n", rc);
85                 goto out;
86         }
87
88         file->f_pos += len - sizeof(rec) - sizeof(tail);
89         rc = fsfilt_write_record(obd, file, &tail, sizeof(tail),&file->f_pos,0);
90         if (rc) {
91                 CERROR("error writing padding record: rc %d\n", rc);
92                 goto out;
93         }
94
95  out:
96         RETURN(rc);
97 }
98
99 static int llog_lvfs_write_blob(struct obd_device *obd, struct l_file *file,
100                                 struct llog_rec_hdr *rec, void *buf, loff_t off)
101 {
102         int rc;
103         struct llog_rec_tail end;
104         loff_t saved_off = file->f_pos;
105         int buflen = rec->lrh_len;
106
107         ENTRY;
108
109         file->f_pos = off;
110
111         if (buflen == 0)
112                 CWARN("0-length record\n");
113
114         if (!buf) {
115                 rc = fsfilt_write_record(obd, file, rec, buflen,&file->f_pos,0);
116                 if (rc) {
117                         CERROR("error writing log record: rc %d\n", rc);
118                         goto out;
119                 }
120                 GOTO(out, rc = 0);
121         }
122
123         /* the buf case */
124         rec->lrh_len = sizeof(*rec) + buflen + sizeof(end);
125         rc = fsfilt_write_record(obd, file, rec, sizeof(*rec), &file->f_pos, 0);
126         if (rc) {
127                 CERROR("error writing log hdr: rc %d\n", rc);
128                 goto out;
129         }
130
131         rc = fsfilt_write_record(obd, file, buf, buflen, &file->f_pos, 0);
132         if (rc) {
133                 CERROR("error writing log buffer: rc %d\n", rc);
134                 goto out;
135         }
136
137         end.lrt_len = rec->lrh_len;
138         end.lrt_index = rec->lrh_index;
139         rc = fsfilt_write_record(obd, file, &end, sizeof(end), &file->f_pos, 0);
140         if (rc) {
141                 CERROR("error writing log tail: rc %d\n", rc);
142                 goto out;
143         }
144
145         rc = 0;
146  out:
147         if (saved_off > file->f_pos)
148                 file->f_pos = saved_off;
149         LASSERT(rc <= 0);
150         RETURN(rc);
151 }
152
153 static int llog_lvfs_read_blob(struct obd_device *obd, struct l_file *file,
154                                 void *buf, int size, loff_t off)
155 {
156         loff_t offset = off;
157         int rc;
158         ENTRY;
159
160         rc = fsfilt_read_record(obd, file, buf, size, &offset);
161         if (rc) {
162                 CERROR("error reading log record: rc %d\n", rc);
163                 RETURN(rc);
164         }
165         RETURN(0);
166 }
167
168 static int llog_lvfs_read_header(struct llog_handle *handle)
169 {
170         struct obd_device *obd;
171         int rc;
172         ENTRY;
173
174         LASSERT(sizeof(*handle->lgh_hdr) == LLOG_CHUNK_SIZE);
175
176         obd = handle->lgh_ctxt->loc_exp->exp_obd;
177
178         if (i_size_read(handle->lgh_file->f_dentry->d_inode) == 0) {
179                 CDEBUG(D_HA, "not reading header from 0-byte log\n");
180                 RETURN(LLOG_EEMPTY);
181         }
182
183         rc = llog_lvfs_read_blob(obd, handle->lgh_file, handle->lgh_hdr,
184                                  LLOG_CHUNK_SIZE, 0);
185         if (rc) {
186                 CERROR("error reading log header from %.*s\n",
187                        handle->lgh_file->f_dentry->d_name.len,
188                        handle->lgh_file->f_dentry->d_name.name);
189         } else {
190                 struct llog_rec_hdr *llh_hdr = &handle->lgh_hdr->llh_hdr;
191
192                 if (LLOG_REC_HDR_NEEDS_SWABBING(llh_hdr))
193                         lustre_swab_llog_hdr(handle->lgh_hdr);
194
195                 if (llh_hdr->lrh_type != LLOG_HDR_MAGIC) {
196                         CERROR("bad log %.*s header magic: %#x (expected %#x)\n",
197                                handle->lgh_file->f_dentry->d_name.len,
198                                handle->lgh_file->f_dentry->d_name.name,
199                                llh_hdr->lrh_type, LLOG_HDR_MAGIC);
200                         rc = -EIO;
201                 } else if (llh_hdr->lrh_len != LLOG_CHUNK_SIZE) {
202                         CERROR("incorrectly sized log %.*s header: %#x "
203                                "(expected %#x)\n",
204                                handle->lgh_file->f_dentry->d_name.len,
205                                handle->lgh_file->f_dentry->d_name.name,
206                                llh_hdr->lrh_len, LLOG_CHUNK_SIZE);
207                         CERROR("you may need to re-run lconf --write_conf.\n");
208                         rc = -EIO;
209                 }
210         }
211
212         handle->lgh_last_idx = handle->lgh_hdr->llh_tail.lrt_index;
213         handle->lgh_file->f_pos = i_size_read(handle->lgh_file->f_dentry->d_inode);
214
215         RETURN(rc);
216 }
217
218 /* returns negative in on error; 0 if success && reccookie == 0; 1 otherwise */
219 /* appends if idx == -1, otherwise overwrites record idx. */
220 static int llog_lvfs_write_rec(struct llog_handle *loghandle,
221                                struct llog_rec_hdr *rec,
222                                struct llog_cookie *reccookie, int cookiecount,
223                                void *buf, int idx)
224 {
225         struct llog_log_hdr *llh;
226         int reclen = rec->lrh_len, index, rc;
227         struct llog_rec_tail *lrt;
228         struct obd_device *obd;
229         struct file *file;
230         size_t left;
231         ENTRY;
232
233         llh = loghandle->lgh_hdr;
234         file = loghandle->lgh_file;
235         obd = loghandle->lgh_ctxt->loc_exp->exp_obd;
236
237         /* record length should not bigger than LLOG_CHUNK_SIZE */
238         if (buf)
239                 rc = (reclen > LLOG_CHUNK_SIZE - sizeof(struct llog_rec_hdr) -
240                       sizeof(struct llog_rec_tail)) ? -E2BIG : 0;
241         else
242                 rc = (reclen > LLOG_CHUNK_SIZE) ? -E2BIG : 0;
243         if (rc)
244                 RETURN(rc);
245
246         if (buf)
247                 /* write_blob adds header and tail to lrh_len. */
248                 reclen = sizeof(*rec) + rec->lrh_len +
249                          sizeof(struct llog_rec_tail);
250
251         if (idx != -1) {
252                 loff_t saved_offset;
253
254                 /* no header: only allowed to insert record 1 */
255                 if (idx != 1 && !i_size_read(file->f_dentry->d_inode)) {
256                         CERROR("idx != -1 in empty log\n");
257                         LBUG();
258                 }
259
260                 if (idx && llh->llh_size && llh->llh_size != rec->lrh_len)
261                         RETURN(-EINVAL);
262
263                 if (!ext2_test_bit(idx, llh->llh_bitmap))
264                         CERROR("Modify unset record %u\n", idx);
265                 if (idx != rec->lrh_index)
266                         CERROR("Index mismatch %d %u\n", idx, rec->lrh_index);
267
268                 rc = llog_lvfs_write_blob(obd, file, &llh->llh_hdr, NULL, 0);
269                 /* we are done if we only write the header or on error */
270                 if (rc || idx == 0)
271                         RETURN(rc);
272
273                 /* Assumes constant lrh_len */
274                 saved_offset = sizeof(*llh) + (idx - 1) * reclen;
275
276                 if (buf) {
277                         struct llog_rec_hdr check;
278
279                         /* We assume that caller has set lgh_cur_* */
280                         saved_offset = loghandle->lgh_cur_offset;
281                         CDEBUG(D_OTHER,
282                                "modify record "LPX64": idx:%d/%u/%d, len:%u "
283                                "offset %llu\n",
284                                loghandle->lgh_id.lgl_oid, idx, rec->lrh_index,
285                                loghandle->lgh_cur_idx, rec->lrh_len,
286                                (long long)(saved_offset - sizeof(*llh)));
287                         if (rec->lrh_index != loghandle->lgh_cur_idx) {
288                                 CERROR("modify idx mismatch %u/%d\n",
289                                        idx, loghandle->lgh_cur_idx);
290                                 RETURN(-EFAULT);
291                         }
292 #if 1  /* FIXME remove this safety check at some point */
293                         /* Verify that the record we're modifying is the
294                            right one. */
295                         rc = llog_lvfs_read_blob(obd, file, &check,
296                                                  sizeof(check), saved_offset);
297                         if (check.lrh_index != idx || check.lrh_len != reclen) {
298                                 CERROR("Bad modify idx %u/%u size %u/%u (%d)\n",
299                                        idx, check.lrh_index, reclen,
300                                        check.lrh_len, rc);
301                                 RETURN(-EFAULT);
302                         }
303 #endif
304                 }
305
306                 rc = llog_lvfs_write_blob(obd, file, rec, buf, saved_offset);
307                 if (rc == 0 && reccookie) {
308                         reccookie->lgc_lgl = loghandle->lgh_id;
309                         reccookie->lgc_index = idx;
310                         rc = 1;
311                 }
312                 RETURN(rc);
313         }
314
315         /* Make sure that records don't cross a chunk boundary, so we can
316          * process them page-at-a-time if needed.  If it will cross a chunk
317          * boundary, write in a fake (but referenced) entry to pad the chunk.
318          *
319          * We know that llog_current_log() will return a loghandle that is
320          * big enough to hold reclen, so all we care about is padding here.
321          */
322         left = LLOG_CHUNK_SIZE - (file->f_pos & (LLOG_CHUNK_SIZE - 1));
323
324         /* NOTE: padding is a record, but no bit is set */
325         if (left != 0 && left != reclen &&
326             left < (reclen + LLOG_MIN_REC_SIZE)) {
327                  index = loghandle->lgh_last_idx + 1;
328                  rc = llog_lvfs_pad(obd, file, left, index);
329                  if (rc)
330                          RETURN(rc);
331                  loghandle->lgh_last_idx++; /*for pad rec*/
332          }
333          /* if it's the last idx in log file, then return -ENOSPC */
334          if (loghandle->lgh_last_idx >= LLOG_BITMAP_SIZE(llh) - 1)
335                  RETURN(-ENOSPC);
336         loghandle->lgh_last_idx++;
337         index = loghandle->lgh_last_idx;
338         LASSERT(index < LLOG_BITMAP_SIZE(llh));
339         rec->lrh_index = index;
340         if (buf == NULL) {
341                 lrt = (struct llog_rec_tail *)
342                         ((char *)rec + rec->lrh_len - sizeof(*lrt));
343                 lrt->lrt_len = rec->lrh_len;
344                 lrt->lrt_index = rec->lrh_index;
345         }
346         /*The caller should make sure only 1 process access the lgh_last_idx,
347          *Otherwise it might hit the assert.*/
348         LASSERT(index < LLOG_BITMAP_SIZE(llh));
349         if (ext2_set_bit(index, llh->llh_bitmap)) {
350                 CERROR("argh, index %u already set in log bitmap?\n", index);
351                 LBUG(); /* should never happen */
352         }
353         llh->llh_count++;
354         llh->llh_tail.lrt_index = index;
355
356         rc = llog_lvfs_write_blob(obd, file, &llh->llh_hdr, NULL, 0);
357         if (rc)
358                 RETURN(rc);
359
360         rc = llog_lvfs_write_blob(obd, file, rec, buf, file->f_pos);
361         if (rc)
362                 RETURN(rc);
363
364         CDEBUG(D_RPCTRACE, "added record "LPX64": idx: %u, %u \n",
365                loghandle->lgh_id.lgl_oid, index, rec->lrh_len);
366         if (rc == 0 && reccookie) {
367                 reccookie->lgc_lgl = loghandle->lgh_id;
368                 reccookie->lgc_index = index;
369                 if ((rec->lrh_type == MDS_UNLINK_REC) ||
370                                 (rec->lrh_type == MDS_SETATTR_REC))
371                         reccookie->lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
372                 else if (rec->lrh_type == OST_SZ_REC)
373                         reccookie->lgc_subsys = LLOG_SIZE_ORIG_CTXT;
374                 else if (rec->lrh_type == OST_RAID1_REC)
375                         reccookie->lgc_subsys = LLOG_RD1_ORIG_CTXT;
376                 else
377                         reccookie->lgc_subsys = -1;
378                 rc = 1;
379         }
380         if (rc == 0 && rec->lrh_type == LLOG_GEN_REC)
381                 rc = 1;
382
383         RETURN(rc);
384 }
385
386 /* We can skip reading at least as many log blocks as the number of
387 * minimum sized log records we are skipping.  If it turns out
388 * that we are not far enough along the log (because the
389 * actual records are larger than minimum size) we just skip
390 * some more records. */
391
392 static void llog_skip_over(__u64 *off, int curr, int goal)
393 {
394         if (goal <= curr)
395                 return;
396         *off = (*off + (goal-curr-1) * LLOG_MIN_REC_SIZE) &
397                 ~(LLOG_CHUNK_SIZE - 1);
398 }
399
400
401 /* sets:
402  *  - cur_offset to the furthest point read in the log file
403  *  - cur_idx to the log index preceeding cur_offset
404  * returns -EIO/-EINVAL on error
405  */
406 static int llog_lvfs_next_block(struct llog_handle *loghandle, int *cur_idx,
407                                 int next_idx, __u64 *cur_offset, void *buf,
408                                 int len)
409 {
410         int rc;
411         ENTRY;
412
413         if (len == 0 || len & (LLOG_CHUNK_SIZE - 1))
414                 RETURN(-EINVAL);
415
416         CDEBUG(D_OTHER, "looking for log index %u (cur idx %u off "LPU64")\n",
417                next_idx, *cur_idx, *cur_offset);
418
419         while (*cur_offset < i_size_read(loghandle->lgh_file->f_dentry->d_inode)) {
420                 struct llog_rec_hdr *rec;
421                 struct llog_rec_tail *tail;
422                 loff_t ppos;
423
424                 llog_skip_over(cur_offset, *cur_idx, next_idx);
425
426                 ppos = *cur_offset;
427                 rc = fsfilt_read_record(loghandle->lgh_ctxt->loc_exp->exp_obd,
428                                         loghandle->lgh_file, buf, len,
429                                         &ppos);
430                 if (rc) {
431                         CERROR("Cant read llog block at log id "LPU64
432                                "/%u offset "LPU64"\n",
433                                loghandle->lgh_id.lgl_oid,
434                                loghandle->lgh_id.lgl_ogen,
435                                *cur_offset);
436                         RETURN(rc);
437                 }
438
439                 /* put number of bytes read into rc to make code simpler */
440                 rc = ppos - *cur_offset;
441                 *cur_offset = ppos;
442
443                 if (rc < len) {
444                         /* signal the end of the valid buffer to llog_process */
445                         memset(buf + rc, 0, len - rc);
446                 }
447
448                 if (rc == 0) /* end of file, nothing to do */
449                         RETURN(0);
450
451                 if (rc < sizeof(*tail)) {
452                         CERROR("Invalid llog block at log id "LPU64"/%u offset "
453                                LPU64"\n", loghandle->lgh_id.lgl_oid,
454                                loghandle->lgh_id.lgl_ogen, *cur_offset);
455                         RETURN(-EINVAL);
456                 }
457
458                 rec = buf;
459                 tail = (struct llog_rec_tail *)((char *)buf + rc -
460                                                 sizeof(struct llog_rec_tail));
461
462                 if (LLOG_REC_HDR_NEEDS_SWABBING(rec)) {
463                         lustre_swab_llog_rec(rec, tail);
464                 }
465
466                 *cur_idx = tail->lrt_index;
467
468                 /* this shouldn't happen */
469                 if (tail->lrt_index == 0) {
470                         CERROR("Invalid llog tail at log id "LPU64"/%u offset "
471                                LPU64"\n", loghandle->lgh_id.lgl_oid,
472                                loghandle->lgh_id.lgl_ogen, *cur_offset);
473                         RETURN(-EINVAL);
474                 }
475                 if (tail->lrt_index < next_idx)
476                         continue;
477
478                 /* sanity check that the start of the new buffer is no farther
479                  * than the record that we wanted.  This shouldn't happen. */
480                 if (rec->lrh_index > next_idx) {
481                         CERROR("missed desired record? %u > %u\n",
482                                rec->lrh_index, next_idx);
483                         RETURN(-ENOENT);
484                 }
485                 RETURN(0);
486         }
487         RETURN(-EIO);
488 }
489
490 static int llog_lvfs_prev_block(struct llog_handle *loghandle,
491                                 int prev_idx, void *buf, int len)
492 {
493         __u64 cur_offset;
494         int rc;
495         ENTRY;
496
497         if (len == 0 || len & (LLOG_CHUNK_SIZE - 1))
498                 RETURN(-EINVAL);
499
500         CDEBUG(D_OTHER, "looking for log index %u\n", prev_idx);
501
502         cur_offset = LLOG_CHUNK_SIZE;
503         llog_skip_over(&cur_offset, 0, prev_idx);
504
505         while (cur_offset < i_size_read(loghandle->lgh_file->f_dentry->d_inode)) {
506                 struct llog_rec_hdr *rec;
507                 struct llog_rec_tail *tail;
508                 loff_t ppos;
509
510                 ppos = cur_offset;
511
512                 rc = fsfilt_read_record(loghandle->lgh_ctxt->loc_exp->exp_obd,
513                                         loghandle->lgh_file, buf, len,
514                                         &ppos);
515                 if (rc) {
516                         CERROR("Cant read llog block at log id "LPU64
517                                "/%u offset "LPU64"\n",
518                                loghandle->lgh_id.lgl_oid,
519                                loghandle->lgh_id.lgl_ogen,
520                                cur_offset);
521                         RETURN(rc);
522                 }
523
524                 /* put number of bytes read into rc to make code simpler */
525                 rc = ppos - cur_offset;
526                 cur_offset = ppos;
527
528                 if (rc == 0) /* end of file, nothing to do */
529                         RETURN(0);
530
531                 if (rc < sizeof(*tail)) {
532                         CERROR("Invalid llog block at log id "LPU64"/%u offset "
533                                LPU64"\n", loghandle->lgh_id.lgl_oid,
534                                loghandle->lgh_id.lgl_ogen, cur_offset);
535                         RETURN(-EINVAL);
536                 }
537
538                 tail = buf + rc - sizeof(struct llog_rec_tail);
539
540                 /* this shouldn't happen */
541                 if (tail->lrt_index == 0) {
542                         CERROR("Invalid llog tail at log id "LPU64"/%u offset "
543                                LPU64"\n", loghandle->lgh_id.lgl_oid,
544                                loghandle->lgh_id.lgl_ogen, cur_offset);
545                         RETURN(-EINVAL);
546                 }
547                 if (le32_to_cpu(tail->lrt_index) < prev_idx)
548                         continue;
549
550                 /* sanity check that the start of the new buffer is no farther
551                  * than the record that we wanted.  This shouldn't happen. */
552                 rec = buf;
553                 if (le32_to_cpu(rec->lrh_index) > prev_idx) {
554                         CERROR("missed desired record? %u > %u\n",
555                                le32_to_cpu(rec->lrh_index), prev_idx);
556                         RETURN(-ENOENT);
557                 }
558                 RETURN(0);
559         }
560         RETURN(-EIO);
561 }
562
563 static struct file *llog_filp_open(char *dir, char *name, int flags, int mode)
564 {
565         char *logname;
566         struct file *filp;
567         int len;
568
569         OBD_ALLOC(logname, PATH_MAX);
570         if (logname == NULL)
571                 return ERR_PTR(-ENOMEM);
572
573         len = snprintf(logname, PATH_MAX, "%s/%s", dir, name);
574         if (len >= PATH_MAX - 1) {
575                 filp = ERR_PTR(-ENAMETOOLONG);
576         } else {
577                 filp = l_filp_open(logname, flags, mode);
578                 if (IS_ERR(filp))
579                         CERROR("logfile creation %s: %ld\n", logname,
580                                PTR_ERR(filp));
581         }
582         OBD_FREE(logname, PATH_MAX);
583         return filp;
584 }
585
586 /* This is a callback from the llog_* functions.
587  * Assumes caller has already pushed us into the kernel context. */
588 static int llog_lvfs_create(struct llog_ctxt *ctxt, struct llog_handle **res,
589                             struct llog_logid *logid, char *name)
590 {
591         struct llog_handle *handle;
592         struct obd_device *obd;
593         struct l_dentry *dchild = NULL;
594         struct obdo *oa = NULL;
595         int rc = 0, cleanup_phase = 1;
596         int open_flags = O_RDWR | O_CREAT | O_LARGEFILE;
597         ENTRY;
598
599         handle = llog_alloc_handle();
600         if (handle == NULL)
601                 RETURN(-ENOMEM);
602         *res = handle;
603
604         LASSERT(ctxt);
605         LASSERT(ctxt->loc_exp);
606         obd = ctxt->loc_exp->exp_obd;
607
608         if (logid != NULL) {
609                 dchild = obd_lvfs_fid2dentry(ctxt->loc_exp, logid->lgl_oid,
610                                              logid->lgl_ogen, logid->lgl_ogr);
611
612                 if (IS_ERR(dchild)) {
613                         rc = PTR_ERR(dchild);
614                         CERROR("error looking up logfile "LPX64":0x%x: rc %d\n",
615                                logid->lgl_oid, logid->lgl_ogen, rc);
616                         GOTO(cleanup, rc);
617                 }
618
619                 cleanup_phase = 2;
620                 if (dchild->d_inode == NULL) {
621                         rc = -ENOENT;
622                         CERROR("nonexistent log file "LPX64":"LPX64": rc %d\n",
623                                logid->lgl_oid, logid->lgl_ogr, rc);
624                         GOTO(cleanup, rc);
625                 }
626
627                 handle->lgh_file = l_dentry_open(&obd->obd_lvfs_ctxt, dchild,
628                                                     O_RDWR | O_LARGEFILE);
629                 if (IS_ERR(handle->lgh_file)) {
630                         rc = PTR_ERR(handle->lgh_file);
631                         CERROR("error opening logfile "LPX64"0x%x: rc %d\n",
632                                logid->lgl_oid, logid->lgl_ogen, rc);
633                         GOTO(cleanup, rc);
634                 }
635
636                 /* assign the value of lgh_id for handle directly */
637                 handle->lgh_id = *logid;
638
639         } else if (name) {
640                 handle->lgh_file = llog_filp_open(MOUNT_CONFIGS_DIR,
641                                                   name, open_flags, 0644);
642                 if (IS_ERR(handle->lgh_file))
643                         GOTO(cleanup, rc = PTR_ERR(handle->lgh_file));
644
645                 handle->lgh_id.lgl_ogr = 1;
646                 handle->lgh_id.lgl_oid =
647                         handle->lgh_file->f_dentry->d_inode->i_ino;
648                 handle->lgh_id.lgl_ogen =
649                         handle->lgh_file->f_dentry->d_inode->i_generation;
650         } else {
651                 OBDO_ALLOC(oa);
652                 if (oa == NULL)
653                         GOTO(cleanup, rc = -ENOMEM);
654
655                 oa->o_gr = FILTER_GROUP_LLOG;
656                 oa->o_valid = OBD_MD_FLGENER | OBD_MD_FLGROUP;
657
658                 rc = obd_create(ctxt->loc_exp, oa, NULL, NULL);
659                 if (rc)
660                         GOTO(cleanup, rc);
661
662                 dchild = obd_lvfs_fid2dentry(ctxt->loc_exp, oa->o_id,
663                                              oa->o_generation, oa->o_gr);
664
665                 if (IS_ERR(dchild))
666                         GOTO(cleanup, rc = PTR_ERR(dchild));
667                 cleanup_phase = 2;
668                 handle->lgh_file = l_dentry_open(&obd->obd_lvfs_ctxt, dchild,
669                                                  open_flags);
670                 if (IS_ERR(handle->lgh_file))
671                         GOTO(cleanup, rc = PTR_ERR(handle->lgh_file));
672
673                 handle->lgh_id.lgl_ogr = oa->o_gr;
674                 handle->lgh_id.lgl_oid = oa->o_id;
675                 handle->lgh_id.lgl_ogen = oa->o_generation;
676         }
677
678         handle->lgh_ctxt = ctxt;
679  finish:
680         if (oa)
681                 OBDO_FREE(oa);
682         RETURN(rc);
683 cleanup:
684         switch (cleanup_phase) {
685         case 2:
686                 l_dput(dchild);
687         case 1:
688                 llog_free_handle(handle);
689         }
690         goto finish;
691 }
692
693 static int llog_lvfs_close(struct llog_handle *handle)
694 {
695         int rc;
696         ENTRY;
697
698         rc = filp_close(handle->lgh_file, 0);
699         if (rc)
700                 CERROR("error closing log: rc %d\n", rc);
701         RETURN(rc);
702 }
703
704 static int llog_lvfs_destroy(struct llog_handle *handle)
705 {
706         struct dentry *fdentry;
707         struct obdo *oa;
708         struct obd_device *obd = handle->lgh_ctxt->loc_exp->exp_obd;
709         char *dir;
710         int rc;
711         ENTRY;
712
713         dir = MOUNT_CONFIGS_DIR;
714
715         fdentry = handle->lgh_file->f_dentry;
716         if (strcmp(fdentry->d_parent->d_name.name, dir) == 0) {
717                 struct inode *inode = fdentry->d_parent->d_inode;
718                 struct lvfs_run_ctxt saved;
719                 struct vfsmount *mnt = mntget(handle->lgh_file->f_vfsmnt);
720
721                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
722                 dget(fdentry);
723                 rc = llog_lvfs_close(handle);
724
725                 if (rc == 0) {
726                         LOCK_INODE_MUTEX_PARENT(inode);
727                         rc = ll_vfs_unlink(inode, fdentry, mnt);
728                         UNLOCK_INODE_MUTEX(inode);
729                 }
730                 mntput(mnt);
731
732                 dput(fdentry);
733                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
734                 RETURN(rc);
735         }
736
737         OBDO_ALLOC(oa);
738         if (oa == NULL)
739                 RETURN(-ENOMEM);
740
741         oa->o_id = handle->lgh_id.lgl_oid;
742         oa->o_gr = handle->lgh_id.lgl_ogr;
743         oa->o_generation = handle->lgh_id.lgl_ogen;
744         oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP | OBD_MD_FLGENER;
745
746         rc = llog_lvfs_close(handle);
747         if (rc)
748                 GOTO(out, rc);
749
750         rc = obd_destroy(handle->lgh_ctxt->loc_exp, oa, NULL, NULL, NULL, NULL);
751  out:
752         OBDO_FREE(oa);
753         RETURN(rc);
754 }
755
756 /* reads the catalog list */
757 int llog_get_cat_list(struct obd_device *obd, struct obd_device *disk_obd,
758                       char *name, int idx, int count, struct llog_catid *idarray)
759 {
760         struct lvfs_run_ctxt saved;
761         struct l_file *file;
762         int rc, rc1 = 0;
763         int size = sizeof(*idarray) * count;
764         loff_t off = idx *  sizeof(*idarray);
765         ENTRY;
766
767         if (!count)
768                 RETURN(0);
769
770         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
771         file = filp_open(name, O_RDWR | O_CREAT | O_LARGEFILE, 0700);
772         if (!file || IS_ERR(file)) {
773                 rc = PTR_ERR(file);
774                 CERROR("OBD filter: cannot open/create %s: rc = %d\n",
775                        name, rc);
776                 GOTO(out, rc);
777         }
778
779         if (!S_ISREG(file->f_dentry->d_inode->i_mode)) {
780                 CERROR("%s is not a regular file!: mode = %o\n", name,
781                        file->f_dentry->d_inode->i_mode);
782                 GOTO(out, rc = -ENOENT);
783         }
784
785         CDEBUG(D_CONFIG, "cat list: disk size=%d, read=%d\n",
786                (int)i_size_read(file->f_dentry->d_inode), size);
787
788         /* read for new ost index or for empty file */
789         memset(idarray, 0, size);
790         if (i_size_read(file->f_dentry->d_inode) < off)
791                 GOTO(out, rc = 0);
792
793         rc = fsfilt_read_record(disk_obd, file, idarray, size, &off);
794         if (rc) {
795                 CERROR("OBD filter: error reading %s: rc %d\n", name, rc);
796                 GOTO(out, rc);
797         }
798
799         EXIT;
800  out:
801         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
802         if (file && !IS_ERR(file))
803                 rc1 = filp_close(file, 0);
804         if (rc == 0)
805                 rc = rc1;
806         return rc;
807 }
808 EXPORT_SYMBOL(llog_get_cat_list);
809
810 /* writes the cat list */
811 int llog_put_cat_list(struct obd_device *obd, struct obd_device *disk_obd,
812                       char *name, int idx, int count, struct llog_catid *idarray)
813 {
814         struct lvfs_run_ctxt saved;
815         struct l_file *file;
816         int rc, rc1 = 0;
817         int size = sizeof(*idarray) * count;
818         loff_t off = idx * sizeof(*idarray);
819
820         if (!count)
821                 GOTO(out1, rc = 0);
822
823         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
824         file = filp_open(name, O_RDWR | O_CREAT | O_LARGEFILE, 0700);
825         if (!file || IS_ERR(file)) {
826                 rc = PTR_ERR(file);
827                 CERROR("OBD filter: cannot open/create %s: rc = %d\n",
828                        name, rc);
829                 GOTO(out, rc);
830         }
831
832         if (!S_ISREG(file->f_dentry->d_inode->i_mode)) {
833                 CERROR("%s is not a regular file!: mode = %o\n", name,
834                        file->f_dentry->d_inode->i_mode);
835                 GOTO(out, rc = -ENOENT);
836         }
837
838         rc = fsfilt_write_record(disk_obd, file, idarray, size, &off, 1);
839         if (rc) {
840                 CDEBUG(D_INODE,"OBD filter: error writeing %s: rc %d\n",
841                        name, rc);
842                 GOTO(out, rc);
843         }
844
845 out:
846         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
847         if (file && !IS_ERR(file))
848                 rc1 = filp_close(file, 0);
849
850         if (rc == 0)
851                 rc = rc1;
852 out1:
853         RETURN(rc);
854 }
855 EXPORT_SYMBOL(llog_put_cat_list);
856
857 struct llog_operations llog_lvfs_ops = {
858         lop_write_rec:   llog_lvfs_write_rec,
859         lop_next_block:  llog_lvfs_next_block,
860         lop_prev_block:  llog_lvfs_prev_block,
861         lop_read_header: llog_lvfs_read_header,
862         lop_create:      llog_lvfs_create,
863         lop_destroy:     llog_lvfs_destroy,
864         lop_close:       llog_lvfs_close,
865         //        lop_cancel: llog_lvfs_cancel,
866 };
867
868 EXPORT_SYMBOL(llog_lvfs_ops);
869
870 #else /* !__KERNEL__ */
871
872 static int llog_lvfs_read_header(struct llog_handle *handle)
873 {
874         LBUG();
875         return 0;
876 }
877
878 static int llog_lvfs_write_rec(struct llog_handle *loghandle,
879                                struct llog_rec_hdr *rec,
880                                struct llog_cookie *reccookie, int cookiecount,
881                                void *buf, int idx)
882 {
883         LBUG();
884         return 0;
885 }
886
887 static int llog_lvfs_next_block(struct llog_handle *loghandle, int *cur_idx,
888                                 int next_idx, __u64 *cur_offset, void *buf,
889                                 int len)
890 {
891         LBUG();
892         return 0;
893 }
894
895 static int llog_lvfs_prev_block(struct llog_handle *loghandle,
896                                 int prev_idx, void *buf, int len)
897 {
898         LBUG();
899         return 0;
900 }
901
902 static int llog_lvfs_create(struct llog_ctxt *ctxt, struct llog_handle **res,
903                             struct llog_logid *logid, char *name)
904 {
905         LBUG();
906         return 0;
907 }
908
909 static int llog_lvfs_close(struct llog_handle *handle)
910 {
911         LBUG();
912         return 0;
913 }
914
915 static int llog_lvfs_destroy(struct llog_handle *handle)
916 {
917         LBUG();
918         return 0;
919 }
920
921 int llog_get_cat_list(struct obd_device *obd, struct obd_device *disk_obd,
922                       char *name, int idx, int count, struct llog_catid *idarray)
923 {
924         LBUG();
925         return 0;
926 }
927
928 int llog_put_cat_list(struct obd_device *obd, struct obd_device *disk_obd,
929                       char *name, int idx, int count, struct llog_catid *idarray)
930 {
931         LBUG();
932         return 0;
933 }
934
935 struct llog_operations llog_lvfs_ops = {
936         lop_write_rec:   llog_lvfs_write_rec,
937         lop_next_block:  llog_lvfs_next_block,
938         lop_prev_block:  llog_lvfs_prev_block,
939         lop_read_header: llog_lvfs_read_header,
940         lop_create:      llog_lvfs_create,
941         lop_destroy:     llog_lvfs_destroy,
942         lop_close:       llog_lvfs_close,
943 //        lop_cancel:      llog_lvfs_cancel,
944 };
945 #endif